Statistics
| Branch: | Tag: | Revision:

root / lib / cmdlib.py @ d04aaa2f

History | View | Annotate | Download (243.4 kB)

1
#
2
#
3

    
4
# Copyright (C) 2006, 2007, 2008 Google Inc.
5
#
6
# This program is free software; you can redistribute it and/or modify
7
# it under the terms of the GNU General Public License as published by
8
# the Free Software Foundation; either version 2 of the License, or
9
# (at your option) any later version.
10
#
11
# This program is distributed in the hope that it will be useful, but
12
# WITHOUT ANY WARRANTY; without even the implied warranty of
13
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
14
# General Public License for more details.
15
#
16
# You should have received a copy of the GNU General Public License
17
# along with this program; if not, write to the Free Software
18
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
19
# 02110-1301, USA.
20

    
21

    
22
"""Module implementing the master-side code."""
23

    
24
# pylint: disable-msg=W0613,W0201
25

    
26
import os
27
import os.path
28
import time
29
import tempfile
30
import re
31
import platform
32
import logging
33
import copy
34
import random
35

    
36
from ganeti import ssh
37
from ganeti import utils
38
from ganeti import errors
39
from ganeti import hypervisor
40
from ganeti import locking
41
from ganeti import constants
42
from ganeti import objects
43
from ganeti import opcodes
44
from ganeti import serializer
45
from ganeti import ssconf
46

    
47

    
48
class LogicalUnit(object):
49
  """Logical Unit base class.
50

51
  Subclasses must follow these rules:
52
    - implement ExpandNames
53
    - implement CheckPrereq
54
    - implement Exec
55
    - implement BuildHooksEnv
56
    - redefine HPATH and HTYPE
57
    - optionally redefine their run requirements:
58
        REQ_BGL: the LU needs to hold the Big Ganeti Lock exclusively
59

60
  Note that all commands require root permissions.
61

62
  """
63
  HPATH = None
64
  HTYPE = None
65
  _OP_REQP = []
66
  REQ_BGL = True
67

    
68
  def __init__(self, processor, op, context, rpc):
69
    """Constructor for LogicalUnit.
70

71
    This needs to be overriden in derived classes in order to check op
72
    validity.
73

74
    """
75
    self.proc = processor
76
    self.op = op
77
    self.cfg = context.cfg
78
    self.context = context
79
    self.rpc = rpc
80
    # Dicts used to declare locking needs to mcpu
81
    self.needed_locks = None
82
    self.acquired_locks = {}
83
    self.share_locks = dict(((i, 0) for i in locking.LEVELS))
84
    self.add_locks = {}
85
    self.remove_locks = {}
86
    # Used to force good behavior when calling helper functions
87
    self.recalculate_locks = {}
88
    self.__ssh = None
89
    # logging
90
    self.LogWarning = processor.LogWarning
91
    self.LogInfo = processor.LogInfo
92

    
93
    for attr_name in self._OP_REQP:
94
      attr_val = getattr(op, attr_name, None)
95
      if attr_val is None:
96
        raise errors.OpPrereqError("Required parameter '%s' missing" %
97
                                   attr_name)
98
    self.CheckArguments()
99

    
100
  def __GetSSH(self):
101
    """Returns the SshRunner object
102

103
    """
104
    if not self.__ssh:
105
      self.__ssh = ssh.SshRunner(self.cfg.GetClusterName())
106
    return self.__ssh
107

    
108
  ssh = property(fget=__GetSSH)
109

    
110
  def CheckArguments(self):
111
    """Check syntactic validity for the opcode arguments.
112

113
    This method is for doing a simple syntactic check and ensure
114
    validity of opcode parameters, without any cluster-related
115
    checks. While the same can be accomplished in ExpandNames and/or
116
    CheckPrereq, doing these separate is better because:
117

118
      - ExpandNames is left as as purely a lock-related function
119
      - CheckPrereq is run after we have aquired locks (and possible
120
        waited for them)
121

122
    The function is allowed to change the self.op attribute so that
123
    later methods can no longer worry about missing parameters.
124

125
    """
126
    pass
127

    
128
  def ExpandNames(self):
129
    """Expand names for this LU.
130

131
    This method is called before starting to execute the opcode, and it should
132
    update all the parameters of the opcode to their canonical form (e.g. a
133
    short node name must be fully expanded after this method has successfully
134
    completed). This way locking, hooks, logging, ecc. can work correctly.
135

136
    LUs which implement this method must also populate the self.needed_locks
137
    member, as a dict with lock levels as keys, and a list of needed lock names
138
    as values. Rules:
139

140
      - use an empty dict if you don't need any lock
141
      - if you don't need any lock at a particular level omit that level
142
      - don't put anything for the BGL level
143
      - if you want all locks at a level use locking.ALL_SET as a value
144

145
    If you need to share locks (rather than acquire them exclusively) at one
146
    level you can modify self.share_locks, setting a true value (usually 1) for
147
    that level. By default locks are not shared.
148

149
    Examples::
150

151
      # Acquire all nodes and one instance
152
      self.needed_locks = {
153
        locking.LEVEL_NODE: locking.ALL_SET,
154
        locking.LEVEL_INSTANCE: ['instance1.example.tld'],
155
      }
156
      # Acquire just two nodes
157
      self.needed_locks = {
158
        locking.LEVEL_NODE: ['node1.example.tld', 'node2.example.tld'],
159
      }
160
      # Acquire no locks
161
      self.needed_locks = {} # No, you can't leave it to the default value None
162

163
    """
164
    # The implementation of this method is mandatory only if the new LU is
165
    # concurrent, so that old LUs don't need to be changed all at the same
166
    # time.
167
    if self.REQ_BGL:
168
      self.needed_locks = {} # Exclusive LUs don't need locks.
169
    else:
170
      raise NotImplementedError
171

    
172
  def DeclareLocks(self, level):
173
    """Declare LU locking needs for a level
174

175
    While most LUs can just declare their locking needs at ExpandNames time,
176
    sometimes there's the need to calculate some locks after having acquired
177
    the ones before. This function is called just before acquiring locks at a
178
    particular level, but after acquiring the ones at lower levels, and permits
179
    such calculations. It can be used to modify self.needed_locks, and by
180
    default it does nothing.
181

182
    This function is only called if you have something already set in
183
    self.needed_locks for the level.
184

185
    @param level: Locking level which is going to be locked
186
    @type level: member of ganeti.locking.LEVELS
187

188
    """
189

    
190
  def CheckPrereq(self):
191
    """Check prerequisites for this LU.
192

193
    This method should check that the prerequisites for the execution
194
    of this LU are fulfilled. It can do internode communication, but
195
    it should be idempotent - no cluster or system changes are
196
    allowed.
197

198
    The method should raise errors.OpPrereqError in case something is
199
    not fulfilled. Its return value is ignored.
200

201
    This method should also update all the parameters of the opcode to
202
    their canonical form if it hasn't been done by ExpandNames before.
203

204
    """
205
    raise NotImplementedError
206

    
207
  def Exec(self, feedback_fn):
208
    """Execute the LU.
209

210
    This method should implement the actual work. It should raise
211
    errors.OpExecError for failures that are somewhat dealt with in
212
    code, or expected.
213

214
    """
215
    raise NotImplementedError
216

    
217
  def BuildHooksEnv(self):
218
    """Build hooks environment for this LU.
219

220
    This method should return a three-node tuple consisting of: a dict
221
    containing the environment that will be used for running the
222
    specific hook for this LU, a list of node names on which the hook
223
    should run before the execution, and a list of node names on which
224
    the hook should run after the execution.
225

226
    The keys of the dict must not have 'GANETI_' prefixed as this will
227
    be handled in the hooks runner. Also note additional keys will be
228
    added by the hooks runner. If the LU doesn't define any
229
    environment, an empty dict (and not None) should be returned.
230

231
    No nodes should be returned as an empty list (and not None).
232

233
    Note that if the HPATH for a LU class is None, this function will
234
    not be called.
235

236
    """
237
    raise NotImplementedError
238

    
239
  def HooksCallBack(self, phase, hook_results, feedback_fn, lu_result):
240
    """Notify the LU about the results of its hooks.
241

242
    This method is called every time a hooks phase is executed, and notifies
243
    the Logical Unit about the hooks' result. The LU can then use it to alter
244
    its result based on the hooks.  By default the method does nothing and the
245
    previous result is passed back unchanged but any LU can define it if it
246
    wants to use the local cluster hook-scripts somehow.
247

248
    @param phase: one of L{constants.HOOKS_PHASE_POST} or
249
        L{constants.HOOKS_PHASE_PRE}; it denotes the hooks phase
250
    @param hook_results: the results of the multi-node hooks rpc call
251
    @param feedback_fn: function used send feedback back to the caller
252
    @param lu_result: the previous Exec result this LU had, or None
253
        in the PRE phase
254
    @return: the new Exec result, based on the previous result
255
        and hook results
256

257
    """
258
    return lu_result
259

    
260
  def _ExpandAndLockInstance(self):
261
    """Helper function to expand and lock an instance.
262

263
    Many LUs that work on an instance take its name in self.op.instance_name
264
    and need to expand it and then declare the expanded name for locking. This
265
    function does it, and then updates self.op.instance_name to the expanded
266
    name. It also initializes needed_locks as a dict, if this hasn't been done
267
    before.
268

269
    """
270
    if self.needed_locks is None:
271
      self.needed_locks = {}
272
    else:
273
      assert locking.LEVEL_INSTANCE not in self.needed_locks, \
274
        "_ExpandAndLockInstance called with instance-level locks set"
275
    expanded_name = self.cfg.ExpandInstanceName(self.op.instance_name)
276
    if expanded_name is None:
277
      raise errors.OpPrereqError("Instance '%s' not known" %
278
                                  self.op.instance_name)
279
    self.needed_locks[locking.LEVEL_INSTANCE] = expanded_name
280
    self.op.instance_name = expanded_name
281

    
282
  def _LockInstancesNodes(self, primary_only=False):
283
    """Helper function to declare instances' nodes for locking.
284

285
    This function should be called after locking one or more instances to lock
286
    their nodes. Its effect is populating self.needed_locks[locking.LEVEL_NODE]
287
    with all primary or secondary nodes for instances already locked and
288
    present in self.needed_locks[locking.LEVEL_INSTANCE].
289

290
    It should be called from DeclareLocks, and for safety only works if
291
    self.recalculate_locks[locking.LEVEL_NODE] is set.
292

293
    In the future it may grow parameters to just lock some instance's nodes, or
294
    to just lock primaries or secondary nodes, if needed.
295

296
    If should be called in DeclareLocks in a way similar to::
297

298
      if level == locking.LEVEL_NODE:
299
        self._LockInstancesNodes()
300

301
    @type primary_only: boolean
302
    @param primary_only: only lock primary nodes of locked instances
303

304
    """
305
    assert locking.LEVEL_NODE in self.recalculate_locks, \
306
      "_LockInstancesNodes helper function called with no nodes to recalculate"
307

    
308
    # TODO: check if we're really been called with the instance locks held
309

    
310
    # For now we'll replace self.needed_locks[locking.LEVEL_NODE], but in the
311
    # future we might want to have different behaviors depending on the value
312
    # of self.recalculate_locks[locking.LEVEL_NODE]
313
    wanted_nodes = []
314
    for instance_name in self.acquired_locks[locking.LEVEL_INSTANCE]:
315
      instance = self.context.cfg.GetInstanceInfo(instance_name)
316
      wanted_nodes.append(instance.primary_node)
317
      if not primary_only:
318
        wanted_nodes.extend(instance.secondary_nodes)
319

    
320
    if self.recalculate_locks[locking.LEVEL_NODE] == constants.LOCKS_REPLACE:
321
      self.needed_locks[locking.LEVEL_NODE] = wanted_nodes
322
    elif self.recalculate_locks[locking.LEVEL_NODE] == constants.LOCKS_APPEND:
323
      self.needed_locks[locking.LEVEL_NODE].extend(wanted_nodes)
324

    
325
    del self.recalculate_locks[locking.LEVEL_NODE]
326

    
327

    
328
class NoHooksLU(LogicalUnit):
329
  """Simple LU which runs no hooks.
330

331
  This LU is intended as a parent for other LogicalUnits which will
332
  run no hooks, in order to reduce duplicate code.
333

334
  """
335
  HPATH = None
336
  HTYPE = None
337

    
338

    
339
def _GetWantedNodes(lu, nodes):
340
  """Returns list of checked and expanded node names.
341

342
  @type lu: L{LogicalUnit}
343
  @param lu: the logical unit on whose behalf we execute
344
  @type nodes: list
345
  @param nodes: list of node names or None for all nodes
346
  @rtype: list
347
  @return: the list of nodes, sorted
348
  @raise errors.OpProgrammerError: if the nodes parameter is wrong type
349

350
  """
351
  if not isinstance(nodes, list):
352
    raise errors.OpPrereqError("Invalid argument type 'nodes'")
353

    
354
  if not nodes:
355
    raise errors.ProgrammerError("_GetWantedNodes should only be called with a"
356
      " non-empty list of nodes whose name is to be expanded.")
357

    
358
  wanted = []
359
  for name in nodes:
360
    node = lu.cfg.ExpandNodeName(name)
361
    if node is None:
362
      raise errors.OpPrereqError("No such node name '%s'" % name)
363
    wanted.append(node)
364

    
365
  return utils.NiceSort(wanted)
366

    
367

    
368
def _GetWantedInstances(lu, instances):
369
  """Returns list of checked and expanded instance names.
370

371
  @type lu: L{LogicalUnit}
372
  @param lu: the logical unit on whose behalf we execute
373
  @type instances: list
374
  @param instances: list of instance names or None for all instances
375
  @rtype: list
376
  @return: the list of instances, sorted
377
  @raise errors.OpPrereqError: if the instances parameter is wrong type
378
  @raise errors.OpPrereqError: if any of the passed instances is not found
379

380
  """
381
  if not isinstance(instances, list):
382
    raise errors.OpPrereqError("Invalid argument type 'instances'")
383

    
384
  if instances:
385
    wanted = []
386

    
387
    for name in instances:
388
      instance = lu.cfg.ExpandInstanceName(name)
389
      if instance is None:
390
        raise errors.OpPrereqError("No such instance name '%s'" % name)
391
      wanted.append(instance)
392

    
393
  else:
394
    wanted = utils.NiceSort(lu.cfg.GetInstanceList())
395
  return wanted
396

    
397

    
398
def _CheckOutputFields(static, dynamic, selected):
399
  """Checks whether all selected fields are valid.
400

401
  @type static: L{utils.FieldSet}
402
  @param static: static fields set
403
  @type dynamic: L{utils.FieldSet}
404
  @param dynamic: dynamic fields set
405

406
  """
407
  f = utils.FieldSet()
408
  f.Extend(static)
409
  f.Extend(dynamic)
410

    
411
  delta = f.NonMatching(selected)
412
  if delta:
413
    raise errors.OpPrereqError("Unknown output fields selected: %s"
414
                               % ",".join(delta))
415

    
416

    
417
def _CheckBooleanOpField(op, name):
418
  """Validates boolean opcode parameters.
419

420
  This will ensure that an opcode parameter is either a boolean value,
421
  or None (but that it always exists).
422

423
  """
424
  val = getattr(op, name, None)
425
  if not (val is None or isinstance(val, bool)):
426
    raise errors.OpPrereqError("Invalid boolean parameter '%s' (%s)" %
427
                               (name, str(val)))
428
  setattr(op, name, val)
429

    
430

    
431
def _CheckNodeOnline(lu, node):
432
  """Ensure that a given node is online.
433

434
  @param lu: the LU on behalf of which we make the check
435
  @param node: the node to check
436
  @raise errors.OpPrereqError: if the node is offline
437

438
  """
439
  if lu.cfg.GetNodeInfo(node).offline:
440
    raise errors.OpPrereqError("Can't use offline node %s" % node)
441

    
442

    
443
def _CheckNodeNotDrained(lu, node):
444
  """Ensure that a given node is not drained.
445

446
  @param lu: the LU on behalf of which we make the check
447
  @param node: the node to check
448
  @raise errors.OpPrereqError: if the node is drained
449

450
  """
451
  if lu.cfg.GetNodeInfo(node).drained:
452
    raise errors.OpPrereqError("Can't use drained node %s" % node)
453

    
454

    
455
def _BuildInstanceHookEnv(name, primary_node, secondary_nodes, os_type, status,
456
                          memory, vcpus, nics, disk_template, disks):
457
  """Builds instance related env variables for hooks
458

459
  This builds the hook environment from individual variables.
460

461
  @type name: string
462
  @param name: the name of the instance
463
  @type primary_node: string
464
  @param primary_node: the name of the instance's primary node
465
  @type secondary_nodes: list
466
  @param secondary_nodes: list of secondary nodes as strings
467
  @type os_type: string
468
  @param os_type: the name of the instance's OS
469
  @type status: boolean
470
  @param status: the should_run status of the instance
471
  @type memory: string
472
  @param memory: the memory size of the instance
473
  @type vcpus: string
474
  @param vcpus: the count of VCPUs the instance has
475
  @type nics: list
476
  @param nics: list of tuples (ip, bridge, mac) representing
477
      the NICs the instance  has
478
  @type disk_template: string
479
  @param disk_template: the distk template of the instance
480
  @type disks: list
481
  @param disks: the list of (size, mode) pairs
482
  @rtype: dict
483
  @return: the hook environment for this instance
484

485
  """
486
  if status:
487
    str_status = "up"
488
  else:
489
    str_status = "down"
490
  env = {
491
    "OP_TARGET": name,
492
    "INSTANCE_NAME": name,
493
    "INSTANCE_PRIMARY": primary_node,
494
    "INSTANCE_SECONDARIES": " ".join(secondary_nodes),
495
    "INSTANCE_OS_TYPE": os_type,
496
    "INSTANCE_STATUS": str_status,
497
    "INSTANCE_MEMORY": memory,
498
    "INSTANCE_VCPUS": vcpus,
499
    "INSTANCE_DISK_TEMPLATE": disk_template,
500
  }
501

    
502
  if nics:
503
    nic_count = len(nics)
504
    for idx, (ip, bridge, mac) in enumerate(nics):
505
      if ip is None:
506
        ip = ""
507
      env["INSTANCE_NIC%d_IP" % idx] = ip
508
      env["INSTANCE_NIC%d_BRIDGE" % idx] = bridge
509
      env["INSTANCE_NIC%d_MAC" % idx] = mac
510
  else:
511
    nic_count = 0
512

    
513
  env["INSTANCE_NIC_COUNT"] = nic_count
514

    
515
  if disks:
516
    disk_count = len(disks)
517
    for idx, (size, mode) in enumerate(disks):
518
      env["INSTANCE_DISK%d_SIZE" % idx] = size
519
      env["INSTANCE_DISK%d_MODE" % idx] = mode
520
  else:
521
    disk_count = 0
522

    
523
  env["INSTANCE_DISK_COUNT"] = disk_count
524

    
525
  return env
526

    
527

    
528
def _BuildInstanceHookEnvByObject(lu, instance, override=None):
529
  """Builds instance related env variables for hooks from an object.
530

531
  @type lu: L{LogicalUnit}
532
  @param lu: the logical unit on whose behalf we execute
533
  @type instance: L{objects.Instance}
534
  @param instance: the instance for which we should build the
535
      environment
536
  @type override: dict
537
  @param override: dictionary with key/values that will override
538
      our values
539
  @rtype: dict
540
  @return: the hook environment dictionary
541

542
  """
543
  bep = lu.cfg.GetClusterInfo().FillBE(instance)
544
  args = {
545
    'name': instance.name,
546
    'primary_node': instance.primary_node,
547
    'secondary_nodes': instance.secondary_nodes,
548
    'os_type': instance.os,
549
    'status': instance.admin_up,
550
    'memory': bep[constants.BE_MEMORY],
551
    'vcpus': bep[constants.BE_VCPUS],
552
    'nics': [(nic.ip, nic.bridge, nic.mac) for nic in instance.nics],
553
    'disk_template': instance.disk_template,
554
    'disks': [(disk.size, disk.mode) for disk in instance.disks],
555
  }
556
  if override:
557
    args.update(override)
558
  return _BuildInstanceHookEnv(**args)
559

    
560

    
561
def _AdjustCandidatePool(lu):
562
  """Adjust the candidate pool after node operations.
563

564
  """
565
  mod_list = lu.cfg.MaintainCandidatePool()
566
  if mod_list:
567
    lu.LogInfo("Promoted nodes to master candidate role: %s",
568
               ", ".join(node.name for node in mod_list))
569
    for name in mod_list:
570
      lu.context.ReaddNode(name)
571
  mc_now, mc_max = lu.cfg.GetMasterCandidateStats()
572
  if mc_now > mc_max:
573
    lu.LogInfo("Note: more nodes are candidates (%d) than desired (%d)" %
574
               (mc_now, mc_max))
575

    
576

    
577
def _CheckInstanceBridgesExist(lu, instance):
578
  """Check that the brigdes needed by an instance exist.
579

580
  """
581
  # check bridges existance
582
  brlist = [nic.bridge for nic in instance.nics]
583
  result = lu.rpc.call_bridges_exist(instance.primary_node, brlist)
584
  result.Raise()
585
  if not result.data:
586
    raise errors.OpPrereqError("One or more target bridges %s does not"
587
                               " exist on destination node '%s'" %
588
                               (brlist, instance.primary_node))
589

    
590

    
591
class LUDestroyCluster(NoHooksLU):
592
  """Logical unit for destroying the cluster.
593

594
  """
595
  _OP_REQP = []
596

    
597
  def CheckPrereq(self):
598
    """Check prerequisites.
599

600
    This checks whether the cluster is empty.
601

602
    Any errors are signalled by raising errors.OpPrereqError.
603

604
    """
605
    master = self.cfg.GetMasterNode()
606

    
607
    nodelist = self.cfg.GetNodeList()
608
    if len(nodelist) != 1 or nodelist[0] != master:
609
      raise errors.OpPrereqError("There are still %d node(s) in"
610
                                 " this cluster." % (len(nodelist) - 1))
611
    instancelist = self.cfg.GetInstanceList()
612
    if instancelist:
613
      raise errors.OpPrereqError("There are still %d instance(s) in"
614
                                 " this cluster." % len(instancelist))
615

    
616
  def Exec(self, feedback_fn):
617
    """Destroys the cluster.
618

619
    """
620
    master = self.cfg.GetMasterNode()
621
    result = self.rpc.call_node_stop_master(master, False)
622
    result.Raise()
623
    if not result.data:
624
      raise errors.OpExecError("Could not disable the master role")
625
    priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
626
    utils.CreateBackup(priv_key)
627
    utils.CreateBackup(pub_key)
628
    return master
629

    
630

    
631
class LUVerifyCluster(LogicalUnit):
632
  """Verifies the cluster status.
633

634
  """
635
  HPATH = "cluster-verify"
636
  HTYPE = constants.HTYPE_CLUSTER
637
  _OP_REQP = ["skip_checks"]
638
  REQ_BGL = False
639

    
640
  def ExpandNames(self):
641
    self.needed_locks = {
642
      locking.LEVEL_NODE: locking.ALL_SET,
643
      locking.LEVEL_INSTANCE: locking.ALL_SET,
644
    }
645
    self.share_locks = dict(((i, 1) for i in locking.LEVELS))
646

    
647
  def _VerifyNode(self, nodeinfo, file_list, local_cksum,
648
                  node_result, feedback_fn, master_files,
649
                  drbd_map, vg_name):
650
    """Run multiple tests against a node.
651

652
    Test list:
653

654
      - compares ganeti version
655
      - checks vg existance and size > 20G
656
      - checks config file checksum
657
      - checks ssh to other nodes
658

659
    @type nodeinfo: L{objects.Node}
660
    @param nodeinfo: the node to check
661
    @param file_list: required list of files
662
    @param local_cksum: dictionary of local files and their checksums
663
    @param node_result: the results from the node
664
    @param feedback_fn: function used to accumulate results
665
    @param master_files: list of files that only masters should have
666
    @param drbd_map: the useddrbd minors for this node, in
667
        form of minor: (instance, must_exist) which correspond to instances
668
        and their running status
669
    @param vg_name: Ganeti Volume Group (result of self.cfg.GetVGName())
670

671
    """
672
    node = nodeinfo.name
673

    
674
    # main result, node_result should be a non-empty dict
675
    if not node_result or not isinstance(node_result, dict):
676
      feedback_fn("  - ERROR: unable to verify node %s." % (node,))
677
      return True
678

    
679
    # compares ganeti version
680
    local_version = constants.PROTOCOL_VERSION
681
    remote_version = node_result.get('version', None)
682
    if not (remote_version and isinstance(remote_version, (list, tuple)) and
683
            len(remote_version) == 2):
684
      feedback_fn("  - ERROR: connection to %s failed" % (node))
685
      return True
686

    
687
    if local_version != remote_version[0]:
688
      feedback_fn("  - ERROR: incompatible protocol versions: master %s,"
689
                  " node %s %s" % (local_version, node, remote_version[0]))
690
      return True
691

    
692
    # node seems compatible, we can actually try to look into its results
693

    
694
    bad = False
695

    
696
    # full package version
697
    if constants.RELEASE_VERSION != remote_version[1]:
698
      feedback_fn("  - WARNING: software version mismatch: master %s,"
699
                  " node %s %s" %
700
                  (constants.RELEASE_VERSION, node, remote_version[1]))
701

    
702
    # checks vg existence and size > 20G
703
    if vg_name is not None:
704
      vglist = node_result.get(constants.NV_VGLIST, None)
705
      if not vglist:
706
        feedback_fn("  - ERROR: unable to check volume groups on node %s." %
707
                        (node,))
708
        bad = True
709
      else:
710
        vgstatus = utils.CheckVolumeGroupSize(vglist, vg_name,
711
                                              constants.MIN_VG_SIZE)
712
        if vgstatus:
713
          feedback_fn("  - ERROR: %s on node %s" % (vgstatus, node))
714
          bad = True
715

    
716
    # checks config file checksum
717

    
718
    remote_cksum = node_result.get(constants.NV_FILELIST, None)
719
    if not isinstance(remote_cksum, dict):
720
      bad = True
721
      feedback_fn("  - ERROR: node hasn't returned file checksum data")
722
    else:
723
      for file_name in file_list:
724
        node_is_mc = nodeinfo.master_candidate
725
        must_have_file = file_name not in master_files
726
        if file_name not in remote_cksum:
727
          if node_is_mc or must_have_file:
728
            bad = True
729
            feedback_fn("  - ERROR: file '%s' missing" % file_name)
730
        elif remote_cksum[file_name] != local_cksum[file_name]:
731
          if node_is_mc or must_have_file:
732
            bad = True
733
            feedback_fn("  - ERROR: file '%s' has wrong checksum" % file_name)
734
          else:
735
            # not candidate and this is not a must-have file
736
            bad = True
737
            feedback_fn("  - ERROR: non master-candidate has old/wrong file"
738
                        " '%s'" % file_name)
739
        else:
740
          # all good, except non-master/non-must have combination
741
          if not node_is_mc and not must_have_file:
742
            feedback_fn("  - ERROR: file '%s' should not exist on non master"
743
                        " candidates" % file_name)
744

    
745
    # checks ssh to any
746

    
747
    if constants.NV_NODELIST not in node_result:
748
      bad = True
749
      feedback_fn("  - ERROR: node hasn't returned node ssh connectivity data")
750
    else:
751
      if node_result[constants.NV_NODELIST]:
752
        bad = True
753
        for node in node_result[constants.NV_NODELIST]:
754
          feedback_fn("  - ERROR: ssh communication with node '%s': %s" %
755
                          (node, node_result[constants.NV_NODELIST][node]))
756

    
757
    if constants.NV_NODENETTEST not in node_result:
758
      bad = True
759
      feedback_fn("  - ERROR: node hasn't returned node tcp connectivity data")
760
    else:
761
      if node_result[constants.NV_NODENETTEST]:
762
        bad = True
763
        nlist = utils.NiceSort(node_result[constants.NV_NODENETTEST].keys())
764
        for node in nlist:
765
          feedback_fn("  - ERROR: tcp communication with node '%s': %s" %
766
                          (node, node_result[constants.NV_NODENETTEST][node]))
767

    
768
    hyp_result = node_result.get(constants.NV_HYPERVISOR, None)
769
    if isinstance(hyp_result, dict):
770
      for hv_name, hv_result in hyp_result.iteritems():
771
        if hv_result is not None:
772
          feedback_fn("  - ERROR: hypervisor %s verify failure: '%s'" %
773
                      (hv_name, hv_result))
774

    
775
    # check used drbd list
776
    if vg_name is not None:
777
      used_minors = node_result.get(constants.NV_DRBDLIST, [])
778
      if not isinstance(used_minors, (tuple, list)):
779
        feedback_fn("  - ERROR: cannot parse drbd status file: %s" %
780
                    str(used_minors))
781
      else:
782
        for minor, (iname, must_exist) in drbd_map.items():
783
          if minor not in used_minors and must_exist:
784
            feedback_fn("  - ERROR: drbd minor %d of instance %s is"
785
                        " not active" % (minor, iname))
786
            bad = True
787
        for minor in used_minors:
788
          if minor not in drbd_map:
789
            feedback_fn("  - ERROR: unallocated drbd minor %d is in use" %
790
                        minor)
791
            bad = True
792

    
793
    return bad
794

    
795
  def _VerifyInstance(self, instance, instanceconfig, node_vol_is,
796
                      node_instance, feedback_fn, n_offline):
797
    """Verify an instance.
798

799
    This function checks to see if the required block devices are
800
    available on the instance's node.
801

802
    """
803
    bad = False
804

    
805
    node_current = instanceconfig.primary_node
806

    
807
    node_vol_should = {}
808
    instanceconfig.MapLVsByNode(node_vol_should)
809

    
810
    for node in node_vol_should:
811
      if node in n_offline:
812
        # ignore missing volumes on offline nodes
813
        continue
814
      for volume in node_vol_should[node]:
815
        if node not in node_vol_is or volume not in node_vol_is[node]:
816
          feedback_fn("  - ERROR: volume %s missing on node %s" %
817
                          (volume, node))
818
          bad = True
819

    
820
    if instanceconfig.admin_up:
821
      if ((node_current not in node_instance or
822
          not instance in node_instance[node_current]) and
823
          node_current not in n_offline):
824
        feedback_fn("  - ERROR: instance %s not running on node %s" %
825
                        (instance, node_current))
826
        bad = True
827

    
828
    for node in node_instance:
829
      if (not node == node_current):
830
        if instance in node_instance[node]:
831
          feedback_fn("  - ERROR: instance %s should not run on node %s" %
832
                          (instance, node))
833
          bad = True
834

    
835
    return bad
836

    
837
  def _VerifyOrphanVolumes(self, node_vol_should, node_vol_is, feedback_fn):
838
    """Verify if there are any unknown volumes in the cluster.
839

840
    The .os, .swap and backup volumes are ignored. All other volumes are
841
    reported as unknown.
842

843
    """
844
    bad = False
845

    
846
    for node in node_vol_is:
847
      for volume in node_vol_is[node]:
848
        if node not in node_vol_should or volume not in node_vol_should[node]:
849
          feedback_fn("  - ERROR: volume %s on node %s should not exist" %
850
                      (volume, node))
851
          bad = True
852
    return bad
853

    
854
  def _VerifyOrphanInstances(self, instancelist, node_instance, feedback_fn):
855
    """Verify the list of running instances.
856

857
    This checks what instances are running but unknown to the cluster.
858

859
    """
860
    bad = False
861
    for node in node_instance:
862
      for runninginstance in node_instance[node]:
863
        if runninginstance not in instancelist:
864
          feedback_fn("  - ERROR: instance %s on node %s should not exist" %
865
                          (runninginstance, node))
866
          bad = True
867
    return bad
868

    
869
  def _VerifyNPlusOneMemory(self, node_info, instance_cfg, feedback_fn):
870
    """Verify N+1 Memory Resilience.
871

872
    Check that if one single node dies we can still start all the instances it
873
    was primary for.
874

875
    """
876
    bad = False
877

    
878
    for node, nodeinfo in node_info.iteritems():
879
      # This code checks that every node which is now listed as secondary has
880
      # enough memory to host all instances it is supposed to should a single
881
      # other node in the cluster fail.
882
      # FIXME: not ready for failover to an arbitrary node
883
      # FIXME: does not support file-backed instances
884
      # WARNING: we currently take into account down instances as well as up
885
      # ones, considering that even if they're down someone might want to start
886
      # them even in the event of a node failure.
887
      for prinode, instances in nodeinfo['sinst-by-pnode'].iteritems():
888
        needed_mem = 0
889
        for instance in instances:
890
          bep = self.cfg.GetClusterInfo().FillBE(instance_cfg[instance])
891
          if bep[constants.BE_AUTO_BALANCE]:
892
            needed_mem += bep[constants.BE_MEMORY]
893
        if nodeinfo['mfree'] < needed_mem:
894
          feedback_fn("  - ERROR: not enough memory on node %s to accomodate"
895
                      " failovers should node %s fail" % (node, prinode))
896
          bad = True
897
    return bad
898

    
899
  def CheckPrereq(self):
900
    """Check prerequisites.
901

902
    Transform the list of checks we're going to skip into a set and check that
903
    all its members are valid.
904

905
    """
906
    self.skip_set = frozenset(self.op.skip_checks)
907
    if not constants.VERIFY_OPTIONAL_CHECKS.issuperset(self.skip_set):
908
      raise errors.OpPrereqError("Invalid checks to be skipped specified")
909

    
910
  def BuildHooksEnv(self):
911
    """Build hooks env.
912

913
    Cluster-Verify hooks just rone in the post phase and their failure makes
914
    the output be logged in the verify output and the verification to fail.
915

916
    """
917
    all_nodes = self.cfg.GetNodeList()
918
    env = {
919
      "CLUSTER_TAGS": " ".join(self.cfg.GetClusterInfo().GetTags())
920
      }
921
    for node in self.cfg.GetAllNodesInfo().values():
922
      env["NODE_TAGS_%s" % node.name] = " ".join(node.GetTags())
923

    
924
    return env, [], all_nodes
925

    
926
  def Exec(self, feedback_fn):
927
    """Verify integrity of cluster, performing various test on nodes.
928

929
    """
930
    bad = False
931
    feedback_fn("* Verifying global settings")
932
    for msg in self.cfg.VerifyConfig():
933
      feedback_fn("  - ERROR: %s" % msg)
934

    
935
    vg_name = self.cfg.GetVGName()
936
    hypervisors = self.cfg.GetClusterInfo().enabled_hypervisors
937
    nodelist = utils.NiceSort(self.cfg.GetNodeList())
938
    nodeinfo = [self.cfg.GetNodeInfo(nname) for nname in nodelist]
939
    instancelist = utils.NiceSort(self.cfg.GetInstanceList())
940
    instanceinfo = dict((iname, self.cfg.GetInstanceInfo(iname))
941
                        for iname in instancelist)
942
    i_non_redundant = [] # Non redundant instances
943
    i_non_a_balanced = [] # Non auto-balanced instances
944
    n_offline = [] # List of offline nodes
945
    n_drained = [] # List of nodes being drained
946
    node_volume = {}
947
    node_instance = {}
948
    node_info = {}
949
    instance_cfg = {}
950

    
951
    # FIXME: verify OS list
952
    # do local checksums
953
    master_files = [constants.CLUSTER_CONF_FILE]
954

    
955
    file_names = ssconf.SimpleStore().GetFileList()
956
    file_names.append(constants.SSL_CERT_FILE)
957
    file_names.append(constants.RAPI_CERT_FILE)
958
    file_names.extend(master_files)
959

    
960
    local_checksums = utils.FingerprintFiles(file_names)
961

    
962
    feedback_fn("* Gathering data (%d nodes)" % len(nodelist))
963
    node_verify_param = {
964
      constants.NV_FILELIST: file_names,
965
      constants.NV_NODELIST: [node.name for node in nodeinfo
966
                              if not node.offline],
967
      constants.NV_HYPERVISOR: hypervisors,
968
      constants.NV_NODENETTEST: [(node.name, node.primary_ip,
969
                                  node.secondary_ip) for node in nodeinfo
970
                                 if not node.offline],
971
      constants.NV_INSTANCELIST: hypervisors,
972
      constants.NV_VERSION: None,
973
      constants.NV_HVINFO: self.cfg.GetHypervisorType(),
974
      }
975
    if vg_name is not None:
976
      node_verify_param[constants.NV_VGLIST] = None
977
      node_verify_param[constants.NV_LVLIST] = vg_name
978
      node_verify_param[constants.NV_DRBDLIST] = None
979
    all_nvinfo = self.rpc.call_node_verify(nodelist, node_verify_param,
980
                                           self.cfg.GetClusterName())
981

    
982
    cluster = self.cfg.GetClusterInfo()
983
    master_node = self.cfg.GetMasterNode()
984
    all_drbd_map = self.cfg.ComputeDRBDMap()
985

    
986
    for node_i in nodeinfo:
987
      node = node_i.name
988
      nresult = all_nvinfo[node].data
989

    
990
      if node_i.offline:
991
        feedback_fn("* Skipping offline node %s" % (node,))
992
        n_offline.append(node)
993
        continue
994

    
995
      if node == master_node:
996
        ntype = "master"
997
      elif node_i.master_candidate:
998
        ntype = "master candidate"
999
      elif node_i.drained:
1000
        ntype = "drained"
1001
        n_drained.append(node)
1002
      else:
1003
        ntype = "regular"
1004
      feedback_fn("* Verifying node %s (%s)" % (node, ntype))
1005

    
1006
      if all_nvinfo[node].failed or not isinstance(nresult, dict):
1007
        feedback_fn("  - ERROR: connection to %s failed" % (node,))
1008
        bad = True
1009
        continue
1010

    
1011
      node_drbd = {}
1012
      for minor, instance in all_drbd_map[node].items():
1013
        if instance not in instanceinfo:
1014
          feedback_fn("  - ERROR: ghost instance '%s' in temporary DRBD map" %
1015
                      instance)
1016
          # ghost instance should not be running, but otherwise we
1017
          # don't give double warnings (both ghost instance and
1018
          # unallocated minor in use)
1019
          node_drbd[minor] = (instance, False)
1020
        else:
1021
          instance = instanceinfo[instance]
1022
          node_drbd[minor] = (instance.name, instance.admin_up)
1023
      result = self._VerifyNode(node_i, file_names, local_checksums,
1024
                                nresult, feedback_fn, master_files,
1025
                                node_drbd, vg_name)
1026
      bad = bad or result
1027

    
1028
      lvdata = nresult.get(constants.NV_LVLIST, "Missing LV data")
1029
      if vg_name is None:
1030
        node_volume[node] = {}
1031
      elif isinstance(lvdata, basestring):
1032
        feedback_fn("  - ERROR: LVM problem on node %s: %s" %
1033
                    (node, utils.SafeEncode(lvdata)))
1034
        bad = True
1035
        node_volume[node] = {}
1036
      elif not isinstance(lvdata, dict):
1037
        feedback_fn("  - ERROR: connection to %s failed (lvlist)" % (node,))
1038
        bad = True
1039
        continue
1040
      else:
1041
        node_volume[node] = lvdata
1042

    
1043
      # node_instance
1044
      idata = nresult.get(constants.NV_INSTANCELIST, None)
1045
      if not isinstance(idata, list):
1046
        feedback_fn("  - ERROR: connection to %s failed (instancelist)" %
1047
                    (node,))
1048
        bad = True
1049
        continue
1050

    
1051
      node_instance[node] = idata
1052

    
1053
      # node_info
1054
      nodeinfo = nresult.get(constants.NV_HVINFO, None)
1055
      if not isinstance(nodeinfo, dict):
1056
        feedback_fn("  - ERROR: connection to %s failed (hvinfo)" % (node,))
1057
        bad = True
1058
        continue
1059

    
1060
      try:
1061
        node_info[node] = {
1062
          "mfree": int(nodeinfo['memory_free']),
1063
          "pinst": [],
1064
          "sinst": [],
1065
          # dictionary holding all instances this node is secondary for,
1066
          # grouped by their primary node. Each key is a cluster node, and each
1067
          # value is a list of instances which have the key as primary and the
1068
          # current node as secondary.  this is handy to calculate N+1 memory
1069
          # availability if you can only failover from a primary to its
1070
          # secondary.
1071
          "sinst-by-pnode": {},
1072
        }
1073
        # FIXME: devise a free space model for file based instances as well
1074
        if vg_name is not None:
1075
          if (constants.NV_VGLIST not in nresult or
1076
              vg_name not in nresult[constants.NV_VGLIST]):
1077
            feedback_fn("  - ERROR: node %s didn't return data for the"
1078
                        " volume group '%s' - it is either missing or broken" %
1079
                        (node, vg_name))
1080
            bad = True
1081
            continue
1082
          node_info[node]["dfree"] = int(nresult[constants.NV_VGLIST][vg_name])
1083
      except (ValueError, KeyError):
1084
        feedback_fn("  - ERROR: invalid nodeinfo value returned"
1085
                    " from node %s" % (node,))
1086
        bad = True
1087
        continue
1088

    
1089
    node_vol_should = {}
1090

    
1091
    for instance in instancelist:
1092
      feedback_fn("* Verifying instance %s" % instance)
1093
      inst_config = instanceinfo[instance]
1094
      result =  self._VerifyInstance(instance, inst_config, node_volume,
1095
                                     node_instance, feedback_fn, n_offline)
1096
      bad = bad or result
1097
      inst_nodes_offline = []
1098

    
1099
      inst_config.MapLVsByNode(node_vol_should)
1100

    
1101
      instance_cfg[instance] = inst_config
1102

    
1103
      pnode = inst_config.primary_node
1104
      if pnode in node_info:
1105
        node_info[pnode]['pinst'].append(instance)
1106
      elif pnode not in n_offline:
1107
        feedback_fn("  - ERROR: instance %s, connection to primary node"
1108
                    " %s failed" % (instance, pnode))
1109
        bad = True
1110

    
1111
      if pnode in n_offline:
1112
        inst_nodes_offline.append(pnode)
1113

    
1114
      # If the instance is non-redundant we cannot survive losing its primary
1115
      # node, so we are not N+1 compliant. On the other hand we have no disk
1116
      # templates with more than one secondary so that situation is not well
1117
      # supported either.
1118
      # FIXME: does not support file-backed instances
1119
      if len(inst_config.secondary_nodes) == 0:
1120
        i_non_redundant.append(instance)
1121
      elif len(inst_config.secondary_nodes) > 1:
1122
        feedback_fn("  - WARNING: multiple secondaries for instance %s"
1123
                    % instance)
1124

    
1125
      if not cluster.FillBE(inst_config)[constants.BE_AUTO_BALANCE]:
1126
        i_non_a_balanced.append(instance)
1127

    
1128
      for snode in inst_config.secondary_nodes:
1129
        if snode in node_info:
1130
          node_info[snode]['sinst'].append(instance)
1131
          if pnode not in node_info[snode]['sinst-by-pnode']:
1132
            node_info[snode]['sinst-by-pnode'][pnode] = []
1133
          node_info[snode]['sinst-by-pnode'][pnode].append(instance)
1134
        elif snode not in n_offline:
1135
          feedback_fn("  - ERROR: instance %s, connection to secondary node"
1136
                      " %s failed" % (instance, snode))
1137
          bad = True
1138
        if snode in n_offline:
1139
          inst_nodes_offline.append(snode)
1140

    
1141
      if inst_nodes_offline:
1142
        # warn that the instance lives on offline nodes, and set bad=True
1143
        feedback_fn("  - ERROR: instance lives on offline node(s) %s" %
1144
                    ", ".join(inst_nodes_offline))
1145
        bad = True
1146

    
1147
    feedback_fn("* Verifying orphan volumes")
1148
    result = self._VerifyOrphanVolumes(node_vol_should, node_volume,
1149
                                       feedback_fn)
1150
    bad = bad or result
1151

    
1152
    feedback_fn("* Verifying remaining instances")
1153
    result = self._VerifyOrphanInstances(instancelist, node_instance,
1154
                                         feedback_fn)
1155
    bad = bad or result
1156

    
1157
    if constants.VERIFY_NPLUSONE_MEM not in self.skip_set:
1158
      feedback_fn("* Verifying N+1 Memory redundancy")
1159
      result = self._VerifyNPlusOneMemory(node_info, instance_cfg, feedback_fn)
1160
      bad = bad or result
1161

    
1162
    feedback_fn("* Other Notes")
1163
    if i_non_redundant:
1164
      feedback_fn("  - NOTICE: %d non-redundant instance(s) found."
1165
                  % len(i_non_redundant))
1166

    
1167
    if i_non_a_balanced:
1168
      feedback_fn("  - NOTICE: %d non-auto-balanced instance(s) found."
1169
                  % len(i_non_a_balanced))
1170

    
1171
    if n_offline:
1172
      feedback_fn("  - NOTICE: %d offline node(s) found." % len(n_offline))
1173

    
1174
    if n_drained:
1175
      feedback_fn("  - NOTICE: %d drained node(s) found." % len(n_drained))
1176

    
1177
    return not bad
1178

    
1179
  def HooksCallBack(self, phase, hooks_results, feedback_fn, lu_result):
1180
    """Analize the post-hooks' result
1181

1182
    This method analyses the hook result, handles it, and sends some
1183
    nicely-formatted feedback back to the user.
1184

1185
    @param phase: one of L{constants.HOOKS_PHASE_POST} or
1186
        L{constants.HOOKS_PHASE_PRE}; it denotes the hooks phase
1187
    @param hooks_results: the results of the multi-node hooks rpc call
1188
    @param feedback_fn: function used send feedback back to the caller
1189
    @param lu_result: previous Exec result
1190
    @return: the new Exec result, based on the previous result
1191
        and hook results
1192

1193
    """
1194
    # We only really run POST phase hooks, and are only interested in
1195
    # their results
1196
    if phase == constants.HOOKS_PHASE_POST:
1197
      # Used to change hooks' output to proper indentation
1198
      indent_re = re.compile('^', re.M)
1199
      feedback_fn("* Hooks Results")
1200
      if not hooks_results:
1201
        feedback_fn("  - ERROR: general communication failure")
1202
        lu_result = 1
1203
      else:
1204
        for node_name in hooks_results:
1205
          show_node_header = True
1206
          res = hooks_results[node_name]
1207
          if res.failed or res.data is False or not isinstance(res.data, list):
1208
            if res.offline:
1209
              # no need to warn or set fail return value
1210
              continue
1211
            feedback_fn("    Communication failure in hooks execution")
1212
            lu_result = 1
1213
            continue
1214
          for script, hkr, output in res.data:
1215
            if hkr == constants.HKR_FAIL:
1216
              # The node header is only shown once, if there are
1217
              # failing hooks on that node
1218
              if show_node_header:
1219
                feedback_fn("  Node %s:" % node_name)
1220
                show_node_header = False
1221
              feedback_fn("    ERROR: Script %s failed, output:" % script)
1222
              output = indent_re.sub('      ', output)
1223
              feedback_fn("%s" % output)
1224
              lu_result = 1
1225

    
1226
      return lu_result
1227

    
1228

    
1229
class LUVerifyDisks(NoHooksLU):
1230
  """Verifies the cluster disks status.
1231

1232
  """
1233
  _OP_REQP = []
1234
  REQ_BGL = False
1235

    
1236
  def ExpandNames(self):
1237
    self.needed_locks = {
1238
      locking.LEVEL_NODE: locking.ALL_SET,
1239
      locking.LEVEL_INSTANCE: locking.ALL_SET,
1240
    }
1241
    self.share_locks = dict(((i, 1) for i in locking.LEVELS))
1242

    
1243
  def CheckPrereq(self):
1244
    """Check prerequisites.
1245

1246
    This has no prerequisites.
1247

1248
    """
1249
    pass
1250

    
1251
  def Exec(self, feedback_fn):
1252
    """Verify integrity of cluster disks.
1253

1254
    """
1255
    result = res_nodes, res_nlvm, res_instances, res_missing = [], {}, [], {}
1256

    
1257
    vg_name = self.cfg.GetVGName()
1258
    nodes = utils.NiceSort(self.cfg.GetNodeList())
1259
    instances = [self.cfg.GetInstanceInfo(name)
1260
                 for name in self.cfg.GetInstanceList()]
1261

    
1262
    nv_dict = {}
1263
    for inst in instances:
1264
      inst_lvs = {}
1265
      if (not inst.admin_up or
1266
          inst.disk_template not in constants.DTS_NET_MIRROR):
1267
        continue
1268
      inst.MapLVsByNode(inst_lvs)
1269
      # transform { iname: {node: [vol,],},} to {(node, vol): iname}
1270
      for node, vol_list in inst_lvs.iteritems():
1271
        for vol in vol_list:
1272
          nv_dict[(node, vol)] = inst
1273

    
1274
    if not nv_dict:
1275
      return result
1276

    
1277
    node_lvs = self.rpc.call_volume_list(nodes, vg_name)
1278

    
1279
    to_act = set()
1280
    for node in nodes:
1281
      # node_volume
1282
      lvs = node_lvs[node]
1283
      if lvs.failed:
1284
        if not lvs.offline:
1285
          self.LogWarning("Connection to node %s failed: %s" %
1286
                          (node, lvs.data))
1287
        continue
1288
      lvs = lvs.data
1289
      if isinstance(lvs, basestring):
1290
        logging.warning("Error enumerating LVs on node %s: %s", node, lvs)
1291
        res_nlvm[node] = lvs
1292
        continue
1293
      elif not isinstance(lvs, dict):
1294
        logging.warning("Connection to node %s failed or invalid data"
1295
                        " returned", node)
1296
        res_nodes.append(node)
1297
        continue
1298

    
1299
      for lv_name, (_, lv_inactive, lv_online) in lvs.iteritems():
1300
        inst = nv_dict.pop((node, lv_name), None)
1301
        if (not lv_online and inst is not None
1302
            and inst.name not in res_instances):
1303
          res_instances.append(inst.name)
1304

    
1305
    # any leftover items in nv_dict are missing LVs, let's arrange the
1306
    # data better
1307
    for key, inst in nv_dict.iteritems():
1308
      if inst.name not in res_missing:
1309
        res_missing[inst.name] = []
1310
      res_missing[inst.name].append(key)
1311

    
1312
    return result
1313

    
1314

    
1315
class LURenameCluster(LogicalUnit):
1316
  """Rename the cluster.
1317

1318
  """
1319
  HPATH = "cluster-rename"
1320
  HTYPE = constants.HTYPE_CLUSTER
1321
  _OP_REQP = ["name"]
1322

    
1323
  def BuildHooksEnv(self):
1324
    """Build hooks env.
1325

1326
    """
1327
    env = {
1328
      "OP_TARGET": self.cfg.GetClusterName(),
1329
      "NEW_NAME": self.op.name,
1330
      }
1331
    mn = self.cfg.GetMasterNode()
1332
    return env, [mn], [mn]
1333

    
1334
  def CheckPrereq(self):
1335
    """Verify that the passed name is a valid one.
1336

1337
    """
1338
    hostname = utils.HostInfo(self.op.name)
1339

    
1340
    new_name = hostname.name
1341
    self.ip = new_ip = hostname.ip
1342
    old_name = self.cfg.GetClusterName()
1343
    old_ip = self.cfg.GetMasterIP()
1344
    if new_name == old_name and new_ip == old_ip:
1345
      raise errors.OpPrereqError("Neither the name nor the IP address of the"
1346
                                 " cluster has changed")
1347
    if new_ip != old_ip:
1348
      if utils.TcpPing(new_ip, constants.DEFAULT_NODED_PORT):
1349
        raise errors.OpPrereqError("The given cluster IP address (%s) is"
1350
                                   " reachable on the network. Aborting." %
1351
                                   new_ip)
1352

    
1353
    self.op.name = new_name
1354

    
1355
  def Exec(self, feedback_fn):
1356
    """Rename the cluster.
1357

1358
    """
1359
    clustername = self.op.name
1360
    ip = self.ip
1361

    
1362
    # shutdown the master IP
1363
    master = self.cfg.GetMasterNode()
1364
    result = self.rpc.call_node_stop_master(master, False)
1365
    if result.failed or not result.data:
1366
      raise errors.OpExecError("Could not disable the master role")
1367

    
1368
    try:
1369
      cluster = self.cfg.GetClusterInfo()
1370
      cluster.cluster_name = clustername
1371
      cluster.master_ip = ip
1372
      self.cfg.Update(cluster)
1373

    
1374
      # update the known hosts file
1375
      ssh.WriteKnownHostsFile(self.cfg, constants.SSH_KNOWN_HOSTS_FILE)
1376
      node_list = self.cfg.GetNodeList()
1377
      try:
1378
        node_list.remove(master)
1379
      except ValueError:
1380
        pass
1381
      result = self.rpc.call_upload_file(node_list,
1382
                                         constants.SSH_KNOWN_HOSTS_FILE)
1383
      for to_node, to_result in result.iteritems():
1384
        if to_result.failed or not to_result.data:
1385
          logging.error("Copy of file %s to node %s failed",
1386
                        constants.SSH_KNOWN_HOSTS_FILE, to_node)
1387

    
1388
    finally:
1389
      result = self.rpc.call_node_start_master(master, False)
1390
      if result.failed or not result.data:
1391
        self.LogWarning("Could not re-enable the master role on"
1392
                        " the master, please restart manually.")
1393

    
1394

    
1395
def _RecursiveCheckIfLVMBased(disk):
1396
  """Check if the given disk or its children are lvm-based.
1397

1398
  @type disk: L{objects.Disk}
1399
  @param disk: the disk to check
1400
  @rtype: booleean
1401
  @return: boolean indicating whether a LD_LV dev_type was found or not
1402

1403
  """
1404
  if disk.children:
1405
    for chdisk in disk.children:
1406
      if _RecursiveCheckIfLVMBased(chdisk):
1407
        return True
1408
  return disk.dev_type == constants.LD_LV
1409

    
1410

    
1411
class LUSetClusterParams(LogicalUnit):
1412
  """Change the parameters of the cluster.
1413

1414
  """
1415
  HPATH = "cluster-modify"
1416
  HTYPE = constants.HTYPE_CLUSTER
1417
  _OP_REQP = []
1418
  REQ_BGL = False
1419

    
1420
  def CheckParameters(self):
1421
    """Check parameters
1422

1423
    """
1424
    if not hasattr(self.op, "candidate_pool_size"):
1425
      self.op.candidate_pool_size = None
1426
    if self.op.candidate_pool_size is not None:
1427
      try:
1428
        self.op.candidate_pool_size = int(self.op.candidate_pool_size)
1429
      except ValueError, err:
1430
        raise errors.OpPrereqError("Invalid candidate_pool_size value: %s" %
1431
                                   str(err))
1432
      if self.op.candidate_pool_size < 1:
1433
        raise errors.OpPrereqError("At least one master candidate needed")
1434

    
1435
  def ExpandNames(self):
1436
    # FIXME: in the future maybe other cluster params won't require checking on
1437
    # all nodes to be modified.
1438
    self.needed_locks = {
1439
      locking.LEVEL_NODE: locking.ALL_SET,
1440
    }
1441
    self.share_locks[locking.LEVEL_NODE] = 1
1442

    
1443
  def BuildHooksEnv(self):
1444
    """Build hooks env.
1445

1446
    """
1447
    env = {
1448
      "OP_TARGET": self.cfg.GetClusterName(),
1449
      "NEW_VG_NAME": self.op.vg_name,
1450
      }
1451
    mn = self.cfg.GetMasterNode()
1452
    return env, [mn], [mn]
1453

    
1454
  def CheckPrereq(self):
1455
    """Check prerequisites.
1456

1457
    This checks whether the given params don't conflict and
1458
    if the given volume group is valid.
1459

1460
    """
1461
    if self.op.vg_name is not None and not self.op.vg_name:
1462
      instances = self.cfg.GetAllInstancesInfo().values()
1463
      for inst in instances:
1464
        for disk in inst.disks:
1465
          if _RecursiveCheckIfLVMBased(disk):
1466
            raise errors.OpPrereqError("Cannot disable lvm storage while"
1467
                                       " lvm-based instances exist")
1468

    
1469
    node_list = self.acquired_locks[locking.LEVEL_NODE]
1470

    
1471
    # if vg_name not None, checks given volume group on all nodes
1472
    if self.op.vg_name:
1473
      vglist = self.rpc.call_vg_list(node_list)
1474
      for node in node_list:
1475
        if vglist[node].failed:
1476
          # ignoring down node
1477
          self.LogWarning("Node %s unreachable/error, ignoring" % node)
1478
          continue
1479
        vgstatus = utils.CheckVolumeGroupSize(vglist[node].data,
1480
                                              self.op.vg_name,
1481
                                              constants.MIN_VG_SIZE)
1482
        if vgstatus:
1483
          raise errors.OpPrereqError("Error on node '%s': %s" %
1484
                                     (node, vgstatus))
1485

    
1486
    self.cluster = cluster = self.cfg.GetClusterInfo()
1487
    # validate beparams changes
1488
    if self.op.beparams:
1489
      utils.ForceDictType(self.op.beparams, constants.BES_PARAMETER_TYPES)
1490
      self.new_beparams = cluster.FillDict(
1491
        cluster.beparams[constants.BEGR_DEFAULT], self.op.beparams)
1492

    
1493
    # hypervisor list/parameters
1494
    self.new_hvparams = cluster.FillDict(cluster.hvparams, {})
1495
    if self.op.hvparams:
1496
      if not isinstance(self.op.hvparams, dict):
1497
        raise errors.OpPrereqError("Invalid 'hvparams' parameter on input")
1498
      for hv_name, hv_dict in self.op.hvparams.items():
1499
        if hv_name not in self.new_hvparams:
1500
          self.new_hvparams[hv_name] = hv_dict
1501
        else:
1502
          self.new_hvparams[hv_name].update(hv_dict)
1503

    
1504
    if self.op.enabled_hypervisors is not None:
1505
      self.hv_list = self.op.enabled_hypervisors
1506
    else:
1507
      self.hv_list = cluster.enabled_hypervisors
1508

    
1509
    if self.op.hvparams or self.op.enabled_hypervisors is not None:
1510
      # either the enabled list has changed, or the parameters have, validate
1511
      for hv_name, hv_params in self.new_hvparams.items():
1512
        if ((self.op.hvparams and hv_name in self.op.hvparams) or
1513
            (self.op.enabled_hypervisors and
1514
             hv_name in self.op.enabled_hypervisors)):
1515
          # either this is a new hypervisor, or its parameters have changed
1516
          hv_class = hypervisor.GetHypervisor(hv_name)
1517
          utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
1518
          hv_class.CheckParameterSyntax(hv_params)
1519
          _CheckHVParams(self, node_list, hv_name, hv_params)
1520

    
1521
  def Exec(self, feedback_fn):
1522
    """Change the parameters of the cluster.
1523

1524
    """
1525
    if self.op.vg_name is not None:
1526
      new_volume = self.op.vg_name
1527
      if not new_volume:
1528
        new_volume = None
1529
      if new_volume != self.cfg.GetVGName():
1530
        self.cfg.SetVGName(new_volume)
1531
      else:
1532
        feedback_fn("Cluster LVM configuration already in desired"
1533
                    " state, not changing")
1534
    if self.op.hvparams:
1535
      self.cluster.hvparams = self.new_hvparams
1536
    if self.op.enabled_hypervisors is not None:
1537
      self.cluster.enabled_hypervisors = self.op.enabled_hypervisors
1538
    if self.op.beparams:
1539
      self.cluster.beparams[constants.BEGR_DEFAULT] = self.new_beparams
1540
    if self.op.candidate_pool_size is not None:
1541
      self.cluster.candidate_pool_size = self.op.candidate_pool_size
1542

    
1543
    self.cfg.Update(self.cluster)
1544

    
1545
    # we want to update nodes after the cluster so that if any errors
1546
    # happen, we have recorded and saved the cluster info
1547
    if self.op.candidate_pool_size is not None:
1548
      _AdjustCandidatePool(self)
1549

    
1550

    
1551
class LURedistributeConfig(NoHooksLU):
1552
  """Force the redistribution of cluster configuration.
1553

1554
  This is a very simple LU.
1555

1556
  """
1557
  _OP_REQP = []
1558
  REQ_BGL = False
1559

    
1560
  def ExpandNames(self):
1561
    self.needed_locks = {
1562
      locking.LEVEL_NODE: locking.ALL_SET,
1563
    }
1564
    self.share_locks[locking.LEVEL_NODE] = 1
1565

    
1566
  def CheckPrereq(self):
1567
    """Check prerequisites.
1568

1569
    """
1570

    
1571
  def Exec(self, feedback_fn):
1572
    """Redistribute the configuration.
1573

1574
    """
1575
    self.cfg.Update(self.cfg.GetClusterInfo())
1576

    
1577

    
1578
def _WaitForSync(lu, instance, oneshot=False, unlock=False):
1579
  """Sleep and poll for an instance's disk to sync.
1580

1581
  """
1582
  if not instance.disks:
1583
    return True
1584

    
1585
  if not oneshot:
1586
    lu.proc.LogInfo("Waiting for instance %s to sync disks." % instance.name)
1587

    
1588
  node = instance.primary_node
1589

    
1590
  for dev in instance.disks:
1591
    lu.cfg.SetDiskID(dev, node)
1592

    
1593
  retries = 0
1594
  while True:
1595
    max_time = 0
1596
    done = True
1597
    cumul_degraded = False
1598
    rstats = lu.rpc.call_blockdev_getmirrorstatus(node, instance.disks)
1599
    if rstats.failed or not rstats.data:
1600
      lu.LogWarning("Can't get any data from node %s", node)
1601
      retries += 1
1602
      if retries >= 10:
1603
        raise errors.RemoteError("Can't contact node %s for mirror data,"
1604
                                 " aborting." % node)
1605
      time.sleep(6)
1606
      continue
1607
    rstats = rstats.data
1608
    retries = 0
1609
    for i, mstat in enumerate(rstats):
1610
      if mstat is None:
1611
        lu.LogWarning("Can't compute data for node %s/%s",
1612
                           node, instance.disks[i].iv_name)
1613
        continue
1614
      # we ignore the ldisk parameter
1615
      perc_done, est_time, is_degraded, _ = mstat
1616
      cumul_degraded = cumul_degraded or (is_degraded and perc_done is None)
1617
      if perc_done is not None:
1618
        done = False
1619
        if est_time is not None:
1620
          rem_time = "%d estimated seconds remaining" % est_time
1621
          max_time = est_time
1622
        else:
1623
          rem_time = "no time estimate"
1624
        lu.proc.LogInfo("- device %s: %5.2f%% done, %s" %
1625
                        (instance.disks[i].iv_name, perc_done, rem_time))
1626
    if done or oneshot:
1627
      break
1628

    
1629
    time.sleep(min(60, max_time))
1630

    
1631
  if done:
1632
    lu.proc.LogInfo("Instance %s's disks are in sync." % instance.name)
1633
  return not cumul_degraded
1634

    
1635

    
1636
def _CheckDiskConsistency(lu, dev, node, on_primary, ldisk=False):
1637
  """Check that mirrors are not degraded.
1638

1639
  The ldisk parameter, if True, will change the test from the
1640
  is_degraded attribute (which represents overall non-ok status for
1641
  the device(s)) to the ldisk (representing the local storage status).
1642

1643
  """
1644
  lu.cfg.SetDiskID(dev, node)
1645
  if ldisk:
1646
    idx = 6
1647
  else:
1648
    idx = 5
1649

    
1650
  result = True
1651
  if on_primary or dev.AssembleOnSecondary():
1652
    rstats = lu.rpc.call_blockdev_find(node, dev)
1653
    msg = rstats.RemoteFailMsg()
1654
    if msg:
1655
      lu.LogWarning("Can't find disk on node %s: %s", node, msg)
1656
      result = False
1657
    elif not rstats.payload:
1658
      lu.LogWarning("Can't find disk on node %s", node)
1659
      result = False
1660
    else:
1661
      result = result and (not rstats.payload[idx])
1662
  if dev.children:
1663
    for child in dev.children:
1664
      result = result and _CheckDiskConsistency(lu, child, node, on_primary)
1665

    
1666
  return result
1667

    
1668

    
1669
class LUDiagnoseOS(NoHooksLU):
1670
  """Logical unit for OS diagnose/query.
1671

1672
  """
1673
  _OP_REQP = ["output_fields", "names"]
1674
  REQ_BGL = False
1675
  _FIELDS_STATIC = utils.FieldSet()
1676
  _FIELDS_DYNAMIC = utils.FieldSet("name", "valid", "node_status")
1677

    
1678
  def ExpandNames(self):
1679
    if self.op.names:
1680
      raise errors.OpPrereqError("Selective OS query not supported")
1681

    
1682
    _CheckOutputFields(static=self._FIELDS_STATIC,
1683
                       dynamic=self._FIELDS_DYNAMIC,
1684
                       selected=self.op.output_fields)
1685

    
1686
    # Lock all nodes, in shared mode
1687
    # Temporary removal of locks, should be reverted later
1688
    # TODO: reintroduce locks when they are lighter-weight
1689
    self.needed_locks = {}
1690
    #self.share_locks[locking.LEVEL_NODE] = 1
1691
    #self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
1692

    
1693
  def CheckPrereq(self):
1694
    """Check prerequisites.
1695

1696
    """
1697

    
1698
  @staticmethod
1699
  def _DiagnoseByOS(node_list, rlist):
1700
    """Remaps a per-node return list into an a per-os per-node dictionary
1701

1702
    @param node_list: a list with the names of all nodes
1703
    @param rlist: a map with node names as keys and OS objects as values
1704

1705
    @rtype: dict
1706
    @return: a dictionary with osnames as keys and as value another map, with
1707
        nodes as keys and list of OS objects as values, eg::
1708

1709
          {"debian-etch": {"node1": [<object>,...],
1710
                           "node2": [<object>,]}
1711
          }
1712

1713
    """
1714
    all_os = {}
1715
    # we build here the list of nodes that didn't fail the RPC (at RPC
1716
    # level), so that nodes with a non-responding node daemon don't
1717
    # make all OSes invalid
1718
    good_nodes = [node_name for node_name in rlist
1719
                  if not rlist[node_name].failed]
1720
    for node_name, nr in rlist.iteritems():
1721
      if nr.failed or not nr.data:
1722
        continue
1723
      for os_obj in nr.data:
1724
        if os_obj.name not in all_os:
1725
          # build a list of nodes for this os containing empty lists
1726
          # for each node in node_list
1727
          all_os[os_obj.name] = {}
1728
          for nname in good_nodes:
1729
            all_os[os_obj.name][nname] = []
1730
        all_os[os_obj.name][node_name].append(os_obj)
1731
    return all_os
1732

    
1733
  def Exec(self, feedback_fn):
1734
    """Compute the list of OSes.
1735

1736
    """
1737
    valid_nodes = [node for node in self.cfg.GetOnlineNodeList()]
1738
    node_data = self.rpc.call_os_diagnose(valid_nodes)
1739
    if node_data == False:
1740
      raise errors.OpExecError("Can't gather the list of OSes")
1741
    pol = self._DiagnoseByOS(valid_nodes, node_data)
1742
    output = []
1743
    for os_name, os_data in pol.iteritems():
1744
      row = []
1745
      for field in self.op.output_fields:
1746
        if field == "name":
1747
          val = os_name
1748
        elif field == "valid":
1749
          val = utils.all([osl and osl[0] for osl in os_data.values()])
1750
        elif field == "node_status":
1751
          val = {}
1752
          for node_name, nos_list in os_data.iteritems():
1753
            val[node_name] = [(v.status, v.path) for v in nos_list]
1754
        else:
1755
          raise errors.ParameterError(field)
1756
        row.append(val)
1757
      output.append(row)
1758

    
1759
    return output
1760

    
1761

    
1762
class LURemoveNode(LogicalUnit):
1763
  """Logical unit for removing a node.
1764

1765
  """
1766
  HPATH = "node-remove"
1767
  HTYPE = constants.HTYPE_NODE
1768
  _OP_REQP = ["node_name"]
1769

    
1770
  def BuildHooksEnv(self):
1771
    """Build hooks env.
1772

1773
    This doesn't run on the target node in the pre phase as a failed
1774
    node would then be impossible to remove.
1775

1776
    """
1777
    env = {
1778
      "OP_TARGET": self.op.node_name,
1779
      "NODE_NAME": self.op.node_name,
1780
      }
1781
    all_nodes = self.cfg.GetNodeList()
1782
    all_nodes.remove(self.op.node_name)
1783
    return env, all_nodes, all_nodes
1784

    
1785
  def CheckPrereq(self):
1786
    """Check prerequisites.
1787

1788
    This checks:
1789
     - the node exists in the configuration
1790
     - it does not have primary or secondary instances
1791
     - it's not the master
1792

1793
    Any errors are signalled by raising errors.OpPrereqError.
1794

1795
    """
1796
    node = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.node_name))
1797
    if node is None:
1798
      raise errors.OpPrereqError, ("Node '%s' is unknown." % self.op.node_name)
1799

    
1800
    instance_list = self.cfg.GetInstanceList()
1801

    
1802
    masternode = self.cfg.GetMasterNode()
1803
    if node.name == masternode:
1804
      raise errors.OpPrereqError("Node is the master node,"
1805
                                 " you need to failover first.")
1806

    
1807
    for instance_name in instance_list:
1808
      instance = self.cfg.GetInstanceInfo(instance_name)
1809
      if node.name in instance.all_nodes:
1810
        raise errors.OpPrereqError("Instance %s is still running on the node,"
1811
                                   " please remove first." % instance_name)
1812
    self.op.node_name = node.name
1813
    self.node = node
1814

    
1815
  def Exec(self, feedback_fn):
1816
    """Removes the node from the cluster.
1817

1818
    """
1819
    node = self.node
1820
    logging.info("Stopping the node daemon and removing configs from node %s",
1821
                 node.name)
1822

    
1823
    self.context.RemoveNode(node.name)
1824

    
1825
    self.rpc.call_node_leave_cluster(node.name)
1826

    
1827
    # Promote nodes to master candidate as needed
1828
    _AdjustCandidatePool(self)
1829

    
1830

    
1831
class LUQueryNodes(NoHooksLU):
1832
  """Logical unit for querying nodes.
1833

1834
  """
1835
  _OP_REQP = ["output_fields", "names", "use_locking"]
1836
  REQ_BGL = False
1837
  _FIELDS_DYNAMIC = utils.FieldSet(
1838
    "dtotal", "dfree",
1839
    "mtotal", "mnode", "mfree",
1840
    "bootid",
1841
    "ctotal", "cnodes", "csockets",
1842
    )
1843

    
1844
  _FIELDS_STATIC = utils.FieldSet(
1845
    "name", "pinst_cnt", "sinst_cnt",
1846
    "pinst_list", "sinst_list",
1847
    "pip", "sip", "tags",
1848
    "serial_no",
1849
    "master_candidate",
1850
    "master",
1851
    "offline",
1852
    "drained",
1853
    )
1854

    
1855
  def ExpandNames(self):
1856
    _CheckOutputFields(static=self._FIELDS_STATIC,
1857
                       dynamic=self._FIELDS_DYNAMIC,
1858
                       selected=self.op.output_fields)
1859

    
1860
    self.needed_locks = {}
1861
    self.share_locks[locking.LEVEL_NODE] = 1
1862

    
1863
    if self.op.names:
1864
      self.wanted = _GetWantedNodes(self, self.op.names)
1865
    else:
1866
      self.wanted = locking.ALL_SET
1867

    
1868
    self.do_node_query = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
1869
    self.do_locking = self.do_node_query and self.op.use_locking
1870
    if self.do_locking:
1871
      # if we don't request only static fields, we need to lock the nodes
1872
      self.needed_locks[locking.LEVEL_NODE] = self.wanted
1873

    
1874

    
1875
  def CheckPrereq(self):
1876
    """Check prerequisites.
1877

1878
    """
1879
    # The validation of the node list is done in the _GetWantedNodes,
1880
    # if non empty, and if empty, there's no validation to do
1881
    pass
1882

    
1883
  def Exec(self, feedback_fn):
1884
    """Computes the list of nodes and their attributes.
1885

1886
    """
1887
    all_info = self.cfg.GetAllNodesInfo()
1888
    if self.do_locking:
1889
      nodenames = self.acquired_locks[locking.LEVEL_NODE]
1890
    elif self.wanted != locking.ALL_SET:
1891
      nodenames = self.wanted
1892
      missing = set(nodenames).difference(all_info.keys())
1893
      if missing:
1894
        raise errors.OpExecError(
1895
          "Some nodes were removed before retrieving their data: %s" % missing)
1896
    else:
1897
      nodenames = all_info.keys()
1898

    
1899
    nodenames = utils.NiceSort(nodenames)
1900
    nodelist = [all_info[name] for name in nodenames]
1901

    
1902
    # begin data gathering
1903

    
1904
    if self.do_node_query:
1905
      live_data = {}
1906
      node_data = self.rpc.call_node_info(nodenames, self.cfg.GetVGName(),
1907
                                          self.cfg.GetHypervisorType())
1908
      for name in nodenames:
1909
        nodeinfo = node_data[name]
1910
        if not nodeinfo.failed and nodeinfo.data:
1911
          nodeinfo = nodeinfo.data
1912
          fn = utils.TryConvert
1913
          live_data[name] = {
1914
            "mtotal": fn(int, nodeinfo.get('memory_total', None)),
1915
            "mnode": fn(int, nodeinfo.get('memory_dom0', None)),
1916
            "mfree": fn(int, nodeinfo.get('memory_free', None)),
1917
            "dtotal": fn(int, nodeinfo.get('vg_size', None)),
1918
            "dfree": fn(int, nodeinfo.get('vg_free', None)),
1919
            "ctotal": fn(int, nodeinfo.get('cpu_total', None)),
1920
            "bootid": nodeinfo.get('bootid', None),
1921
            "cnodes": fn(int, nodeinfo.get('cpu_nodes', None)),
1922
            "csockets": fn(int, nodeinfo.get('cpu_sockets', None)),
1923
            }
1924
        else:
1925
          live_data[name] = {}
1926
    else:
1927
      live_data = dict.fromkeys(nodenames, {})
1928

    
1929
    node_to_primary = dict([(name, set()) for name in nodenames])
1930
    node_to_secondary = dict([(name, set()) for name in nodenames])
1931

    
1932
    inst_fields = frozenset(("pinst_cnt", "pinst_list",
1933
                             "sinst_cnt", "sinst_list"))
1934
    if inst_fields & frozenset(self.op.output_fields):
1935
      instancelist = self.cfg.GetInstanceList()
1936

    
1937
      for instance_name in instancelist:
1938
        inst = self.cfg.GetInstanceInfo(instance_name)
1939
        if inst.primary_node in node_to_primary:
1940
          node_to_primary[inst.primary_node].add(inst.name)
1941
        for secnode in inst.secondary_nodes:
1942
          if secnode in node_to_secondary:
1943
            node_to_secondary[secnode].add(inst.name)
1944

    
1945
    master_node = self.cfg.GetMasterNode()
1946

    
1947
    # end data gathering
1948

    
1949
    output = []
1950
    for node in nodelist:
1951
      node_output = []
1952
      for field in self.op.output_fields:
1953
        if field == "name":
1954
          val = node.name
1955
        elif field == "pinst_list":
1956
          val = list(node_to_primary[node.name])
1957
        elif field == "sinst_list":
1958
          val = list(node_to_secondary[node.name])
1959
        elif field == "pinst_cnt":
1960
          val = len(node_to_primary[node.name])
1961
        elif field == "sinst_cnt":
1962
          val = len(node_to_secondary[node.name])
1963
        elif field == "pip":
1964
          val = node.primary_ip
1965
        elif field == "sip":
1966
          val = node.secondary_ip
1967
        elif field == "tags":
1968
          val = list(node.GetTags())
1969
        elif field == "serial_no":
1970
          val = node.serial_no
1971
        elif field == "master_candidate":
1972
          val = node.master_candidate
1973
        elif field == "master":
1974
          val = node.name == master_node
1975
        elif field == "offline":
1976
          val = node.offline
1977
        elif field == "drained":
1978
          val = node.drained
1979
        elif self._FIELDS_DYNAMIC.Matches(field):
1980
          val = live_data[node.name].get(field, None)
1981
        else:
1982
          raise errors.ParameterError(field)
1983
        node_output.append(val)
1984
      output.append(node_output)
1985

    
1986
    return output
1987

    
1988

    
1989
class LUQueryNodeVolumes(NoHooksLU):
1990
  """Logical unit for getting volumes on node(s).
1991

1992
  """
1993
  _OP_REQP = ["nodes", "output_fields"]
1994
  REQ_BGL = False
1995
  _FIELDS_DYNAMIC = utils.FieldSet("phys", "vg", "name", "size", "instance")
1996
  _FIELDS_STATIC = utils.FieldSet("node")
1997

    
1998
  def ExpandNames(self):
1999
    _CheckOutputFields(static=self._FIELDS_STATIC,
2000
                       dynamic=self._FIELDS_DYNAMIC,
2001
                       selected=self.op.output_fields)
2002

    
2003
    self.needed_locks = {}
2004
    self.share_locks[locking.LEVEL_NODE] = 1
2005
    if not self.op.nodes:
2006
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
2007
    else:
2008
      self.needed_locks[locking.LEVEL_NODE] = \
2009
        _GetWantedNodes(self, self.op.nodes)
2010

    
2011
  def CheckPrereq(self):
2012
    """Check prerequisites.
2013

2014
    This checks that the fields required are valid output fields.
2015

2016
    """
2017
    self.nodes = self.acquired_locks[locking.LEVEL_NODE]
2018

    
2019
  def Exec(self, feedback_fn):
2020
    """Computes the list of nodes and their attributes.
2021

2022
    """
2023
    nodenames = self.nodes
2024
    volumes = self.rpc.call_node_volumes(nodenames)
2025

    
2026
    ilist = [self.cfg.GetInstanceInfo(iname) for iname
2027
             in self.cfg.GetInstanceList()]
2028

    
2029
    lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
2030

    
2031
    output = []
2032
    for node in nodenames:
2033
      if node not in volumes or volumes[node].failed or not volumes[node].data:
2034
        continue
2035

    
2036
      node_vols = volumes[node].data[:]
2037
      node_vols.sort(key=lambda vol: vol['dev'])
2038

    
2039
      for vol in node_vols:
2040
        node_output = []
2041
        for field in self.op.output_fields:
2042
          if field == "node":
2043
            val = node
2044
          elif field == "phys":
2045
            val = vol['dev']
2046
          elif field == "vg":
2047
            val = vol['vg']
2048
          elif field == "name":
2049
            val = vol['name']
2050
          elif field == "size":
2051
            val = int(float(vol['size']))
2052
          elif field == "instance":
2053
            for inst in ilist:
2054
              if node not in lv_by_node[inst]:
2055
                continue
2056
              if vol['name'] in lv_by_node[inst][node]:
2057
                val = inst.name
2058
                break
2059
            else:
2060
              val = '-'
2061
          else:
2062
            raise errors.ParameterError(field)
2063
          node_output.append(str(val))
2064

    
2065
        output.append(node_output)
2066

    
2067
    return output
2068

    
2069

    
2070
class LUAddNode(LogicalUnit):
2071
  """Logical unit for adding node to the cluster.
2072

2073
  """
2074
  HPATH = "node-add"
2075
  HTYPE = constants.HTYPE_NODE
2076
  _OP_REQP = ["node_name"]
2077

    
2078
  def BuildHooksEnv(self):
2079
    """Build hooks env.
2080

2081
    This will run on all nodes before, and on all nodes + the new node after.
2082

2083
    """
2084
    env = {
2085
      "OP_TARGET": self.op.node_name,
2086
      "NODE_NAME": self.op.node_name,
2087
      "NODE_PIP": self.op.primary_ip,
2088
      "NODE_SIP": self.op.secondary_ip,
2089
      }
2090
    nodes_0 = self.cfg.GetNodeList()
2091
    nodes_1 = nodes_0 + [self.op.node_name, ]
2092
    return env, nodes_0, nodes_1
2093

    
2094
  def CheckPrereq(self):
2095
    """Check prerequisites.
2096

2097
    This checks:
2098
     - the new node is not already in the config
2099
     - it is resolvable
2100
     - its parameters (single/dual homed) matches the cluster
2101

2102
    Any errors are signalled by raising errors.OpPrereqError.
2103

2104
    """
2105
    node_name = self.op.node_name
2106
    cfg = self.cfg
2107

    
2108
    dns_data = utils.HostInfo(node_name)
2109

    
2110
    node = dns_data.name
2111
    primary_ip = self.op.primary_ip = dns_data.ip
2112
    secondary_ip = getattr(self.op, "secondary_ip", None)
2113
    if secondary_ip is None:
2114
      secondary_ip = primary_ip
2115
    if not utils.IsValidIP(secondary_ip):
2116
      raise errors.OpPrereqError("Invalid secondary IP given")
2117
    self.op.secondary_ip = secondary_ip
2118

    
2119
    node_list = cfg.GetNodeList()
2120
    if not self.op.readd and node in node_list:
2121
      raise errors.OpPrereqError("Node %s is already in the configuration" %
2122
                                 node)
2123
    elif self.op.readd and node not in node_list:
2124
      raise errors.OpPrereqError("Node %s is not in the configuration" % node)
2125

    
2126
    for existing_node_name in node_list:
2127
      existing_node = cfg.GetNodeInfo(existing_node_name)
2128

    
2129
      if self.op.readd and node == existing_node_name:
2130
        if (existing_node.primary_ip != primary_ip or
2131
            existing_node.secondary_ip != secondary_ip):
2132
          raise errors.OpPrereqError("Readded node doesn't have the same IP"
2133
                                     " address configuration as before")
2134
        continue
2135

    
2136
      if (existing_node.primary_ip == primary_ip or
2137
          existing_node.secondary_ip == primary_ip or
2138
          existing_node.primary_ip == secondary_ip or
2139
          existing_node.secondary_ip == secondary_ip):
2140
        raise errors.OpPrereqError("New node ip address(es) conflict with"
2141
                                   " existing node %s" % existing_node.name)
2142

    
2143
    # check that the type of the node (single versus dual homed) is the
2144
    # same as for the master
2145
    myself = cfg.GetNodeInfo(self.cfg.GetMasterNode())
2146
    master_singlehomed = myself.secondary_ip == myself.primary_ip
2147
    newbie_singlehomed = secondary_ip == primary_ip
2148
    if master_singlehomed != newbie_singlehomed:
2149
      if master_singlehomed:
2150
        raise errors.OpPrereqError("The master has no private ip but the"
2151
                                   " new node has one")
2152
      else:
2153
        raise errors.OpPrereqError("The master has a private ip but the"
2154
                                   " new node doesn't have one")
2155

    
2156
    # checks reachablity
2157
    if not utils.TcpPing(primary_ip, constants.DEFAULT_NODED_PORT):
2158
      raise errors.OpPrereqError("Node not reachable by ping")
2159

    
2160
    if not newbie_singlehomed:
2161
      # check reachability from my secondary ip to newbie's secondary ip
2162
      if not utils.TcpPing(secondary_ip, constants.DEFAULT_NODED_PORT,
2163
                           source=myself.secondary_ip):
2164
        raise errors.OpPrereqError("Node secondary ip not reachable by TCP"
2165
                                   " based ping to noded port")
2166

    
2167
    cp_size = self.cfg.GetClusterInfo().candidate_pool_size
2168
    mc_now, _ = self.cfg.GetMasterCandidateStats()
2169
    master_candidate = mc_now < cp_size
2170

    
2171
    self.new_node = objects.Node(name=node,
2172
                                 primary_ip=primary_ip,
2173
                                 secondary_ip=secondary_ip,
2174
                                 master_candidate=master_candidate,
2175
                                 offline=False, drained=False)
2176

    
2177
  def Exec(self, feedback_fn):
2178
    """Adds the new node to the cluster.
2179

2180
    """
2181
    new_node = self.new_node
2182
    node = new_node.name
2183

    
2184
    # check connectivity
2185
    result = self.rpc.call_version([node])[node]
2186
    result.Raise()
2187
    if result.data:
2188
      if constants.PROTOCOL_VERSION == result.data:
2189
        logging.info("Communication to node %s fine, sw version %s match",
2190
                     node, result.data)
2191
      else:
2192
        raise errors.OpExecError("Version mismatch master version %s,"
2193
                                 " node version %s" %
2194
                                 (constants.PROTOCOL_VERSION, result.data))
2195
    else:
2196
      raise errors.OpExecError("Cannot get version from the new node")
2197

    
2198
    # setup ssh on node
2199
    logging.info("Copy ssh key to node %s", node)
2200
    priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
2201
    keyarray = []
2202
    keyfiles = [constants.SSH_HOST_DSA_PRIV, constants.SSH_HOST_DSA_PUB,
2203
                constants.SSH_HOST_RSA_PRIV, constants.SSH_HOST_RSA_PUB,
2204
                priv_key, pub_key]
2205

    
2206
    for i in keyfiles:
2207
      f = open(i, 'r')
2208
      try:
2209
        keyarray.append(f.read())
2210
      finally:
2211
        f.close()
2212

    
2213
    result = self.rpc.call_node_add(node, keyarray[0], keyarray[1],
2214
                                    keyarray[2],
2215
                                    keyarray[3], keyarray[4], keyarray[5])
2216

    
2217
    msg = result.RemoteFailMsg()
2218
    if msg:
2219
      raise errors.OpExecError("Cannot transfer ssh keys to the"
2220
                               " new node: %s" % msg)
2221

    
2222
    # Add node to our /etc/hosts, and add key to known_hosts
2223
    utils.AddHostToEtcHosts(new_node.name)
2224

    
2225
    if new_node.secondary_ip != new_node.primary_ip:
2226
      result = self.rpc.call_node_has_ip_address(new_node.name,
2227
                                                 new_node.secondary_ip)
2228
      if result.failed or not result.data:
2229
        raise errors.OpExecError("Node claims it doesn't have the secondary ip"
2230
                                 " you gave (%s). Please fix and re-run this"
2231
                                 " command." % new_node.secondary_ip)
2232

    
2233
    node_verify_list = [self.cfg.GetMasterNode()]
2234
    node_verify_param = {
2235
      'nodelist': [node],
2236
      # TODO: do a node-net-test as well?
2237
    }
2238

    
2239
    result = self.rpc.call_node_verify(node_verify_list, node_verify_param,
2240
                                       self.cfg.GetClusterName())
2241
    for verifier in node_verify_list:
2242
      if result[verifier].failed or not result[verifier].data:
2243
        raise errors.OpExecError("Cannot communicate with %s's node daemon"
2244
                                 " for remote verification" % verifier)
2245
      if result[verifier].data['nodelist']:
2246
        for failed in result[verifier].data['nodelist']:
2247
          feedback_fn("ssh/hostname verification failed %s -> %s" %
2248
                      (verifier, result[verifier].data['nodelist'][failed]))
2249
        raise errors.OpExecError("ssh/hostname verification failed.")
2250

    
2251
    # Distribute updated /etc/hosts and known_hosts to all nodes,
2252
    # including the node just added
2253
    myself = self.cfg.GetNodeInfo(self.cfg.GetMasterNode())
2254
    dist_nodes = self.cfg.GetNodeList()
2255
    if not self.op.readd:
2256
      dist_nodes.append(node)
2257
    if myself.name in dist_nodes:
2258
      dist_nodes.remove(myself.name)
2259

    
2260
    logging.debug("Copying hosts and known_hosts to all nodes")
2261
    for fname in (constants.ETC_HOSTS, constants.SSH_KNOWN_HOSTS_FILE):
2262
      result = self.rpc.call_upload_file(dist_nodes, fname)
2263
      for to_node, to_result in result.iteritems():
2264
        if to_result.failed or not to_result.data:
2265
          logging.error("Copy of file %s to node %s failed", fname, to_node)
2266

    
2267
    to_copy = []
2268
    enabled_hypervisors = self.cfg.GetClusterInfo().enabled_hypervisors
2269
    if constants.HTS_COPY_VNC_PASSWORD.intersection(enabled_hypervisors):
2270
      to_copy.append(constants.VNC_PASSWORD_FILE)
2271

    
2272
    for fname in to_copy:
2273
      result = self.rpc.call_upload_file([node], fname)
2274
      if result[node].failed or not result[node]:
2275
        logging.error("Could not copy file %s to node %s", fname, node)
2276

    
2277
    if self.op.readd:
2278
      self.context.ReaddNode(new_node)
2279
    else:
2280
      self.context.AddNode(new_node)
2281

    
2282

    
2283
class LUSetNodeParams(LogicalUnit):
2284
  """Modifies the parameters of a node.
2285

2286
  """
2287
  HPATH = "node-modify"
2288
  HTYPE = constants.HTYPE_NODE
2289
  _OP_REQP = ["node_name"]
2290
  REQ_BGL = False
2291

    
2292
  def CheckArguments(self):
2293
    node_name = self.cfg.ExpandNodeName(self.op.node_name)
2294
    if node_name is None:
2295
      raise errors.OpPrereqError("Invalid node name '%s'" % self.op.node_name)
2296
    self.op.node_name = node_name
2297
    _CheckBooleanOpField(self.op, 'master_candidate')
2298
    _CheckBooleanOpField(self.op, 'offline')
2299
    _CheckBooleanOpField(self.op, 'drained')
2300
    all_mods = [self.op.offline, self.op.master_candidate, self.op.drained]
2301
    if all_mods.count(None) == 3:
2302
      raise errors.OpPrereqError("Please pass at least one modification")
2303
    if all_mods.count(True) > 1:
2304
      raise errors.OpPrereqError("Can't set the node into more than one"
2305
                                 " state at the same time")
2306

    
2307
  def ExpandNames(self):
2308
    self.needed_locks = {locking.LEVEL_NODE: self.op.node_name}
2309

    
2310
  def BuildHooksEnv(self):
2311
    """Build hooks env.
2312

2313
    This runs on the master node.
2314

2315
    """
2316
    env = {
2317
      "OP_TARGET": self.op.node_name,
2318
      "MASTER_CANDIDATE": str(self.op.master_candidate),
2319
      "OFFLINE": str(self.op.offline),
2320
      "DRAINED": str(self.op.drained),
2321
      }
2322
    nl = [self.cfg.GetMasterNode(),
2323
          self.op.node_name]
2324
    return env, nl, nl
2325

    
2326
  def CheckPrereq(self):
2327
    """Check prerequisites.
2328

2329
    This only checks the instance list against the existing names.
2330

2331
    """
2332
    node = self.node = self.cfg.GetNodeInfo(self.op.node_name)
2333

    
2334
    if ((self.op.master_candidate == False or self.op.offline == True or
2335
         self.op.drained == True) and node.master_candidate):
2336
      # we will demote the node from master_candidate
2337
      if self.op.node_name == self.cfg.GetMasterNode():
2338
        raise errors.OpPrereqError("The master node has to be a"
2339
                                   " master candidate, online and not drained")
2340
      cp_size = self.cfg.GetClusterInfo().candidate_pool_size
2341
      num_candidates, _ = self.cfg.GetMasterCandidateStats()
2342
      if num_candidates <= cp_size:
2343
        msg = ("Not enough master candidates (desired"
2344
               " %d, new value will be %d)" % (cp_size, num_candidates-1))
2345
        if self.op.force:
2346
          self.LogWarning(msg)
2347
        else:
2348
          raise errors.OpPrereqError(msg)
2349

    
2350
    if (self.op.master_candidate == True and
2351
        ((node.offline and not self.op.offline == False) or
2352
         (node.drained and not self.op.drained == False))):
2353
      raise errors.OpPrereqError("Node '%s' is offline or drained, can't set"
2354
                                 " to master_candidate" % node.name)
2355

    
2356
    return
2357

    
2358
  def Exec(self, feedback_fn):
2359
    """Modifies a node.
2360

2361
    """
2362
    node = self.node
2363

    
2364
    result = []
2365
    changed_mc = False
2366

    
2367
    if self.op.offline is not None:
2368
      node.offline = self.op.offline
2369
      result.append(("offline", str(self.op.offline)))
2370
      if self.op.offline == True:
2371
        if node.master_candidate:
2372
          node.master_candidate = False
2373
          changed_mc = True
2374
          result.append(("master_candidate", "auto-demotion due to offline"))
2375
        if node.drained:
2376
          node.drained = False
2377
          result.append(("drained", "clear drained status due to offline"))
2378

    
2379
    if self.op.master_candidate is not None:
2380
      node.master_candidate = self.op.master_candidate
2381
      changed_mc = True
2382
      result.append(("master_candidate", str(self.op.master_candidate)))
2383
      if self.op.master_candidate == False:
2384
        rrc = self.rpc.call_node_demote_from_mc(node.name)
2385
        msg = rrc.RemoteFailMsg()
2386
        if msg:
2387
          self.LogWarning("Node failed to demote itself: %s" % msg)
2388

    
2389
    if self.op.drained is not None:
2390
      node.drained = self.op.drained
2391
      result.append(("drained", str(self.op.drained)))
2392
      if self.op.drained == True:
2393
        if node.master_candidate:
2394
          node.master_candidate = False
2395
          changed_mc = True
2396
          result.append(("master_candidate", "auto-demotion due to drain"))
2397
        if node.offline:
2398
          node.offline = False
2399
          result.append(("offline", "clear offline status due to drain"))
2400

    
2401
    # this will trigger configuration file update, if needed
2402
    self.cfg.Update(node)
2403
    # this will trigger job queue propagation or cleanup
2404
    if changed_mc:
2405
      self.context.ReaddNode(node)
2406

    
2407
    return result
2408

    
2409

    
2410
class LUQueryClusterInfo(NoHooksLU):
2411
  """Query cluster configuration.
2412

2413
  """
2414
  _OP_REQP = []
2415
  REQ_BGL = False
2416

    
2417
  def ExpandNames(self):
2418
    self.needed_locks = {}
2419

    
2420
  def CheckPrereq(self):
2421
    """No prerequsites needed for this LU.
2422

2423
    """
2424
    pass
2425

    
2426
  def Exec(self, feedback_fn):
2427
    """Return cluster config.
2428

2429
    """
2430
    cluster = self.cfg.GetClusterInfo()
2431
    result = {
2432
      "software_version": constants.RELEASE_VERSION,
2433
      "protocol_version": constants.PROTOCOL_VERSION,
2434
      "config_version": constants.CONFIG_VERSION,
2435
      "os_api_version": constants.OS_API_VERSION,
2436
      "export_version": constants.EXPORT_VERSION,
2437
      "architecture": (platform.architecture()[0], platform.machine()),
2438
      "name": cluster.cluster_name,
2439
      "master": cluster.master_node,
2440
      "default_hypervisor": cluster.default_hypervisor,
2441
      "enabled_hypervisors": cluster.enabled_hypervisors,
2442
      "hvparams": dict([(hypervisor, cluster.hvparams[hypervisor])
2443
                        for hypervisor in cluster.enabled_hypervisors]),
2444
      "beparams": cluster.beparams,
2445
      "candidate_pool_size": cluster.candidate_pool_size,
2446
      "default_bridge": cluster.default_bridge,
2447
      "master_netdev": cluster.master_netdev,
2448
      "volume_group_name": cluster.volume_group_name,
2449
      "file_storage_dir": cluster.file_storage_dir,
2450
      }
2451

    
2452
    return result
2453

    
2454

    
2455
class LUQueryConfigValues(NoHooksLU):
2456
  """Return configuration values.
2457

2458
  """
2459
  _OP_REQP = []
2460
  REQ_BGL = False
2461
  _FIELDS_DYNAMIC = utils.FieldSet()
2462
  _FIELDS_STATIC = utils.FieldSet("cluster_name", "master_node", "drain_flag")
2463

    
2464
  def ExpandNames(self):
2465
    self.needed_locks = {}
2466

    
2467
    _CheckOutputFields(static=self._FIELDS_STATIC,
2468
                       dynamic=self._FIELDS_DYNAMIC,
2469
                       selected=self.op.output_fields)
2470

    
2471
  def CheckPrereq(self):
2472
    """No prerequisites.
2473

2474
    """
2475
    pass
2476

    
2477
  def Exec(self, feedback_fn):
2478
    """Dump a representation of the cluster config to the standard output.
2479

2480
    """
2481
    values = []
2482
    for field in self.op.output_fields:
2483
      if field == "cluster_name":
2484
        entry = self.cfg.GetClusterName()
2485
      elif field == "master_node":
2486
        entry = self.cfg.GetMasterNode()
2487
      elif field == "drain_flag":
2488
        entry = os.path.exists(constants.JOB_QUEUE_DRAIN_FILE)
2489
      else:
2490
        raise errors.ParameterError(field)
2491
      values.append(entry)
2492
    return values
2493

    
2494

    
2495
class LUActivateInstanceDisks(NoHooksLU):
2496
  """Bring up an instance's disks.
2497

2498
  """
2499
  _OP_REQP = ["instance_name"]
2500
  REQ_BGL = False
2501

    
2502
  def ExpandNames(self):
2503
    self._ExpandAndLockInstance()
2504
    self.needed_locks[locking.LEVEL_NODE] = []
2505
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2506

    
2507
  def DeclareLocks(self, level):
2508
    if level == locking.LEVEL_NODE:
2509
      self._LockInstancesNodes()
2510

    
2511
  def CheckPrereq(self):
2512
    """Check prerequisites.
2513

2514
    This checks that the instance is in the cluster.
2515

2516
    """
2517
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2518
    assert self.instance is not None, \
2519
      "Cannot retrieve locked instance %s" % self.op.instance_name
2520
    _CheckNodeOnline(self, self.instance.primary_node)
2521

    
2522
  def Exec(self, feedback_fn):
2523
    """Activate the disks.
2524

2525
    """
2526
    disks_ok, disks_info = _AssembleInstanceDisks(self, self.instance)
2527
    if not disks_ok:
2528
      raise errors.OpExecError("Cannot activate block devices")
2529

    
2530
    return disks_info
2531

    
2532

    
2533
def _AssembleInstanceDisks(lu, instance, ignore_secondaries=False):
2534
  """Prepare the block devices for an instance.
2535

2536
  This sets up the block devices on all nodes.
2537

2538
  @type lu: L{LogicalUnit}
2539
  @param lu: the logical unit on whose behalf we execute
2540
  @type instance: L{objects.Instance}
2541
  @param instance: the instance for whose disks we assemble
2542
  @type ignore_secondaries: boolean
2543
  @param ignore_secondaries: if true, errors on secondary nodes
2544
      won't result in an error return from the function
2545
  @return: False if the operation failed, otherwise a list of
2546
      (host, instance_visible_name, node_visible_name)
2547
      with the mapping from node devices to instance devices
2548

2549
  """
2550
  device_info = []
2551
  disks_ok = True
2552
  iname = instance.name
2553
  # With the two passes mechanism we try to reduce the window of
2554
  # opportunity for the race condition of switching DRBD to primary
2555
  # before handshaking occured, but we do not eliminate it
2556

    
2557
  # The proper fix would be to wait (with some limits) until the
2558
  # connection has been made and drbd transitions from WFConnection
2559
  # into any other network-connected state (Connected, SyncTarget,
2560
  # SyncSource, etc.)
2561

    
2562
  # 1st pass, assemble on all nodes in secondary mode
2563
  for inst_disk in instance.disks:
2564
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
2565
      lu.cfg.SetDiskID(node_disk, node)
2566
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, False)
2567
      msg = result.RemoteFailMsg()
2568
      if msg:
2569
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
2570
                           " (is_primary=False, pass=1): %s",
2571
                           inst_disk.iv_name, node, msg)
2572
        if not ignore_secondaries:
2573
          disks_ok = False
2574

    
2575
  # FIXME: race condition on drbd migration to primary
2576

    
2577
  # 2nd pass, do only the primary node
2578
  for inst_disk in instance.disks:
2579
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
2580
      if node != instance.primary_node:
2581
        continue
2582
      lu.cfg.SetDiskID(node_disk, node)
2583
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, True)
2584
      msg = result.RemoteFailMsg()
2585
      if msg:
2586
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
2587
                           " (is_primary=True, pass=2): %s",
2588
                           inst_disk.iv_name, node, msg)
2589
        disks_ok = False
2590
    device_info.append((instance.primary_node, inst_disk.iv_name,
2591
                        result.payload))
2592

    
2593
  # leave the disks configured for the primary node
2594
  # this is a workaround that would be fixed better by
2595
  # improving the logical/physical id handling
2596
  for disk in instance.disks:
2597
    lu.cfg.SetDiskID(disk, instance.primary_node)
2598

    
2599
  return disks_ok, device_info
2600

    
2601

    
2602
def _StartInstanceDisks(lu, instance, force):
2603
  """Start the disks of an instance.
2604

2605
  """
2606
  disks_ok, dummy = _AssembleInstanceDisks(lu, instance,
2607
                                           ignore_secondaries=force)
2608
  if not disks_ok:
2609
    _ShutdownInstanceDisks(lu, instance)
2610
    if force is not None and not force:
2611
      lu.proc.LogWarning("", hint="If the message above refers to a"
2612
                         " secondary node,"
2613
                         " you can retry the operation using '--force'.")
2614
    raise errors.OpExecError("Disk consistency error")
2615

    
2616

    
2617
class LUDeactivateInstanceDisks(NoHooksLU):
2618
  """Shutdown an instance's disks.
2619

2620
  """
2621
  _OP_REQP = ["instance_name"]
2622
  REQ_BGL = False
2623

    
2624
  def ExpandNames(self):
2625
    self._ExpandAndLockInstance()
2626
    self.needed_locks[locking.LEVEL_NODE] = []
2627
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2628

    
2629
  def DeclareLocks(self, level):
2630
    if level == locking.LEVEL_NODE:
2631
      self._LockInstancesNodes()
2632

    
2633
  def CheckPrereq(self):
2634
    """Check prerequisites.
2635

2636
    This checks that the instance is in the cluster.
2637

2638
    """
2639
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2640
    assert self.instance is not None, \
2641
      "Cannot retrieve locked instance %s" % self.op.instance_name
2642

    
2643
  def Exec(self, feedback_fn):
2644
    """Deactivate the disks
2645

2646
    """
2647
    instance = self.instance
2648
    _SafeShutdownInstanceDisks(self, instance)
2649

    
2650

    
2651
def _SafeShutdownInstanceDisks(lu, instance):
2652
  """Shutdown block devices of an instance.
2653

2654
  This function checks if an instance is running, before calling
2655
  _ShutdownInstanceDisks.
2656

2657
  """
2658
  ins_l = lu.rpc.call_instance_list([instance.primary_node],
2659
                                      [instance.hypervisor])
2660
  ins_l = ins_l[instance.primary_node]
2661
  if ins_l.failed or not isinstance(ins_l.data, list):
2662
    raise errors.OpExecError("Can't contact node '%s'" %
2663
                             instance.primary_node)
2664

    
2665
  if instance.name in ins_l.data:
2666
    raise errors.OpExecError("Instance is running, can't shutdown"
2667
                             " block devices.")
2668

    
2669
  _ShutdownInstanceDisks(lu, instance)
2670

    
2671

    
2672
def _ShutdownInstanceDisks(lu, instance, ignore_primary=False):
2673
  """Shutdown block devices of an instance.
2674

2675
  This does the shutdown on all nodes of the instance.
2676

2677
  If the ignore_primary is false, errors on the primary node are
2678
  ignored.
2679

2680
  """
2681
  all_result = True
2682
  for disk in instance.disks:
2683
    for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
2684
      lu.cfg.SetDiskID(top_disk, node)
2685
      result = lu.rpc.call_blockdev_shutdown(node, top_disk)
2686
      msg = result.RemoteFailMsg()
2687
      if msg:
2688
        lu.LogWarning("Could not shutdown block device %s on node %s: %s",
2689
                      disk.iv_name, node, msg)
2690
        if not ignore_primary or node != instance.primary_node:
2691
          all_result = False
2692
  return all_result
2693

    
2694

    
2695
def _CheckNodeFreeMemory(lu, node, reason, requested, hypervisor_name):
2696
  """Checks if a node has enough free memory.
2697

2698
  This function check if a given node has the needed amount of free
2699
  memory. In case the node has less memory or we cannot get the
2700
  information from the node, this function raise an OpPrereqError
2701
  exception.
2702

2703
  @type lu: C{LogicalUnit}
2704
  @param lu: a logical unit from which we get configuration data
2705
  @type node: C{str}
2706
  @param node: the node to check
2707
  @type reason: C{str}
2708
  @param reason: string to use in the error message
2709
  @type requested: C{int}
2710
  @param requested: the amount of memory in MiB to check for
2711
  @type hypervisor_name: C{str}
2712
  @param hypervisor_name: the hypervisor to ask for memory stats
2713
  @raise errors.OpPrereqError: if the node doesn't have enough memory, or
2714
      we cannot check the node
2715

2716
  """
2717
  nodeinfo = lu.rpc.call_node_info([node], lu.cfg.GetVGName(), hypervisor_name)
2718
  nodeinfo[node].Raise()
2719
  free_mem = nodeinfo[node].data.get('memory_free')
2720
  if not isinstance(free_mem, int):
2721
    raise errors.OpPrereqError("Can't compute free memory on node %s, result"
2722
                             " was '%s'" % (node, free_mem))
2723
  if requested > free_mem:
2724
    raise errors.OpPrereqError("Not enough memory on node %s for %s:"
2725
                             " needed %s MiB, available %s MiB" %
2726
                             (node, reason, requested, free_mem))
2727

    
2728

    
2729
class LUStartupInstance(LogicalUnit):
2730
  """Starts an instance.
2731

2732
  """
2733
  HPATH = "instance-start"
2734
  HTYPE = constants.HTYPE_INSTANCE
2735
  _OP_REQP = ["instance_name", "force"]
2736
  REQ_BGL = False
2737

    
2738
  def ExpandNames(self):
2739
    self._ExpandAndLockInstance()
2740

    
2741
  def BuildHooksEnv(self):
2742
    """Build hooks env.
2743

2744
    This runs on master, primary and secondary nodes of the instance.
2745

2746
    """
2747
    env = {
2748
      "FORCE": self.op.force,
2749
      }
2750
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
2751
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2752
    return env, nl, nl
2753

    
2754
  def CheckPrereq(self):
2755
    """Check prerequisites.
2756

2757
    This checks that the instance is in the cluster.
2758

2759
    """
2760
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2761
    assert self.instance is not None, \
2762
      "Cannot retrieve locked instance %s" % self.op.instance_name
2763

    
2764
    # extra beparams
2765
    self.beparams = getattr(self.op, "beparams", {})
2766
    if self.beparams:
2767
      if not isinstance(self.beparams, dict):
2768
        raise errors.OpPrereqError("Invalid beparams passed: %s, expected"
2769
                                   " dict" % (type(self.beparams), ))
2770
      # fill the beparams dict
2771
      utils.ForceDictType(self.beparams, constants.BES_PARAMETER_TYPES)
2772
      self.op.beparams = self.beparams
2773

    
2774
    # extra hvparams
2775
    self.hvparams = getattr(self.op, "hvparams", {})
2776
    if self.hvparams:
2777
      if not isinstance(self.hvparams, dict):
2778
        raise errors.OpPrereqError("Invalid hvparams passed: %s, expected"
2779
                                   " dict" % (type(self.hvparams), ))
2780

    
2781
      # check hypervisor parameter syntax (locally)
2782
      cluster = self.cfg.GetClusterInfo()
2783
      utils.ForceDictType(self.hvparams, constants.HVS_PARAMETER_TYPES)
2784
      filled_hvp = cluster.FillDict(cluster.hvparams[instance.hypervisor],
2785
                                    instance.hvparams)
2786
      filled_hvp.update(self.hvparams)
2787
      hv_type = hypervisor.GetHypervisor(instance.hypervisor)
2788
      hv_type.CheckParameterSyntax(filled_hvp)
2789
      _CheckHVParams(self, instance.all_nodes, instance.hypervisor, filled_hvp)
2790
      self.op.hvparams = self.hvparams
2791

    
2792
    _CheckNodeOnline(self, instance.primary_node)
2793

    
2794
    bep = self.cfg.GetClusterInfo().FillBE(instance)
2795
    # check bridges existance
2796
    _CheckInstanceBridgesExist(self, instance)
2797

    
2798
    remote_info = self.rpc.call_instance_info(instance.primary_node,
2799
                                              instance.name,
2800
                                              instance.hypervisor)
2801
    remote_info.Raise()
2802
    if not remote_info.data:
2803
      _CheckNodeFreeMemory(self, instance.primary_node,
2804
                           "starting instance %s" % instance.name,
2805
                           bep[constants.BE_MEMORY], instance.hypervisor)
2806

    
2807
  def Exec(self, feedback_fn):
2808
    """Start the instance.
2809

2810
    """
2811
    instance = self.instance
2812
    force = self.op.force
2813

    
2814
    self.cfg.MarkInstanceUp(instance.name)
2815

    
2816
    node_current = instance.primary_node
2817

    
2818
    _StartInstanceDisks(self, instance, force)
2819

    
2820
    result = self.rpc.call_instance_start(node_current, instance,
2821
                                          self.hvparams, self.beparams)
2822
    msg = result.RemoteFailMsg()
2823
    if msg:
2824
      _ShutdownInstanceDisks(self, instance)
2825
      raise errors.OpExecError("Could not start instance: %s" % msg)
2826

    
2827

    
2828
class LURebootInstance(LogicalUnit):
2829
  """Reboot an instance.
2830

2831
  """
2832
  HPATH = "instance-reboot"
2833
  HTYPE = constants.HTYPE_INSTANCE
2834
  _OP_REQP = ["instance_name", "ignore_secondaries", "reboot_type"]
2835
  REQ_BGL = False
2836

    
2837
  def ExpandNames(self):
2838
    if self.op.reboot_type not in [constants.INSTANCE_REBOOT_SOFT,
2839
                                   constants.INSTANCE_REBOOT_HARD,
2840
                                   constants.INSTANCE_REBOOT_FULL]:
2841
      raise errors.ParameterError("reboot type not in [%s, %s, %s]" %
2842
                                  (constants.INSTANCE_REBOOT_SOFT,
2843
                                   constants.INSTANCE_REBOOT_HARD,
2844
                                   constants.INSTANCE_REBOOT_FULL))
2845
    self._ExpandAndLockInstance()
2846

    
2847
  def BuildHooksEnv(self):
2848
    """Build hooks env.
2849

2850
    This runs on master, primary and secondary nodes of the instance.
2851

2852
    """
2853
    env = {
2854
      "IGNORE_SECONDARIES": self.op.ignore_secondaries,
2855
      "REBOOT_TYPE": self.op.reboot_type,
2856
      }
2857
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
2858
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2859
    return env, nl, nl
2860

    
2861
  def CheckPrereq(self):
2862
    """Check prerequisites.
2863

2864
    This checks that the instance is in the cluster.
2865

2866
    """
2867
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2868
    assert self.instance is not None, \
2869
      "Cannot retrieve locked instance %s" % self.op.instance_name
2870

    
2871
    _CheckNodeOnline(self, instance.primary_node)
2872

    
2873
    # check bridges existance
2874
    _CheckInstanceBridgesExist(self, instance)
2875

    
2876
  def Exec(self, feedback_fn):
2877
    """Reboot the instance.
2878

2879
    """
2880
    instance = self.instance
2881
    ignore_secondaries = self.op.ignore_secondaries
2882
    reboot_type = self.op.reboot_type
2883

    
2884
    node_current = instance.primary_node
2885

    
2886
    if reboot_type in [constants.INSTANCE_REBOOT_SOFT,
2887
                       constants.INSTANCE_REBOOT_HARD]:
2888
      for disk in instance.disks:
2889
        self.cfg.SetDiskID(disk, node_current)
2890
      result = self.rpc.call_instance_reboot(node_current, instance,
2891
                                             reboot_type)
2892
      msg = result.RemoteFailMsg()
2893
      if msg:
2894
        raise errors.OpExecError("Could not reboot instance: %s" % msg)
2895
    else:
2896
      result = self.rpc.call_instance_shutdown(node_current, instance)
2897
      msg = result.RemoteFailMsg()
2898
      if msg:
2899
        raise errors.OpExecError("Could not shutdown instance for"
2900
                                 " full reboot: %s" % msg)
2901
      _ShutdownInstanceDisks(self, instance)
2902
      _StartInstanceDisks(self, instance, ignore_secondaries)
2903
      result = self.rpc.call_instance_start(node_current, instance, None, None)
2904
      msg = result.RemoteFailMsg()
2905
      if msg:
2906
        _ShutdownInstanceDisks(self, instance)
2907
        raise errors.OpExecError("Could not start instance for"
2908
                                 " full reboot: %s" % msg)
2909

    
2910
    self.cfg.MarkInstanceUp(instance.name)
2911

    
2912

    
2913
class LUShutdownInstance(LogicalUnit):
2914
  """Shutdown an instance.
2915

2916
  """
2917
  HPATH = "instance-stop"
2918
  HTYPE = constants.HTYPE_INSTANCE
2919
  _OP_REQP = ["instance_name"]
2920
  REQ_BGL = False
2921

    
2922
  def ExpandNames(self):
2923
    self._ExpandAndLockInstance()
2924

    
2925
  def BuildHooksEnv(self):
2926
    """Build hooks env.
2927

2928
    This runs on master, primary and secondary nodes of the instance.
2929

2930
    """
2931
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2932
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2933
    return env, nl, nl
2934

    
2935
  def CheckPrereq(self):
2936
    """Check prerequisites.
2937

2938
    This checks that the instance is in the cluster.
2939

2940
    """
2941
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2942
    assert self.instance is not None, \
2943
      "Cannot retrieve locked instance %s" % self.op.instance_name
2944
    _CheckNodeOnline(self, self.instance.primary_node)
2945

    
2946
  def Exec(self, feedback_fn):
2947
    """Shutdown the instance.
2948

2949
    """
2950
    instance = self.instance
2951
    node_current = instance.primary_node
2952
    self.cfg.MarkInstanceDown(instance.name)
2953
    result = self.rpc.call_instance_shutdown(node_current, instance)
2954
    msg = result.RemoteFailMsg()
2955
    if msg:
2956
      self.proc.LogWarning("Could not shutdown instance: %s" % msg)
2957

    
2958
    _ShutdownInstanceDisks(self, instance)
2959

    
2960

    
2961
class LUReinstallInstance(LogicalUnit):
2962
  """Reinstall an instance.
2963

2964
  """
2965
  HPATH = "instance-reinstall"
2966
  HTYPE = constants.HTYPE_INSTANCE
2967
  _OP_REQP = ["instance_name"]
2968
  REQ_BGL = False
2969

    
2970
  def ExpandNames(self):
2971
    self._ExpandAndLockInstance()
2972

    
2973
  def BuildHooksEnv(self):
2974
    """Build hooks env.
2975

2976
    This runs on master, primary and secondary nodes of the instance.
2977

2978
    """
2979
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2980
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2981
    return env, nl, nl
2982

    
2983
  def CheckPrereq(self):
2984
    """Check prerequisites.
2985

2986
    This checks that the instance is in the cluster and is not running.
2987

2988
    """
2989
    instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2990
    assert instance is not None, \
2991
      "Cannot retrieve locked instance %s" % self.op.instance_name
2992
    _CheckNodeOnline(self, instance.primary_node)
2993

    
2994
    if instance.disk_template == constants.DT_DISKLESS:
2995
      raise errors.OpPrereqError("Instance '%s' has no disks" %
2996
                                 self.op.instance_name)
2997
    if instance.admin_up:
2998
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2999
                                 self.op.instance_name)
3000
    remote_info = self.rpc.call_instance_info(instance.primary_node,
3001
                                              instance.name,
3002
                                              instance.hypervisor)
3003
    remote_info.Raise()
3004
    if remote_info.data:
3005
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
3006
                                 (self.op.instance_name,
3007
                                  instance.primary_node))
3008

    
3009
    self.op.os_type = getattr(self.op, "os_type", None)
3010
    if self.op.os_type is not None:
3011
      # OS verification
3012
      pnode = self.cfg.GetNodeInfo(
3013
        self.cfg.ExpandNodeName(instance.primary_node))
3014
      if pnode is None:
3015
        raise errors.OpPrereqError("Primary node '%s' is unknown" %
3016
                                   self.op.pnode)
3017
      result = self.rpc.call_os_get(pnode.name, self.op.os_type)
3018
      result.Raise()
3019
      if not isinstance(result.data, objects.OS):
3020
        raise errors.OpPrereqError("OS '%s' not in supported OS list for"
3021
                                   " primary node"  % self.op.os_type)
3022

    
3023
    self.instance = instance
3024

    
3025
  def Exec(self, feedback_fn):
3026
    """Reinstall the instance.
3027

3028
    """
3029
    inst = self.instance
3030

    
3031
    if self.op.os_type is not None:
3032
      feedback_fn("Changing OS to '%s'..." % self.op.os_type)
3033
      inst.os = self.op.os_type
3034
      self.cfg.Update(inst)
3035

    
3036
    _StartInstanceDisks(self, inst, None)
3037
    try:
3038
      feedback_fn("Running the instance OS create scripts...")
3039
      result = self.rpc.call_instance_os_add(inst.primary_node, inst)
3040
      msg = result.RemoteFailMsg()
3041
      if msg:
3042
        raise errors.OpExecError("Could not install OS for instance %s"
3043
                                 " on node %s: %s" %
3044
                                 (inst.name, inst.primary_node, msg))
3045
    finally:
3046
      _ShutdownInstanceDisks(self, inst)
3047

    
3048

    
3049
class LURenameInstance(LogicalUnit):
3050
  """Rename an instance.
3051

3052
  """
3053
  HPATH = "instance-rename"
3054
  HTYPE = constants.HTYPE_INSTANCE
3055
  _OP_REQP = ["instance_name", "new_name"]
3056

    
3057
  def BuildHooksEnv(self):
3058
    """Build hooks env.
3059

3060
    This runs on master, primary and secondary nodes of the instance.
3061

3062
    """
3063
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3064
    env["INSTANCE_NEW_NAME"] = self.op.new_name
3065
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3066
    return env, nl, nl
3067

    
3068
  def CheckPrereq(self):
3069
    """Check prerequisites.
3070

3071
    This checks that the instance is in the cluster and is not running.
3072

3073
    """
3074
    instance = self.cfg.GetInstanceInfo(
3075
      self.cfg.ExpandInstanceName(self.op.instance_name))
3076
    if instance is None:
3077
      raise errors.OpPrereqError("Instance '%s' not known" %
3078
                                 self.op.instance_name)
3079
    _CheckNodeOnline(self, instance.primary_node)
3080

    
3081
    if instance.admin_up:
3082
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
3083
                                 self.op.instance_name)
3084
    remote_info = self.rpc.call_instance_info(instance.primary_node,
3085
                                              instance.name,
3086
                                              instance.hypervisor)
3087
    remote_info.Raise()
3088
    if remote_info.data:
3089
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
3090
                                 (self.op.instance_name,
3091
                                  instance.primary_node))
3092
    self.instance = instance
3093

    
3094
    # new name verification
3095
    name_info = utils.HostInfo(self.op.new_name)
3096

    
3097
    self.op.new_name = new_name = name_info.name
3098
    instance_list = self.cfg.GetInstanceList()
3099
    if new_name in instance_list:
3100
      raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
3101
                                 new_name)
3102

    
3103
    if not getattr(self.op, "ignore_ip", False):
3104
      if utils.TcpPing(name_info.ip, constants.DEFAULT_NODED_PORT):
3105
        raise errors.OpPrereqError("IP %s of instance %s already in use" %
3106
                                   (name_info.ip, new_name))
3107

    
3108

    
3109
  def Exec(self, feedback_fn):
3110
    """Reinstall the instance.
3111

3112
    """
3113
    inst = self.instance
3114
    old_name = inst.name
3115

    
3116
    if inst.disk_template == constants.DT_FILE:
3117
      old_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
3118

    
3119
    self.cfg.RenameInstance(inst.name, self.op.new_name)
3120
    # Change the instance lock. This is definitely safe while we hold the BGL
3121
    self.context.glm.remove(locking.LEVEL_INSTANCE, old_name)
3122
    self.context.glm.add(locking.LEVEL_INSTANCE, self.op.new_name)
3123

    
3124
    # re-read the instance from the configuration after rename
3125
    inst = self.cfg.GetInstanceInfo(self.op.new_name)
3126

    
3127
    if inst.disk_template == constants.DT_FILE:
3128
      new_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
3129
      result = self.rpc.call_file_storage_dir_rename(inst.primary_node,
3130
                                                     old_file_storage_dir,
3131
                                                     new_file_storage_dir)
3132
      result.Raise()
3133
      if not result.data:
3134
        raise errors.OpExecError("Could not connect to node '%s' to rename"
3135
                                 " directory '%s' to '%s' (but the instance"
3136
                                 " has been renamed in Ganeti)" % (
3137
                                 inst.primary_node, old_file_storage_dir,
3138
                                 new_file_storage_dir))
3139

    
3140
      if not result.data[0]:
3141
        raise errors.OpExecError("Could not rename directory '%s' to '%s'"
3142
                                 " (but the instance has been renamed in"
3143
                                 " Ganeti)" % (old_file_storage_dir,
3144
                                               new_file_storage_dir))
3145

    
3146
    _StartInstanceDisks(self, inst, None)
3147
    try:
3148
      result = self.rpc.call_instance_run_rename(inst.primary_node, inst,
3149
                                                 old_name)
3150
      msg = result.RemoteFailMsg()
3151
      if msg:
3152
        msg = ("Could not run OS rename script for instance %s on node %s"
3153
               " (but the instance has been renamed in Ganeti): %s" %
3154
               (inst.name, inst.primary_node, msg))
3155
        self.proc.LogWarning(msg)
3156
    finally:
3157
      _ShutdownInstanceDisks(self, inst)
3158

    
3159

    
3160
class LURemoveInstance(LogicalUnit):
3161
  """Remove an instance.
3162

3163
  """
3164
  HPATH = "instance-remove"
3165
  HTYPE = constants.HTYPE_INSTANCE
3166
  _OP_REQP = ["instance_name", "ignore_failures"]
3167
  REQ_BGL = False
3168

    
3169
  def ExpandNames(self):
3170
    self._ExpandAndLockInstance()
3171
    self.needed_locks[locking.LEVEL_NODE] = []
3172
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3173

    
3174
  def DeclareLocks(self, level):
3175
    if level == locking.LEVEL_NODE:
3176
      self._LockInstancesNodes()
3177

    
3178
  def BuildHooksEnv(self):
3179
    """Build hooks env.
3180

3181
    This runs on master, primary and secondary nodes of the instance.
3182

3183
    """
3184
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3185
    nl = [self.cfg.GetMasterNode()]
3186
    return env, nl, nl
3187

    
3188
  def CheckPrereq(self):
3189
    """Check prerequisites.
3190

3191
    This checks that the instance is in the cluster.
3192

3193
    """
3194
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3195
    assert self.instance is not None, \
3196
      "Cannot retrieve locked instance %s" % self.op.instance_name
3197

    
3198
  def Exec(self, feedback_fn):
3199
    """Remove the instance.
3200

3201
    """
3202
    instance = self.instance
3203
    logging.info("Shutting down instance %s on node %s",
3204
                 instance.name, instance.primary_node)
3205

    
3206
    result = self.rpc.call_instance_shutdown(instance.primary_node, instance)
3207
    msg = result.RemoteFailMsg()
3208
    if msg:
3209
      if self.op.ignore_failures:
3210
        feedback_fn("Warning: can't shutdown instance: %s" % msg)
3211
      else:
3212
        raise errors.OpExecError("Could not shutdown instance %s on"
3213
                                 " node %s: %s" %
3214
                                 (instance.name, instance.primary_node, msg))
3215

    
3216
    logging.info("Removing block devices for instance %s", instance.name)
3217

    
3218
    if not _RemoveDisks(self, instance):
3219
      if self.op.ignore_failures:
3220
        feedback_fn("Warning: can't remove instance's disks")
3221
      else:
3222
        raise errors.OpExecError("Can't remove instance's disks")
3223

    
3224
    logging.info("Removing instance %s out of cluster config", instance.name)
3225

    
3226
    self.cfg.RemoveInstance(instance.name)
3227
    self.remove_locks[locking.LEVEL_INSTANCE] = instance.name
3228

    
3229

    
3230
class LUQueryInstances(NoHooksLU):
3231
  """Logical unit for querying instances.
3232

3233
  """
3234
  _OP_REQP = ["output_fields", "names", "use_locking"]
3235
  REQ_BGL = False
3236
  _FIELDS_STATIC = utils.FieldSet(*["name", "os", "pnode", "snodes",
3237
                                    "admin_state",
3238
                                    "disk_template", "ip", "mac", "bridge",
3239
                                    "sda_size", "sdb_size", "vcpus", "tags",
3240
                                    "network_port", "beparams",
3241
                                    r"(disk)\.(size)/([0-9]+)",
3242
                                    r"(disk)\.(sizes)", "disk_usage",
3243
                                    r"(nic)\.(mac|ip|bridge)/([0-9]+)",
3244
                                    r"(nic)\.(macs|ips|bridges)",
3245
                                    r"(disk|nic)\.(count)",
3246
                                    "serial_no", "hypervisor", "hvparams",] +
3247
                                  ["hv/%s" % name
3248
                                   for name in constants.HVS_PARAMETERS] +
3249
                                  ["be/%s" % name
3250
                                   for name in constants.BES_PARAMETERS])
3251
  _FIELDS_DYNAMIC = utils.FieldSet("oper_state", "oper_ram", "status")
3252

    
3253

    
3254
  def ExpandNames(self):
3255
    _CheckOutputFields(static=self._FIELDS_STATIC,
3256
                       dynamic=self._FIELDS_DYNAMIC,
3257
                       selected=self.op.output_fields)
3258

    
3259
    self.needed_locks = {}
3260
    self.share_locks[locking.LEVEL_INSTANCE] = 1
3261
    self.share_locks[locking.LEVEL_NODE] = 1
3262

    
3263
    if self.op.names:
3264
      self.wanted = _GetWantedInstances(self, self.op.names)
3265
    else:
3266
      self.wanted = locking.ALL_SET
3267

    
3268
    self.do_node_query = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
3269
    self.do_locking = self.do_node_query and self.op.use_locking
3270
    if self.do_locking:
3271
      self.needed_locks[locking.LEVEL_INSTANCE] = self.wanted
3272
      self.needed_locks[locking.LEVEL_NODE] = []
3273
      self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3274

    
3275
  def DeclareLocks(self, level):
3276
    if level == locking.LEVEL_NODE and self.do_locking:
3277
      self._LockInstancesNodes()
3278

    
3279
  def CheckPrereq(self):
3280
    """Check prerequisites.
3281

3282
    """
3283
    pass
3284

    
3285
  def Exec(self, feedback_fn):
3286
    """Computes the list of nodes and their attributes.
3287

3288
    """
3289
    all_info = self.cfg.GetAllInstancesInfo()
3290
    if self.wanted == locking.ALL_SET:
3291
      # caller didn't specify instance names, so ordering is not important
3292
      if self.do_locking:
3293
        instance_names = self.acquired_locks[locking.LEVEL_INSTANCE]
3294
      else:
3295
        instance_names = all_info.keys()
3296
      instance_names = utils.NiceSort(instance_names)
3297
    else:
3298
      # caller did specify names, so we must keep the ordering
3299
      if self.do_locking:
3300
        tgt_set = self.acquired_locks[locking.LEVEL_INSTANCE]
3301
      else:
3302
        tgt_set = all_info.keys()
3303
      missing = set(self.wanted).difference(tgt_set)
3304
      if missing:
3305
        raise errors.OpExecError("Some instances were removed before"
3306
                                 " retrieving their data: %s" % missing)
3307
      instance_names = self.wanted
3308

    
3309
    instance_list = [all_info[iname] for iname in instance_names]
3310

    
3311
    # begin data gathering
3312

    
3313
    nodes = frozenset([inst.primary_node for inst in instance_list])
3314
    hv_list = list(set([inst.hypervisor for inst in instance_list]))
3315

    
3316
    bad_nodes = []
3317
    off_nodes = []
3318
    if self.do_node_query:
3319
      live_data = {}
3320
      node_data = self.rpc.call_all_instances_info(nodes, hv_list)
3321
      for name in nodes:
3322
        result = node_data[name]
3323
        if result.offline:
3324
          # offline nodes will be in both lists
3325
          off_nodes.append(name)
3326
        if result.failed:
3327
          bad_nodes.append(name)
3328
        else:
3329
          if result.data:
3330
            live_data.update(result.data)
3331
            # else no instance is alive
3332
    else:
3333
      live_data = dict([(name, {}) for name in instance_names])
3334

    
3335
    # end data gathering
3336

    
3337
    HVPREFIX = "hv/"
3338
    BEPREFIX = "be/"
3339
    output = []
3340
    for instance in instance_list:
3341
      iout = []
3342
      i_hv = self.cfg.GetClusterInfo().FillHV(instance)
3343
      i_be = self.cfg.GetClusterInfo().FillBE(instance)
3344
      for field in self.op.output_fields:
3345
        st_match = self._FIELDS_STATIC.Matches(field)
3346
        if field == "name":
3347
          val = instance.name
3348
        elif field == "os":
3349
          val = instance.os
3350
        elif field == "pnode":
3351
          val = instance.primary_node
3352
        elif field == "snodes":
3353
          val = list(instance.secondary_nodes)
3354
        elif field == "admin_state":
3355
          val = instance.admin_up
3356
        elif field == "oper_state":
3357
          if instance.primary_node in bad_nodes:
3358
            val = None
3359
          else:
3360
            val = bool(live_data.get(instance.name))
3361
        elif field == "status":
3362
          if instance.primary_node in off_nodes:
3363
            val = "ERROR_nodeoffline"
3364
          elif instance.primary_node in bad_nodes:
3365
            val = "ERROR_nodedown"
3366
          else:
3367
            running = bool(live_data.get(instance.name))
3368
            if running:
3369
              if instance.admin_up:
3370
                val = "running"
3371
              else:
3372
                val = "ERROR_up"
3373
            else:
3374
              if instance.admin_up:
3375
                val = "ERROR_down"
3376
              else:
3377
                val = "ADMIN_down"
3378
        elif field == "oper_ram":
3379
          if instance.primary_node in bad_nodes:
3380
            val = None
3381
          elif instance.name in live_data:
3382
            val = live_data[instance.name].get("memory", "?")
3383
          else:
3384
            val = "-"
3385
        elif field == "disk_template":
3386
          val = instance.disk_template
3387
        elif field == "ip":
3388
          val = instance.nics[0].ip
3389
        elif field == "bridge":
3390
          val = instance.nics[0].bridge
3391
        elif field == "mac":
3392
          val = instance.nics[0].mac
3393
        elif field == "sda_size" or field == "sdb_size":
3394
          idx = ord(field[2]) - ord('a')
3395
          try:
3396
            val = instance.FindDisk(idx).size
3397
          except errors.OpPrereqError:
3398
            val = None
3399
        elif field == "disk_usage": # total disk usage per node
3400
          disk_sizes = [{'size': disk.size} for disk in instance.disks]
3401
          val = _ComputeDiskSize(instance.disk_template, disk_sizes)
3402
        elif field == "tags":
3403
          val = list(instance.GetTags())
3404
        elif field == "serial_no":
3405
          val = instance.serial_no
3406
        elif field == "network_port":
3407
          val = instance.network_port
3408
        elif field == "hypervisor":
3409
          val = instance.hypervisor
3410
        elif field == "hvparams":
3411
          val = i_hv
3412
        elif (field.startswith(HVPREFIX) and
3413
              field[len(HVPREFIX):] in constants.HVS_PARAMETERS):
3414
          val = i_hv.get(field[len(HVPREFIX):], None)
3415
        elif field == "beparams":
3416
          val = i_be
3417
        elif (field.startswith(BEPREFIX) and
3418
              field[len(BEPREFIX):] in constants.BES_PARAMETERS):
3419
          val = i_be.get(field[len(BEPREFIX):], None)
3420
        elif st_match and st_match.groups():
3421
          # matches a variable list
3422
          st_groups = st_match.groups()
3423
          if st_groups and st_groups[0] == "disk":
3424
            if st_groups[1] == "count":
3425
              val = len(instance.disks)
3426
            elif st_groups[1] == "sizes":
3427
              val = [disk.size for disk in instance.disks]
3428
            elif st_groups[1] == "size":
3429
              try:
3430
                val = instance.FindDisk(st_groups[2]).size
3431
              except errors.OpPrereqError:
3432
                val = None
3433
            else:
3434
              assert False, "Unhandled disk parameter"
3435
          elif st_groups[0] == "nic":
3436
            if st_groups[1] == "count":
3437
              val = len(instance.nics)
3438
            elif st_groups[1] == "macs":
3439
              val = [nic.mac for nic in instance.nics]
3440
            elif st_groups[1] == "ips":
3441
              val = [nic.ip for nic in instance.nics]
3442
            elif st_groups[1] == "bridges":
3443
              val = [nic.bridge for nic in instance.nics]
3444
            else:
3445
              # index-based item
3446
              nic_idx = int(st_groups[2])
3447
              if nic_idx >= len(instance.nics):
3448
                val = None
3449
              else:
3450
                if st_groups[1] == "mac":
3451
                  val = instance.nics[nic_idx].mac
3452
                elif st_groups[1] == "ip":
3453
                  val = instance.nics[nic_idx].ip
3454
                elif st_groups[1] == "bridge":
3455
                  val = instance.nics[nic_idx].bridge
3456
                else:
3457
                  assert False, "Unhandled NIC parameter"
3458
          else:
3459
            assert False, "Unhandled variable parameter"
3460
        else:
3461
          raise errors.ParameterError(field)
3462
        iout.append(val)
3463
      output.append(iout)
3464

    
3465
    return output
3466

    
3467

    
3468
class LUFailoverInstance(LogicalUnit):
3469
  """Failover an instance.
3470

3471
  """
3472
  HPATH = "instance-failover"
3473
  HTYPE = constants.HTYPE_INSTANCE
3474
  _OP_REQP = ["instance_name", "ignore_consistency"]
3475
  REQ_BGL = False
3476

    
3477
  def ExpandNames(self):
3478
    self._ExpandAndLockInstance()
3479
    self.needed_locks[locking.LEVEL_NODE] = []
3480
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3481

    
3482
  def DeclareLocks(self, level):
3483
    if level == locking.LEVEL_NODE:
3484
      self._LockInstancesNodes()
3485

    
3486
  def BuildHooksEnv(self):
3487
    """Build hooks env.
3488

3489
    This runs on master, primary and secondary nodes of the instance.
3490

3491
    """
3492
    env = {
3493
      "IGNORE_CONSISTENCY": self.op.ignore_consistency,
3494
      }
3495
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3496
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3497
    return env, nl, nl
3498

    
3499
  def CheckPrereq(self):
3500
    """Check prerequisites.
3501

3502
    This checks that the instance is in the cluster.
3503

3504
    """
3505
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3506
    assert self.instance is not None, \
3507
      "Cannot retrieve locked instance %s" % self.op.instance_name
3508

    
3509
    bep = self.cfg.GetClusterInfo().FillBE(instance)
3510
    if instance.disk_template not in constants.DTS_NET_MIRROR:
3511
      raise errors.OpPrereqError("Instance's disk layout is not"
3512
                                 " network mirrored, cannot failover.")
3513

    
3514
    secondary_nodes = instance.secondary_nodes
3515
    if not secondary_nodes:
3516
      raise errors.ProgrammerError("no secondary node but using "
3517
                                   "a mirrored disk template")
3518

    
3519
    target_node = secondary_nodes[0]
3520
    _CheckNodeOnline(self, target_node)
3521
    _CheckNodeNotDrained(self, target_node)
3522
    # check memory requirements on the secondary node
3523
    _CheckNodeFreeMemory(self, target_node, "failing over instance %s" %
3524
                         instance.name, bep[constants.BE_MEMORY],
3525
                         instance.hypervisor)
3526

    
3527
    # check bridge existance
3528
    brlist = [nic.bridge for nic in instance.nics]
3529
    result = self.rpc.call_bridges_exist(target_node, brlist)
3530
    result.Raise()
3531
    if not result.data:
3532
      raise errors.OpPrereqError("One or more target bridges %s does not"
3533
                                 " exist on destination node '%s'" %
3534
                                 (brlist, target_node))
3535

    
3536
  def Exec(self, feedback_fn):
3537
    """Failover an instance.
3538

3539
    The failover is done by shutting it down on its present node and
3540
    starting it on the secondary.
3541

3542
    """
3543
    instance = self.instance
3544

    
3545
    source_node = instance.primary_node
3546
    target_node = instance.secondary_nodes[0]
3547

    
3548
    feedback_fn("* checking disk consistency between source and target")
3549
    for dev in instance.disks:
3550
      # for drbd, these are drbd over lvm
3551
      if not _CheckDiskConsistency(self, dev, target_node, False):
3552
        if instance.admin_up and not self.op.ignore_consistency:
3553
          raise errors.OpExecError("Disk %s is degraded on target node,"
3554
                                   " aborting failover." % dev.iv_name)
3555

    
3556
    feedback_fn("* shutting down instance on source node")
3557
    logging.info("Shutting down instance %s on node %s",
3558
                 instance.name, source_node)
3559

    
3560
    result = self.rpc.call_instance_shutdown(source_node, instance)
3561
    msg = result.RemoteFailMsg()
3562
    if msg:
3563
      if self.op.ignore_consistency:
3564
        self.proc.LogWarning("Could not shutdown instance %s on node %s."
3565
                             " Proceeding anyway. Please make sure node"
3566
                             " %s is down. Error details: %s",
3567
                             instance.name, source_node, source_node, msg)
3568
      else:
3569
        raise errors.OpExecError("Could not shutdown instance %s on"
3570
                                 " node %s: %s" %
3571
                                 (instance.name, source_node, msg))
3572

    
3573
    feedback_fn("* deactivating the instance's disks on source node")
3574
    if not _ShutdownInstanceDisks(self, instance, ignore_primary=True):
3575
      raise errors.OpExecError("Can't shut down the instance's disks.")
3576

    
3577
    instance.primary_node = target_node
3578
    # distribute new instance config to the other nodes
3579
    self.cfg.Update(instance)
3580

    
3581
    # Only start the instance if it's marked as up
3582
    if instance.admin_up:
3583
      feedback_fn("* activating the instance's disks on target node")
3584
      logging.info("Starting instance %s on node %s",
3585
                   instance.name, target_node)
3586

    
3587
      disks_ok, dummy = _AssembleInstanceDisks(self, instance,
3588
                                               ignore_secondaries=True)
3589
      if not disks_ok:
3590
        _ShutdownInstanceDisks(self, instance)
3591
        raise errors.OpExecError("Can't activate the instance's disks")
3592

    
3593
      feedback_fn("* starting the instance on the target node")
3594
      result = self.rpc.call_instance_start(target_node, instance, None, None)
3595
      msg = result.RemoteFailMsg()
3596
      if msg:
3597
        _ShutdownInstanceDisks(self, instance)
3598
        raise errors.OpExecError("Could not start instance %s on node %s: %s" %
3599
                                 (instance.name, target_node, msg))
3600

    
3601

    
3602
class LUMigrateInstance(LogicalUnit):
3603
  """Migrate an instance.
3604

3605
  This is migration without shutting down, compared to the failover,
3606
  which is done with shutdown.
3607

3608
  """
3609
  HPATH = "instance-migrate"
3610
  HTYPE = constants.HTYPE_INSTANCE
3611
  _OP_REQP = ["instance_name", "live", "cleanup"]
3612

    
3613
  REQ_BGL = False
3614

    
3615
  def ExpandNames(self):
3616
    self._ExpandAndLockInstance()
3617
    self.needed_locks[locking.LEVEL_NODE] = []
3618
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3619

    
3620
  def DeclareLocks(self, level):
3621
    if level == locking.LEVEL_NODE:
3622
      self._LockInstancesNodes()
3623

    
3624
  def BuildHooksEnv(self):
3625
    """Build hooks env.
3626

3627
    This runs on master, primary and secondary nodes of the instance.
3628

3629
    """
3630
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3631
    env["MIGRATE_LIVE"] = self.op.live
3632
    env["MIGRATE_CLEANUP"] = self.op.cleanup
3633
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3634
    return env, nl, nl
3635

    
3636
  def CheckPrereq(self):
3637
    """Check prerequisites.
3638

3639
    This checks that the instance is in the cluster.
3640

3641
    """
3642
    instance = self.cfg.GetInstanceInfo(
3643
      self.cfg.ExpandInstanceName(self.op.instance_name))
3644
    if instance is None:
3645
      raise errors.OpPrereqError("Instance '%s' not known" %
3646
                                 self.op.instance_name)
3647

    
3648
    if instance.disk_template != constants.DT_DRBD8:
3649
      raise errors.OpPrereqError("Instance's disk layout is not"
3650
                                 " drbd8, cannot migrate.")
3651

    
3652
    secondary_nodes = instance.secondary_nodes
3653
    if not secondary_nodes:
3654
      raise errors.ConfigurationError("No secondary node but using"
3655
                                      " drbd8 disk template")
3656

    
3657
    i_be = self.cfg.GetClusterInfo().FillBE(instance)
3658

    
3659
    target_node = secondary_nodes[0]
3660
    # check memory requirements on the secondary node
3661
    _CheckNodeFreeMemory(self, target_node, "migrating instance %s" %
3662
                         instance.name, i_be[constants.BE_MEMORY],
3663
                         instance.hypervisor)
3664

    
3665
    # check bridge existance
3666
    brlist = [nic.bridge for nic in instance.nics]
3667
    result = self.rpc.call_bridges_exist(target_node, brlist)
3668
    if result.failed or not result.data:
3669
      raise errors.OpPrereqError("One or more target bridges %s does not"
3670
                                 " exist on destination node '%s'" %
3671
                                 (brlist, target_node))
3672

    
3673
    if not self.op.cleanup:
3674
      _CheckNodeNotDrained(self, target_node)
3675
      result = self.rpc.call_instance_migratable(instance.primary_node,
3676
                                                 instance)
3677
      msg = result.RemoteFailMsg()
3678
      if msg:
3679
        raise errors.OpPrereqError("Can't migrate: %s - please use failover" %
3680
                                   msg)
3681

    
3682
    self.instance = instance
3683

    
3684
  def _WaitUntilSync(self):
3685
    """Poll with custom rpc for disk sync.
3686

3687
    This uses our own step-based rpc call.
3688

3689
    """
3690
    self.feedback_fn("* wait until resync is done")
3691
    all_done = False
3692
    while not all_done:
3693
      all_done = True
3694
      result = self.rpc.call_drbd_wait_sync(self.all_nodes,
3695
                                            self.nodes_ip,
3696
                                            self.instance.disks)
3697
      min_percent = 100
3698
      for node, nres in result.items():
3699
        msg = nres.RemoteFailMsg()
3700
        if msg:
3701
          raise errors.OpExecError("Cannot resync disks on node %s: %s" %
3702
                                   (node, msg))
3703
        node_done, node_percent = nres.payload
3704
        all_done = all_done and node_done
3705
        if node_percent is not None:
3706
          min_percent = min(min_percent, node_percent)
3707
      if not all_done:
3708
        if min_percent < 100:
3709
          self.feedback_fn("   - progress: %.1f%%" % min_percent)
3710
        time.sleep(2)
3711

    
3712
  def _EnsureSecondary(self, node):
3713
    """Demote a node to secondary.
3714

3715
    """
3716
    self.feedback_fn("* switching node %s to secondary mode" % node)
3717

    
3718
    for dev in self.instance.disks:
3719
      self.cfg.SetDiskID(dev, node)
3720

    
3721
    result = self.rpc.call_blockdev_close(node, self.instance.name,
3722
                                          self.instance.disks)
3723
    msg = result.RemoteFailMsg()
3724
    if msg:
3725
      raise errors.OpExecError("Cannot change disk to secondary on node %s,"
3726
                               " error %s" % (node, msg))
3727

    
3728
  def _GoStandalone(self):
3729
    """Disconnect from the network.
3730

3731
    """
3732
    self.feedback_fn("* changing into standalone mode")
3733
    result = self.rpc.call_drbd_disconnect_net(self.all_nodes, self.nodes_ip,
3734
                                               self.instance.disks)
3735
    for node, nres in result.items():
3736
      msg = nres.RemoteFailMsg()
3737
      if msg:
3738
        raise errors.OpExecError("Cannot disconnect disks node %s,"
3739
                                 " error %s" % (node, msg))
3740

    
3741
  def _GoReconnect(self, multimaster):
3742
    """Reconnect to the network.
3743

3744
    """
3745
    if multimaster:
3746
      msg = "dual-master"
3747
    else:
3748
      msg = "single-master"
3749
    self.feedback_fn("* changing disks into %s mode" % msg)
3750
    result = self.rpc.call_drbd_attach_net(self.all_nodes, self.nodes_ip,
3751
                                           self.instance.disks,
3752
                                           self.instance.name, multimaster)
3753
    for node, nres in result.items():
3754
      msg = nres.RemoteFailMsg()
3755
      if msg:
3756
        raise errors.OpExecError("Cannot change disks config on node %s,"
3757
                                 " error: %s" % (node, msg))
3758

    
3759
  def _ExecCleanup(self):
3760
    """Try to cleanup after a failed migration.
3761

3762
    The cleanup is done by:
3763
      - check that the instance is running only on one node
3764
        (and update the config if needed)
3765
      - change disks on its secondary node to secondary
3766
      - wait until disks are fully synchronized
3767
      - disconnect from the network
3768
      - change disks into single-master mode
3769
      - wait again until disks are fully synchronized
3770

3771
    """
3772
    instance = self.instance
3773
    target_node = self.target_node
3774
    source_node = self.source_node
3775

    
3776
    # check running on only one node
3777
    self.feedback_fn("* checking where the instance actually runs"
3778
                     " (if this hangs, the hypervisor might be in"
3779
                     " a bad state)")
3780
    ins_l = self.rpc.call_instance_list(self.all_nodes, [instance.hypervisor])
3781
    for node, result in ins_l.items():
3782
      result.Raise()
3783
      if not isinstance(result.data, list):
3784
        raise errors.OpExecError("Can't contact node '%s'" % node)
3785

    
3786
    runningon_source = instance.name in ins_l[source_node].data
3787
    runningon_target = instance.name in ins_l[target_node].data
3788

    
3789
    if runningon_source and runningon_target:
3790
      raise errors.OpExecError("Instance seems to be running on two nodes,"
3791
                               " or the hypervisor is confused. You will have"
3792
                               " to ensure manually that it runs only on one"
3793
                               " and restart this operation.")
3794

    
3795
    if not (runningon_source or runningon_target):
3796
      raise errors.OpExecError("Instance does not seem to be running at all."
3797
                               " In this case, it's safer to repair by"
3798
                               " running 'gnt-instance stop' to ensure disk"
3799
                               " shutdown, and then restarting it.")
3800

    
3801
    if runningon_target:
3802
      # the migration has actually succeeded, we need to update the config
3803
      self.feedback_fn("* instance running on secondary node (%s),"
3804
                       " updating config" % target_node)
3805
      instance.primary_node = target_node
3806
      self.cfg.Update(instance)
3807
      demoted_node = source_node
3808
    else:
3809
      self.feedback_fn("* instance confirmed to be running on its"
3810
                       " primary node (%s)" % source_node)
3811
      demoted_node = target_node
3812

    
3813
    self._EnsureSecondary(demoted_node)
3814
    try:
3815
      self._WaitUntilSync()
3816
    except errors.OpExecError:
3817
      # we ignore here errors, since if the device is standalone, it
3818
      # won't be able to sync
3819
      pass
3820
    self._GoStandalone()
3821
    self._GoReconnect(False)
3822
    self._WaitUntilSync()
3823

    
3824
    self.feedback_fn("* done")
3825

    
3826
  def _RevertDiskStatus(self):
3827
    """Try to revert the disk status after a failed migration.
3828

3829
    """
3830
    target_node = self.target_node
3831
    try:
3832
      self._EnsureSecondary(target_node)
3833
      self._GoStandalone()
3834
      self._GoReconnect(False)
3835
      self._WaitUntilSync()
3836
    except errors.OpExecError, err:
3837
      self.LogWarning("Migration failed and I can't reconnect the"
3838
                      " drives: error '%s'\n"
3839
                      "Please look and recover the instance status" %
3840
                      str(err))
3841

    
3842
  def _AbortMigration(self):
3843
    """Call the hypervisor code to abort a started migration.
3844

3845
    """
3846
    instance = self.instance
3847
    target_node = self.target_node
3848
    migration_info = self.migration_info
3849

    
3850
    abort_result = self.rpc.call_finalize_migration(target_node,
3851
                                                    instance,
3852
                                                    migration_info,
3853
                                                    False)
3854
    abort_msg = abort_result.RemoteFailMsg()
3855
    if abort_msg:
3856
      logging.error("Aborting migration failed on target node %s: %s" %
3857
                    (target_node, abort_msg))
3858
      # Don't raise an exception here, as we stil have to try to revert the
3859
      # disk status, even if this step failed.
3860

    
3861
  def _ExecMigration(self):
3862
    """Migrate an instance.
3863

3864
    The migrate is done by:
3865
      - change the disks into dual-master mode
3866
      - wait until disks are fully synchronized again
3867
      - migrate the instance
3868
      - change disks on the new secondary node (the old primary) to secondary
3869
      - wait until disks are fully synchronized
3870
      - change disks into single-master mode
3871

3872
    """
3873
    instance = self.instance
3874
    target_node = self.target_node
3875
    source_node = self.source_node
3876

    
3877
    self.feedback_fn("* checking disk consistency between source and target")
3878
    for dev in instance.disks:
3879
      if not _CheckDiskConsistency(self, dev, target_node, False):
3880
        raise errors.OpExecError("Disk %s is degraded or not fully"
3881
                                 " synchronized on target node,"
3882
                                 " aborting migrate." % dev.iv_name)
3883

    
3884
    # First get the migration information from the remote node
3885
    result = self.rpc.call_migration_info(source_node, instance)
3886
    msg = result.RemoteFailMsg()
3887
    if msg:
3888
      log_err = ("Failed fetching source migration information from %s: %s" %
3889
                 (source_node, msg))
3890
      logging.error(log_err)
3891
      raise errors.OpExecError(log_err)
3892

    
3893
    self.migration_info = migration_info = result.payload
3894

    
3895
    # Then switch the disks to master/master mode
3896
    self._EnsureSecondary(target_node)
3897
    self._GoStandalone()
3898
    self._GoReconnect(True)
3899
    self._WaitUntilSync()
3900

    
3901
    self.feedback_fn("* preparing %s to accept the instance" % target_node)
3902
    result = self.rpc.call_accept_instance(target_node,
3903
                                           instance,
3904
                                           migration_info,
3905
                                           self.nodes_ip[target_node])
3906

    
3907
    msg = result.RemoteFailMsg()
3908
    if msg:
3909
      logging.error("Instance pre-migration failed, trying to revert"
3910
                    " disk status: %s", msg)
3911
      self._AbortMigration()
3912
      self._RevertDiskStatus()
3913
      raise errors.OpExecError("Could not pre-migrate instance %s: %s" %
3914
                               (instance.name, msg))
3915

    
3916
    self.feedback_fn("* migrating instance to %s" % target_node)
3917
    time.sleep(10)
3918
    result = self.rpc.call_instance_migrate(source_node, instance,
3919
                                            self.nodes_ip[target_node],
3920
                                            self.op.live)
3921
    msg = result.RemoteFailMsg()
3922
    if msg:
3923
      logging.error("Instance migration failed, trying to revert"
3924
                    " disk status: %s", msg)
3925
      self._AbortMigration()
3926
      self._RevertDiskStatus()
3927
      raise errors.OpExecError("Could not migrate instance %s: %s" %
3928
                               (instance.name, msg))
3929
    time.sleep(10)
3930

    
3931
    instance.primary_node = target_node
3932
    # distribute new instance config to the other nodes
3933
    self.cfg.Update(instance)
3934

    
3935
    result = self.rpc.call_finalize_migration(target_node,
3936
                                              instance,
3937
                                              migration_info,
3938
                                              True)
3939
    msg = result.RemoteFailMsg()
3940
    if msg:
3941
      logging.error("Instance migration succeeded, but finalization failed:"
3942
                    " %s" % msg)
3943
      raise errors.OpExecError("Could not finalize instance migration: %s" %
3944
                               msg)
3945

    
3946
    self._EnsureSecondary(source_node)
3947
    self._WaitUntilSync()
3948
    self._GoStandalone()
3949
    self._GoReconnect(False)
3950
    self._WaitUntilSync()
3951

    
3952
    self.feedback_fn("* done")
3953

    
3954
  def Exec(self, feedback_fn):
3955
    """Perform the migration.
3956

3957
    """
3958
    self.feedback_fn = feedback_fn
3959

    
3960
    self.source_node = self.instance.primary_node
3961
    self.target_node = self.instance.secondary_nodes[0]
3962
    self.all_nodes = [self.source_node, self.target_node]
3963
    self.nodes_ip = {
3964
      self.source_node: self.cfg.GetNodeInfo(self.source_node).secondary_ip,
3965
      self.target_node: self.cfg.GetNodeInfo(self.target_node).secondary_ip,
3966
      }
3967
    if self.op.cleanup:
3968
      return self._ExecCleanup()
3969
    else:
3970
      return self._ExecMigration()
3971

    
3972

    
3973
def _CreateBlockDev(lu, node, instance, device, force_create,
3974
                    info, force_open):
3975
  """Create a tree of block devices on a given node.
3976

3977
  If this device type has to be created on secondaries, create it and
3978
  all its children.
3979

3980
  If not, just recurse to children keeping the same 'force' value.
3981

3982
  @param lu: the lu on whose behalf we execute
3983
  @param node: the node on which to create the device
3984
  @type instance: L{objects.Instance}
3985
  @param instance: the instance which owns the device
3986
  @type device: L{objects.Disk}
3987
  @param device: the device to create
3988
  @type force_create: boolean
3989
  @param force_create: whether to force creation of this device; this
3990
      will be change to True whenever we find a device which has
3991
      CreateOnSecondary() attribute
3992
  @param info: the extra 'metadata' we should attach to the device
3993
      (this will be represented as a LVM tag)
3994
  @type force_open: boolean
3995
  @param force_open: this parameter will be passes to the
3996
      L{backend.BlockdevCreate} function where it specifies
3997
      whether we run on primary or not, and it affects both
3998
      the child assembly and the device own Open() execution
3999

4000
  """
4001
  if device.CreateOnSecondary():
4002
    force_create = True
4003

    
4004
  if device.children: