Statistics
| Branch: | Tag: | Revision:

root / lib / cmdlib.py @ 5af3da74

History | View | Annotate | Download (245.9 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
         msg = to_result.RemoteFailMsg()
1385
         if msg:
1386
           msg = ("Copy of file %s to node %s failed: %s" %
1387
                   (constants.SSH_KNOWN_HOSTS_FILE, to_node, msg))
1388
           self.proc.LogWarning(msg)
1389

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

    
1396

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

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

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

    
1412

    
1413
class LUSetClusterParams(LogicalUnit):
1414
  """Change the parameters of the cluster.
1415

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

    
1422
  def CheckArguments(self):
1423
    """Check parameters
1424

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

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

    
1445
  def BuildHooksEnv(self):
1446
    """Build hooks env.
1447

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

    
1456
  def CheckPrereq(self):
1457
    """Check prerequisites.
1458

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

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

    
1471
    node_list = self.acquired_locks[locking.LEVEL_NODE]
1472

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

    
1488
    self.cluster = cluster = self.cfg.GetClusterInfo()
1489
    # validate params changes
1490
    if self.op.beparams:
1491
      utils.ForceDictType(self.op.beparams, constants.BES_PARAMETER_TYPES)
1492
      self.new_beparams = objects.FillDict(
1493
        cluster.beparams[constants.PP_DEFAULT], self.op.beparams)
1494

    
1495
    if self.op.nicparams:
1496
      utils.ForceDictType(self.op.nicparams, constants.NICS_PARAMETER_TYPES)
1497
      self.new_nicparams = objects.FillDict(
1498
        cluster.nicparams[constants.PP_DEFAULT], self.op.nicparams)
1499
      objects.NIC.CheckParameterSyntax(self.new_nicparams)
1500

    
1501
    # hypervisor list/parameters
1502
    self.new_hvparams = objects.FillDict(cluster.hvparams, {})
1503
    if self.op.hvparams:
1504
      if not isinstance(self.op.hvparams, dict):
1505
        raise errors.OpPrereqError("Invalid 'hvparams' parameter on input")
1506
      for hv_name, hv_dict in self.op.hvparams.items():
1507
        if hv_name not in self.new_hvparams:
1508
          self.new_hvparams[hv_name] = hv_dict
1509
        else:
1510
          self.new_hvparams[hv_name].update(hv_dict)
1511

    
1512
    if self.op.enabled_hypervisors is not None:
1513
      self.hv_list = self.op.enabled_hypervisors
1514
    else:
1515
      self.hv_list = cluster.enabled_hypervisors
1516

    
1517
    if self.op.hvparams or self.op.enabled_hypervisors is not None:
1518
      # either the enabled list has changed, or the parameters have, validate
1519
      for hv_name, hv_params in self.new_hvparams.items():
1520
        if ((self.op.hvparams and hv_name in self.op.hvparams) or
1521
            (self.op.enabled_hypervisors and
1522
             hv_name in self.op.enabled_hypervisors)):
1523
          # either this is a new hypervisor, or its parameters have changed
1524
          hv_class = hypervisor.GetHypervisor(hv_name)
1525
          utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
1526
          hv_class.CheckParameterSyntax(hv_params)
1527
          _CheckHVParams(self, node_list, hv_name, hv_params)
1528

    
1529
  def Exec(self, feedback_fn):
1530
    """Change the parameters of the cluster.
1531

1532
    """
1533
    if self.op.vg_name is not None:
1534
      new_volume = self.op.vg_name
1535
      if not new_volume:
1536
        new_volume = None
1537
      if new_volume != self.cfg.GetVGName():
1538
        self.cfg.SetVGName(new_volume)
1539
      else:
1540
        feedback_fn("Cluster LVM configuration already in desired"
1541
                    " state, not changing")
1542
    if self.op.hvparams:
1543
      self.cluster.hvparams = self.new_hvparams
1544
    if self.op.enabled_hypervisors is not None:
1545
      self.cluster.enabled_hypervisors = self.op.enabled_hypervisors
1546
    if self.op.beparams:
1547
      self.cluster.beparams[constants.PP_DEFAULT] = self.new_beparams
1548
    if self.op.nicparams:
1549
      self.cluster.nicparams[constants.PP_DEFAULT] = self.new_nicparams
1550

    
1551
    if self.op.candidate_pool_size is not None:
1552
      self.cluster.candidate_pool_size = self.op.candidate_pool_size
1553

    
1554
    self.cfg.Update(self.cluster)
1555

    
1556
    # we want to update nodes after the cluster so that if any errors
1557
    # happen, we have recorded and saved the cluster info
1558
    if self.op.candidate_pool_size is not None:
1559
      _AdjustCandidatePool(self)
1560

    
1561

    
1562
def _RedistributeAncillaryFiles(lu, additional_nodes=None):
1563
  """Distribute additional files which are part of the cluster configuration.
1564

1565
  ConfigWriter takes care of distributing the config and ssconf files, but
1566
  there are more files which should be distributed to all nodes. This function
1567
  makes sure those are copied.
1568

1569
  @param lu: calling logical unit
1570
  @param additional_nodes: list of nodes not in the config to distribute to
1571

1572
  """
1573
  # 1. Gather target nodes
1574
  myself = lu.cfg.GetNodeInfo(lu.cfg.GetMasterNode())
1575
  dist_nodes = lu.cfg.GetNodeList()
1576
  if additional_nodes is not None:
1577
    dist_nodes.extend(additional_nodes)
1578
  if myself.name in dist_nodes:
1579
    dist_nodes.remove(myself.name)
1580
  # 2. Gather files to distribute
1581
  dist_files = set([constants.ETC_HOSTS,
1582
                    constants.SSH_KNOWN_HOSTS_FILE,
1583
                    constants.RAPI_CERT_FILE,
1584
                    constants.RAPI_USERS_FILE,
1585
                   ])
1586

    
1587
  enabled_hypervisors = lu.cfg.GetClusterInfo().enabled_hypervisors
1588
  for hv_name in enabled_hypervisors:
1589
    hv_class = hypervisor.GetHypervisor(hv_name)
1590
    dist_files.update(hv_class.GetAncillaryFiles())
1591

    
1592
  # 3. Perform the files upload
1593
  for fname in dist_files:
1594
    if os.path.exists(fname):
1595
      result = lu.rpc.call_upload_file(dist_nodes, fname)
1596
      for to_node, to_result in result.items():
1597
         msg = to_result.RemoteFailMsg()
1598
         if msg:
1599
           msg = ("Copy of file %s to node %s failed: %s" %
1600
                   (fname, to_node, msg))
1601
           lu.proc.LogWarning(msg)
1602

    
1603

    
1604
class LURedistributeConfig(NoHooksLU):
1605
  """Force the redistribution of cluster configuration.
1606

1607
  This is a very simple LU.
1608

1609
  """
1610
  _OP_REQP = []
1611
  REQ_BGL = False
1612

    
1613
  def ExpandNames(self):
1614
    self.needed_locks = {
1615
      locking.LEVEL_NODE: locking.ALL_SET,
1616
    }
1617
    self.share_locks[locking.LEVEL_NODE] = 1
1618

    
1619
  def CheckPrereq(self):
1620
    """Check prerequisites.
1621

1622
    """
1623

    
1624
  def Exec(self, feedback_fn):
1625
    """Redistribute the configuration.
1626

1627
    """
1628
    self.cfg.Update(self.cfg.GetClusterInfo())
1629
    _RedistributeAncillaryFiles(self)
1630

    
1631

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

1635
  """
1636
  if not instance.disks:
1637
    return True
1638

    
1639
  if not oneshot:
1640
    lu.proc.LogInfo("Waiting for instance %s to sync disks." % instance.name)
1641

    
1642
  node = instance.primary_node
1643

    
1644
  for dev in instance.disks:
1645
    lu.cfg.SetDiskID(dev, node)
1646

    
1647
  retries = 0
1648
  while True:
1649
    max_time = 0
1650
    done = True
1651
    cumul_degraded = False
1652
    rstats = lu.rpc.call_blockdev_getmirrorstatus(node, instance.disks)
1653
    if rstats.failed or not rstats.data:
1654
      lu.LogWarning("Can't get any data from node %s", node)
1655
      retries += 1
1656
      if retries >= 10:
1657
        raise errors.RemoteError("Can't contact node %s for mirror data,"
1658
                                 " aborting." % node)
1659
      time.sleep(6)
1660
      continue
1661
    rstats = rstats.data
1662
    retries = 0
1663
    for i, mstat in enumerate(rstats):
1664
      if mstat is None:
1665
        lu.LogWarning("Can't compute data for node %s/%s",
1666
                           node, instance.disks[i].iv_name)
1667
        continue
1668
      # we ignore the ldisk parameter
1669
      perc_done, est_time, is_degraded, _ = mstat
1670
      cumul_degraded = cumul_degraded or (is_degraded and perc_done is None)
1671
      if perc_done is not None:
1672
        done = False
1673
        if est_time is not None:
1674
          rem_time = "%d estimated seconds remaining" % est_time
1675
          max_time = est_time
1676
        else:
1677
          rem_time = "no time estimate"
1678
        lu.proc.LogInfo("- device %s: %5.2f%% done, %s" %
1679
                        (instance.disks[i].iv_name, perc_done, rem_time))
1680
    if done or oneshot:
1681
      break
1682

    
1683
    time.sleep(min(60, max_time))
1684

    
1685
  if done:
1686
    lu.proc.LogInfo("Instance %s's disks are in sync." % instance.name)
1687
  return not cumul_degraded
1688

    
1689

    
1690
def _CheckDiskConsistency(lu, dev, node, on_primary, ldisk=False):
1691
  """Check that mirrors are not degraded.
1692

1693
  The ldisk parameter, if True, will change the test from the
1694
  is_degraded attribute (which represents overall non-ok status for
1695
  the device(s)) to the ldisk (representing the local storage status).
1696

1697
  """
1698
  lu.cfg.SetDiskID(dev, node)
1699
  if ldisk:
1700
    idx = 6
1701
  else:
1702
    idx = 5
1703

    
1704
  result = True
1705
  if on_primary or dev.AssembleOnSecondary():
1706
    rstats = lu.rpc.call_blockdev_find(node, dev)
1707
    msg = rstats.RemoteFailMsg()
1708
    if msg:
1709
      lu.LogWarning("Can't find disk on node %s: %s", node, msg)
1710
      result = False
1711
    elif not rstats.payload:
1712
      lu.LogWarning("Can't find disk on node %s", node)
1713
      result = False
1714
    else:
1715
      result = result and (not rstats.payload[idx])
1716
  if dev.children:
1717
    for child in dev.children:
1718
      result = result and _CheckDiskConsistency(lu, child, node, on_primary)
1719

    
1720
  return result
1721

    
1722

    
1723
class LUDiagnoseOS(NoHooksLU):
1724
  """Logical unit for OS diagnose/query.
1725

1726
  """
1727
  _OP_REQP = ["output_fields", "names"]
1728
  REQ_BGL = False
1729
  _FIELDS_STATIC = utils.FieldSet()
1730
  _FIELDS_DYNAMIC = utils.FieldSet("name", "valid", "node_status")
1731

    
1732
  def ExpandNames(self):
1733
    if self.op.names:
1734
      raise errors.OpPrereqError("Selective OS query not supported")
1735

    
1736
    _CheckOutputFields(static=self._FIELDS_STATIC,
1737
                       dynamic=self._FIELDS_DYNAMIC,
1738
                       selected=self.op.output_fields)
1739

    
1740
    # Lock all nodes, in shared mode
1741
    # Temporary removal of locks, should be reverted later
1742
    # TODO: reintroduce locks when they are lighter-weight
1743
    self.needed_locks = {}
1744
    #self.share_locks[locking.LEVEL_NODE] = 1
1745
    #self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
1746

    
1747
  def CheckPrereq(self):
1748
    """Check prerequisites.
1749

1750
    """
1751

    
1752
  @staticmethod
1753
  def _DiagnoseByOS(node_list, rlist):
1754
    """Remaps a per-node return list into an a per-os per-node dictionary
1755

1756
    @param node_list: a list with the names of all nodes
1757
    @param rlist: a map with node names as keys and OS objects as values
1758

1759
    @rtype: dict
1760
    @return: a dictionary with osnames as keys and as value another map, with
1761
        nodes as keys and list of OS objects as values, eg::
1762

1763
          {"debian-etch": {"node1": [<object>,...],
1764
                           "node2": [<object>,]}
1765
          }
1766

1767
    """
1768
    all_os = {}
1769
    # we build here the list of nodes that didn't fail the RPC (at RPC
1770
    # level), so that nodes with a non-responding node daemon don't
1771
    # make all OSes invalid
1772
    good_nodes = [node_name for node_name in rlist
1773
                  if not rlist[node_name].failed]
1774
    for node_name, nr in rlist.iteritems():
1775
      if nr.failed or not nr.data:
1776
        continue
1777
      for os_obj in nr.data:
1778
        if os_obj.name not in all_os:
1779
          # build a list of nodes for this os containing empty lists
1780
          # for each node in node_list
1781
          all_os[os_obj.name] = {}
1782
          for nname in good_nodes:
1783
            all_os[os_obj.name][nname] = []
1784
        all_os[os_obj.name][node_name].append(os_obj)
1785
    return all_os
1786

    
1787
  def Exec(self, feedback_fn):
1788
    """Compute the list of OSes.
1789

1790
    """
1791
    valid_nodes = [node for node in self.cfg.GetOnlineNodeList()]
1792
    node_data = self.rpc.call_os_diagnose(valid_nodes)
1793
    if node_data == False:
1794
      raise errors.OpExecError("Can't gather the list of OSes")
1795
    pol = self._DiagnoseByOS(valid_nodes, node_data)
1796
    output = []
1797
    for os_name, os_data in pol.iteritems():
1798
      row = []
1799
      for field in self.op.output_fields:
1800
        if field == "name":
1801
          val = os_name
1802
        elif field == "valid":
1803
          val = utils.all([osl and osl[0] for osl in os_data.values()])
1804
        elif field == "node_status":
1805
          val = {}
1806
          for node_name, nos_list in os_data.iteritems():
1807
            val[node_name] = [(v.status, v.path) for v in nos_list]
1808
        else:
1809
          raise errors.ParameterError(field)
1810
        row.append(val)
1811
      output.append(row)
1812

    
1813
    return output
1814

    
1815

    
1816
class LURemoveNode(LogicalUnit):
1817
  """Logical unit for removing a node.
1818

1819
  """
1820
  HPATH = "node-remove"
1821
  HTYPE = constants.HTYPE_NODE
1822
  _OP_REQP = ["node_name"]
1823

    
1824
  def BuildHooksEnv(self):
1825
    """Build hooks env.
1826

1827
    This doesn't run on the target node in the pre phase as a failed
1828
    node would then be impossible to remove.
1829

1830
    """
1831
    env = {
1832
      "OP_TARGET": self.op.node_name,
1833
      "NODE_NAME": self.op.node_name,
1834
      }
1835
    all_nodes = self.cfg.GetNodeList()
1836
    all_nodes.remove(self.op.node_name)
1837
    return env, all_nodes, all_nodes
1838

    
1839
  def CheckPrereq(self):
1840
    """Check prerequisites.
1841

1842
    This checks:
1843
     - the node exists in the configuration
1844
     - it does not have primary or secondary instances
1845
     - it's not the master
1846

1847
    Any errors are signalled by raising errors.OpPrereqError.
1848

1849
    """
1850
    node = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.node_name))
1851
    if node is None:
1852
      raise errors.OpPrereqError, ("Node '%s' is unknown." % self.op.node_name)
1853

    
1854
    instance_list = self.cfg.GetInstanceList()
1855

    
1856
    masternode = self.cfg.GetMasterNode()
1857
    if node.name == masternode:
1858
      raise errors.OpPrereqError("Node is the master node,"
1859
                                 " you need to failover first.")
1860

    
1861
    for instance_name in instance_list:
1862
      instance = self.cfg.GetInstanceInfo(instance_name)
1863
      if node.name in instance.all_nodes:
1864
        raise errors.OpPrereqError("Instance %s is still running on the node,"
1865
                                   " please remove first." % instance_name)
1866
    self.op.node_name = node.name
1867
    self.node = node
1868

    
1869
  def Exec(self, feedback_fn):
1870
    """Removes the node from the cluster.
1871

1872
    """
1873
    node = self.node
1874
    logging.info("Stopping the node daemon and removing configs from node %s",
1875
                 node.name)
1876

    
1877
    self.context.RemoveNode(node.name)
1878

    
1879
    self.rpc.call_node_leave_cluster(node.name)
1880

    
1881
    # Promote nodes to master candidate as needed
1882
    _AdjustCandidatePool(self)
1883

    
1884

    
1885
class LUQueryNodes(NoHooksLU):
1886
  """Logical unit for querying nodes.
1887

1888
  """
1889
  _OP_REQP = ["output_fields", "names", "use_locking"]
1890
  REQ_BGL = False
1891
  _FIELDS_DYNAMIC = utils.FieldSet(
1892
    "dtotal", "dfree",
1893
    "mtotal", "mnode", "mfree",
1894
    "bootid",
1895
    "ctotal", "cnodes", "csockets",
1896
    )
1897

    
1898
  _FIELDS_STATIC = utils.FieldSet(
1899
    "name", "pinst_cnt", "sinst_cnt",
1900
    "pinst_list", "sinst_list",
1901
    "pip", "sip", "tags",
1902
    "serial_no",
1903
    "master_candidate",
1904
    "master",
1905
    "offline",
1906
    "drained",
1907
    )
1908

    
1909
  def ExpandNames(self):
1910
    _CheckOutputFields(static=self._FIELDS_STATIC,
1911
                       dynamic=self._FIELDS_DYNAMIC,
1912
                       selected=self.op.output_fields)
1913

    
1914
    self.needed_locks = {}
1915
    self.share_locks[locking.LEVEL_NODE] = 1
1916

    
1917
    if self.op.names:
1918
      self.wanted = _GetWantedNodes(self, self.op.names)
1919
    else:
1920
      self.wanted = locking.ALL_SET
1921

    
1922
    self.do_node_query = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
1923
    self.do_locking = self.do_node_query and self.op.use_locking
1924
    if self.do_locking:
1925
      # if we don't request only static fields, we need to lock the nodes
1926
      self.needed_locks[locking.LEVEL_NODE] = self.wanted
1927

    
1928

    
1929
  def CheckPrereq(self):
1930
    """Check prerequisites.
1931

1932
    """
1933
    # The validation of the node list is done in the _GetWantedNodes,
1934
    # if non empty, and if empty, there's no validation to do
1935
    pass
1936

    
1937
  def Exec(self, feedback_fn):
1938
    """Computes the list of nodes and their attributes.
1939

1940
    """
1941
    all_info = self.cfg.GetAllNodesInfo()
1942
    if self.do_locking:
1943
      nodenames = self.acquired_locks[locking.LEVEL_NODE]
1944
    elif self.wanted != locking.ALL_SET:
1945
      nodenames = self.wanted
1946
      missing = set(nodenames).difference(all_info.keys())
1947
      if missing:
1948
        raise errors.OpExecError(
1949
          "Some nodes were removed before retrieving their data: %s" % missing)
1950
    else:
1951
      nodenames = all_info.keys()
1952

    
1953
    nodenames = utils.NiceSort(nodenames)
1954
    nodelist = [all_info[name] for name in nodenames]
1955

    
1956
    # begin data gathering
1957

    
1958
    if self.do_node_query:
1959
      live_data = {}
1960
      node_data = self.rpc.call_node_info(nodenames, self.cfg.GetVGName(),
1961
                                          self.cfg.GetHypervisorType())
1962
      for name in nodenames:
1963
        nodeinfo = node_data[name]
1964
        if not nodeinfo.failed and nodeinfo.data:
1965
          nodeinfo = nodeinfo.data
1966
          fn = utils.TryConvert
1967
          live_data[name] = {
1968
            "mtotal": fn(int, nodeinfo.get('memory_total', None)),
1969
            "mnode": fn(int, nodeinfo.get('memory_dom0', None)),
1970
            "mfree": fn(int, nodeinfo.get('memory_free', None)),
1971
            "dtotal": fn(int, nodeinfo.get('vg_size', None)),
1972
            "dfree": fn(int, nodeinfo.get('vg_free', None)),
1973
            "ctotal": fn(int, nodeinfo.get('cpu_total', None)),
1974
            "bootid": nodeinfo.get('bootid', None),
1975
            "cnodes": fn(int, nodeinfo.get('cpu_nodes', None)),
1976
            "csockets": fn(int, nodeinfo.get('cpu_sockets', None)),
1977
            }
1978
        else:
1979
          live_data[name] = {}
1980
    else:
1981
      live_data = dict.fromkeys(nodenames, {})
1982

    
1983
    node_to_primary = dict([(name, set()) for name in nodenames])
1984
    node_to_secondary = dict([(name, set()) for name in nodenames])
1985

    
1986
    inst_fields = frozenset(("pinst_cnt", "pinst_list",
1987
                             "sinst_cnt", "sinst_list"))
1988
    if inst_fields & frozenset(self.op.output_fields):
1989
      instancelist = self.cfg.GetInstanceList()
1990

    
1991
      for instance_name in instancelist:
1992
        inst = self.cfg.GetInstanceInfo(instance_name)
1993
        if inst.primary_node in node_to_primary:
1994
          node_to_primary[inst.primary_node].add(inst.name)
1995
        for secnode in inst.secondary_nodes:
1996
          if secnode in node_to_secondary:
1997
            node_to_secondary[secnode].add(inst.name)
1998

    
1999
    master_node = self.cfg.GetMasterNode()
2000

    
2001
    # end data gathering
2002

    
2003
    output = []
2004
    for node in nodelist:
2005
      node_output = []
2006
      for field in self.op.output_fields:
2007
        if field == "name":
2008
          val = node.name
2009
        elif field == "pinst_list":
2010
          val = list(node_to_primary[node.name])
2011
        elif field == "sinst_list":
2012
          val = list(node_to_secondary[node.name])
2013
        elif field == "pinst_cnt":
2014
          val = len(node_to_primary[node.name])
2015
        elif field == "sinst_cnt":
2016
          val = len(node_to_secondary[node.name])
2017
        elif field == "pip":
2018
          val = node.primary_ip
2019
        elif field == "sip":
2020
          val = node.secondary_ip
2021
        elif field == "tags":
2022
          val = list(node.GetTags())
2023
        elif field == "serial_no":
2024
          val = node.serial_no
2025
        elif field == "master_candidate":
2026
          val = node.master_candidate
2027
        elif field == "master":
2028
          val = node.name == master_node
2029
        elif field == "offline":
2030
          val = node.offline
2031
        elif field == "drained":
2032
          val = node.drained
2033
        elif self._FIELDS_DYNAMIC.Matches(field):
2034
          val = live_data[node.name].get(field, None)
2035
        else:
2036
          raise errors.ParameterError(field)
2037
        node_output.append(val)
2038
      output.append(node_output)
2039

    
2040
    return output
2041

    
2042

    
2043
class LUQueryNodeVolumes(NoHooksLU):
2044
  """Logical unit for getting volumes on node(s).
2045

2046
  """
2047
  _OP_REQP = ["nodes", "output_fields"]
2048
  REQ_BGL = False
2049
  _FIELDS_DYNAMIC = utils.FieldSet("phys", "vg", "name", "size", "instance")
2050
  _FIELDS_STATIC = utils.FieldSet("node")
2051

    
2052
  def ExpandNames(self):
2053
    _CheckOutputFields(static=self._FIELDS_STATIC,
2054
                       dynamic=self._FIELDS_DYNAMIC,
2055
                       selected=self.op.output_fields)
2056

    
2057
    self.needed_locks = {}
2058
    self.share_locks[locking.LEVEL_NODE] = 1
2059
    if not self.op.nodes:
2060
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
2061
    else:
2062
      self.needed_locks[locking.LEVEL_NODE] = \
2063
        _GetWantedNodes(self, self.op.nodes)
2064

    
2065
  def CheckPrereq(self):
2066
    """Check prerequisites.
2067

2068
    This checks that the fields required are valid output fields.
2069

2070
    """
2071
    self.nodes = self.acquired_locks[locking.LEVEL_NODE]
2072

    
2073
  def Exec(self, feedback_fn):
2074
    """Computes the list of nodes and their attributes.
2075

2076
    """
2077
    nodenames = self.nodes
2078
    volumes = self.rpc.call_node_volumes(nodenames)
2079

    
2080
    ilist = [self.cfg.GetInstanceInfo(iname) for iname
2081
             in self.cfg.GetInstanceList()]
2082

    
2083
    lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
2084

    
2085
    output = []
2086
    for node in nodenames:
2087
      if node not in volumes or volumes[node].failed or not volumes[node].data:
2088
        continue
2089

    
2090
      node_vols = volumes[node].data[:]
2091
      node_vols.sort(key=lambda vol: vol['dev'])
2092

    
2093
      for vol in node_vols:
2094
        node_output = []
2095
        for field in self.op.output_fields:
2096
          if field == "node":
2097
            val = node
2098
          elif field == "phys":
2099
            val = vol['dev']
2100
          elif field == "vg":
2101
            val = vol['vg']
2102
          elif field == "name":
2103
            val = vol['name']
2104
          elif field == "size":
2105
            val = int(float(vol['size']))
2106
          elif field == "instance":
2107
            for inst in ilist:
2108
              if node not in lv_by_node[inst]:
2109
                continue
2110
              if vol['name'] in lv_by_node[inst][node]:
2111
                val = inst.name
2112
                break
2113
            else:
2114
              val = '-'
2115
          else:
2116
            raise errors.ParameterError(field)
2117
          node_output.append(str(val))
2118

    
2119
        output.append(node_output)
2120

    
2121
    return output
2122

    
2123

    
2124
class LUAddNode(LogicalUnit):
2125
  """Logical unit for adding node to the cluster.
2126

2127
  """
2128
  HPATH = "node-add"
2129
  HTYPE = constants.HTYPE_NODE
2130
  _OP_REQP = ["node_name"]
2131

    
2132
  def BuildHooksEnv(self):
2133
    """Build hooks env.
2134

2135
    This will run on all nodes before, and on all nodes + the new node after.
2136

2137
    """
2138
    env = {
2139
      "OP_TARGET": self.op.node_name,
2140
      "NODE_NAME": self.op.node_name,
2141
      "NODE_PIP": self.op.primary_ip,
2142
      "NODE_SIP": self.op.secondary_ip,
2143
      }
2144
    nodes_0 = self.cfg.GetNodeList()
2145
    nodes_1 = nodes_0 + [self.op.node_name, ]
2146
    return env, nodes_0, nodes_1
2147

    
2148
  def CheckPrereq(self):
2149
    """Check prerequisites.
2150

2151
    This checks:
2152
     - the new node is not already in the config
2153
     - it is resolvable
2154
     - its parameters (single/dual homed) matches the cluster
2155

2156
    Any errors are signalled by raising errors.OpPrereqError.
2157

2158
    """
2159
    node_name = self.op.node_name
2160
    cfg = self.cfg
2161

    
2162
    dns_data = utils.HostInfo(node_name)
2163

    
2164
    node = dns_data.name
2165
    primary_ip = self.op.primary_ip = dns_data.ip
2166
    secondary_ip = getattr(self.op, "secondary_ip", None)
2167
    if secondary_ip is None:
2168
      secondary_ip = primary_ip
2169
    if not utils.IsValidIP(secondary_ip):
2170
      raise errors.OpPrereqError("Invalid secondary IP given")
2171
    self.op.secondary_ip = secondary_ip
2172

    
2173
    node_list = cfg.GetNodeList()
2174
    if not self.op.readd and node in node_list:
2175
      raise errors.OpPrereqError("Node %s is already in the configuration" %
2176
                                 node)
2177
    elif self.op.readd and node not in node_list:
2178
      raise errors.OpPrereqError("Node %s is not in the configuration" % node)
2179

    
2180
    for existing_node_name in node_list:
2181
      existing_node = cfg.GetNodeInfo(existing_node_name)
2182

    
2183
      if self.op.readd and node == existing_node_name:
2184
        if (existing_node.primary_ip != primary_ip or
2185
            existing_node.secondary_ip != secondary_ip):
2186
          raise errors.OpPrereqError("Readded node doesn't have the same IP"
2187
                                     " address configuration as before")
2188
        continue
2189

    
2190
      if (existing_node.primary_ip == primary_ip or
2191
          existing_node.secondary_ip == primary_ip or
2192
          existing_node.primary_ip == secondary_ip or
2193
          existing_node.secondary_ip == secondary_ip):
2194
        raise errors.OpPrereqError("New node ip address(es) conflict with"
2195
                                   " existing node %s" % existing_node.name)
2196

    
2197
    # check that the type of the node (single versus dual homed) is the
2198
    # same as for the master
2199
    myself = cfg.GetNodeInfo(self.cfg.GetMasterNode())
2200
    master_singlehomed = myself.secondary_ip == myself.primary_ip
2201
    newbie_singlehomed = secondary_ip == primary_ip
2202
    if master_singlehomed != newbie_singlehomed:
2203
      if master_singlehomed:
2204
        raise errors.OpPrereqError("The master has no private ip but the"
2205
                                   " new node has one")
2206
      else:
2207
        raise errors.OpPrereqError("The master has a private ip but the"
2208
                                   " new node doesn't have one")
2209

    
2210
    # checks reachablity
2211
    if not utils.TcpPing(primary_ip, constants.DEFAULT_NODED_PORT):
2212
      raise errors.OpPrereqError("Node not reachable by ping")
2213

    
2214
    if not newbie_singlehomed:
2215
      # check reachability from my secondary ip to newbie's secondary ip
2216
      if not utils.TcpPing(secondary_ip, constants.DEFAULT_NODED_PORT,
2217
                           source=myself.secondary_ip):
2218
        raise errors.OpPrereqError("Node secondary ip not reachable by TCP"
2219
                                   " based ping to noded port")
2220

    
2221
    cp_size = self.cfg.GetClusterInfo().candidate_pool_size
2222
    mc_now, _ = self.cfg.GetMasterCandidateStats()
2223
    master_candidate = mc_now < cp_size
2224

    
2225
    self.new_node = objects.Node(name=node,
2226
                                 primary_ip=primary_ip,
2227
                                 secondary_ip=secondary_ip,
2228
                                 master_candidate=master_candidate,
2229
                                 offline=False, drained=False)
2230

    
2231
  def Exec(self, feedback_fn):
2232
    """Adds the new node to the cluster.
2233

2234
    """
2235
    new_node = self.new_node
2236
    node = new_node.name
2237

    
2238
    # check connectivity
2239
    result = self.rpc.call_version([node])[node]
2240
    result.Raise()
2241
    if result.data:
2242
      if constants.PROTOCOL_VERSION == result.data:
2243
        logging.info("Communication to node %s fine, sw version %s match",
2244
                     node, result.data)
2245
      else:
2246
        raise errors.OpExecError("Version mismatch master version %s,"
2247
                                 " node version %s" %
2248
                                 (constants.PROTOCOL_VERSION, result.data))
2249
    else:
2250
      raise errors.OpExecError("Cannot get version from the new node")
2251

    
2252
    # setup ssh on node
2253
    logging.info("Copy ssh key to node %s", node)
2254
    priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
2255
    keyarray = []
2256
    keyfiles = [constants.SSH_HOST_DSA_PRIV, constants.SSH_HOST_DSA_PUB,
2257
                constants.SSH_HOST_RSA_PRIV, constants.SSH_HOST_RSA_PUB,
2258
                priv_key, pub_key]
2259

    
2260
    for i in keyfiles:
2261
      f = open(i, 'r')
2262
      try:
2263
        keyarray.append(f.read())
2264
      finally:
2265
        f.close()
2266

    
2267
    result = self.rpc.call_node_add(node, keyarray[0], keyarray[1],
2268
                                    keyarray[2],
2269
                                    keyarray[3], keyarray[4], keyarray[5])
2270

    
2271
    msg = result.RemoteFailMsg()
2272
    if msg:
2273
      raise errors.OpExecError("Cannot transfer ssh keys to the"
2274
                               " new node: %s" % msg)
2275

    
2276
    # Add node to our /etc/hosts, and add key to known_hosts
2277
    if self.cfg.GetClusterInfo().modify_etc_hosts:
2278
      utils.AddHostToEtcHosts(new_node.name)
2279

    
2280
    if new_node.secondary_ip != new_node.primary_ip:
2281
      result = self.rpc.call_node_has_ip_address(new_node.name,
2282
                                                 new_node.secondary_ip)
2283
      if result.failed or not result.data:
2284
        raise errors.OpExecError("Node claims it doesn't have the secondary ip"
2285
                                 " you gave (%s). Please fix and re-run this"
2286
                                 " command." % new_node.secondary_ip)
2287

    
2288
    node_verify_list = [self.cfg.GetMasterNode()]
2289
    node_verify_param = {
2290
      'nodelist': [node],
2291
      # TODO: do a node-net-test as well?
2292
    }
2293

    
2294
    result = self.rpc.call_node_verify(node_verify_list, node_verify_param,
2295
                                       self.cfg.GetClusterName())
2296
    for verifier in node_verify_list:
2297
      if result[verifier].failed or not result[verifier].data:
2298
        raise errors.OpExecError("Cannot communicate with %s's node daemon"
2299
                                 " for remote verification" % verifier)
2300
      if result[verifier].data['nodelist']:
2301
        for failed in result[verifier].data['nodelist']:
2302
          feedback_fn("ssh/hostname verification failed %s -> %s" %
2303
                      (verifier, result[verifier].data['nodelist'][failed]))
2304
        raise errors.OpExecError("ssh/hostname verification failed.")
2305

    
2306
    if self.op.readd:
2307
      _RedistributeAncillaryFiles(self)
2308
      self.context.ReaddNode(new_node)
2309
    else:
2310
      _RedistributeAncillaryFiles(self, additional_nodes=node)
2311
      self.context.AddNode(new_node)
2312

    
2313

    
2314
class LUSetNodeParams(LogicalUnit):
2315
  """Modifies the parameters of a node.
2316

2317
  """
2318
  HPATH = "node-modify"
2319
  HTYPE = constants.HTYPE_NODE
2320
  _OP_REQP = ["node_name"]
2321
  REQ_BGL = False
2322

    
2323
  def CheckArguments(self):
2324
    node_name = self.cfg.ExpandNodeName(self.op.node_name)
2325
    if node_name is None:
2326
      raise errors.OpPrereqError("Invalid node name '%s'" % self.op.node_name)
2327
    self.op.node_name = node_name
2328
    _CheckBooleanOpField(self.op, 'master_candidate')
2329
    _CheckBooleanOpField(self.op, 'offline')
2330
    _CheckBooleanOpField(self.op, 'drained')
2331
    all_mods = [self.op.offline, self.op.master_candidate, self.op.drained]
2332
    if all_mods.count(None) == 3:
2333
      raise errors.OpPrereqError("Please pass at least one modification")
2334
    if all_mods.count(True) > 1:
2335
      raise errors.OpPrereqError("Can't set the node into more than one"
2336
                                 " state at the same time")
2337

    
2338
  def ExpandNames(self):
2339
    self.needed_locks = {locking.LEVEL_NODE: self.op.node_name}
2340

    
2341
  def BuildHooksEnv(self):
2342
    """Build hooks env.
2343

2344
    This runs on the master node.
2345

2346
    """
2347
    env = {
2348
      "OP_TARGET": self.op.node_name,
2349
      "MASTER_CANDIDATE": str(self.op.master_candidate),
2350
      "OFFLINE": str(self.op.offline),
2351
      "DRAINED": str(self.op.drained),
2352
      }
2353
    nl = [self.cfg.GetMasterNode(),
2354
          self.op.node_name]
2355
    return env, nl, nl
2356

    
2357
  def CheckPrereq(self):
2358
    """Check prerequisites.
2359

2360
    This only checks the instance list against the existing names.
2361

2362
    """
2363
    node = self.node = self.cfg.GetNodeInfo(self.op.node_name)
2364

    
2365
    if ((self.op.master_candidate == False or self.op.offline == True or
2366
         self.op.drained == True) and node.master_candidate):
2367
      # we will demote the node from master_candidate
2368
      if self.op.node_name == self.cfg.GetMasterNode():
2369
        raise errors.OpPrereqError("The master node has to be a"
2370
                                   " master candidate, online and not drained")
2371
      cp_size = self.cfg.GetClusterInfo().candidate_pool_size
2372
      num_candidates, _ = self.cfg.GetMasterCandidateStats()
2373
      if num_candidates <= cp_size:
2374
        msg = ("Not enough master candidates (desired"
2375
               " %d, new value will be %d)" % (cp_size, num_candidates-1))
2376
        if self.op.force:
2377
          self.LogWarning(msg)
2378
        else:
2379
          raise errors.OpPrereqError(msg)
2380

    
2381
    if (self.op.master_candidate == True and
2382
        ((node.offline and not self.op.offline == False) or
2383
         (node.drained and not self.op.drained == False))):
2384
      raise errors.OpPrereqError("Node '%s' is offline or drained, can't set"
2385
                                 " to master_candidate" % node.name)
2386

    
2387
    return
2388

    
2389
  def Exec(self, feedback_fn):
2390
    """Modifies a node.
2391

2392
    """
2393
    node = self.node
2394

    
2395
    result = []
2396
    changed_mc = False
2397

    
2398
    if self.op.offline is not None:
2399
      node.offline = self.op.offline
2400
      result.append(("offline", str(self.op.offline)))
2401
      if self.op.offline == True:
2402
        if node.master_candidate:
2403
          node.master_candidate = False
2404
          changed_mc = True
2405
          result.append(("master_candidate", "auto-demotion due to offline"))
2406
        if node.drained:
2407
          node.drained = False
2408
          result.append(("drained", "clear drained status due to offline"))
2409

    
2410
    if self.op.master_candidate is not None:
2411
      node.master_candidate = self.op.master_candidate
2412
      changed_mc = True
2413
      result.append(("master_candidate", str(self.op.master_candidate)))
2414
      if self.op.master_candidate == False:
2415
        rrc = self.rpc.call_node_demote_from_mc(node.name)
2416
        msg = rrc.RemoteFailMsg()
2417
        if msg:
2418
          self.LogWarning("Node failed to demote itself: %s" % msg)
2419

    
2420
    if self.op.drained is not None:
2421
      node.drained = self.op.drained
2422
      result.append(("drained", str(self.op.drained)))
2423
      if self.op.drained == True:
2424
        if node.master_candidate:
2425
          node.master_candidate = False
2426
          changed_mc = True
2427
          result.append(("master_candidate", "auto-demotion due to drain"))
2428
        if node.offline:
2429
          node.offline = False
2430
          result.append(("offline", "clear offline status due to drain"))
2431

    
2432
    # this will trigger configuration file update, if needed
2433
    self.cfg.Update(node)
2434
    # this will trigger job queue propagation or cleanup
2435
    if changed_mc:
2436
      self.context.ReaddNode(node)
2437

    
2438
    return result
2439

    
2440

    
2441
class LUPowercycleNode(NoHooksLU):
2442
  """Powercycles a node.
2443

2444
  """
2445
  _OP_REQP = ["node_name", "force"]
2446
  REQ_BGL = False
2447

    
2448
  def CheckArguments(self):
2449
    node_name = self.cfg.ExpandNodeName(self.op.node_name)
2450
    if node_name is None:
2451
      raise errors.OpPrereqError("Invalid node name '%s'" % self.op.node_name)
2452
    self.op.node_name = node_name
2453
    if node_name == self.cfg.GetMasterNode() and not self.op.force:
2454
      raise errors.OpPrereqError("The node is the master and the force"
2455
                                 " parameter was not set")
2456

    
2457
  def ExpandNames(self):
2458
    """Locking for PowercycleNode.
2459

2460
    This is a last-resource option and shouldn't block on other
2461
    jobs. Therefore, we grab no locks.
2462

2463
    """
2464
    self.needed_locks = {}
2465

    
2466
  def CheckPrereq(self):
2467
    """Check prerequisites.
2468

2469
    This LU has no prereqs.
2470

2471
    """
2472
    pass
2473

    
2474
  def Exec(self, feedback_fn):
2475
    """Reboots a node.
2476

2477
    """
2478
    result = self.rpc.call_node_powercycle(self.op.node_name,
2479
                                           self.cfg.GetHypervisorType())
2480
    msg = result.RemoteFailMsg()
2481
    if msg:
2482
      raise errors.OpExecError("Failed to schedule the reboot: %s" % msg)
2483
    return result.payload
2484

    
2485

    
2486
class LUQueryClusterInfo(NoHooksLU):
2487
  """Query cluster configuration.
2488

2489
  """
2490
  _OP_REQP = []
2491
  REQ_BGL = False
2492

    
2493
  def ExpandNames(self):
2494
    self.needed_locks = {}
2495

    
2496
  def CheckPrereq(self):
2497
    """No prerequsites needed for this LU.
2498

2499
    """
2500
    pass
2501

    
2502
  def Exec(self, feedback_fn):
2503
    """Return cluster config.
2504

2505
    """
2506
    cluster = self.cfg.GetClusterInfo()
2507
    result = {
2508
      "software_version": constants.RELEASE_VERSION,
2509
      "protocol_version": constants.PROTOCOL_VERSION,
2510
      "config_version": constants.CONFIG_VERSION,
2511
      "os_api_version": constants.OS_API_VERSION,
2512
      "export_version": constants.EXPORT_VERSION,
2513
      "architecture": (platform.architecture()[0], platform.machine()),
2514
      "name": cluster.cluster_name,
2515
      "master": cluster.master_node,
2516
      "default_hypervisor": cluster.default_hypervisor,
2517
      "enabled_hypervisors": cluster.enabled_hypervisors,
2518
      "hvparams": dict([(hypervisor, cluster.hvparams[hypervisor])
2519
                        for hypervisor in cluster.enabled_hypervisors]),
2520
      "beparams": cluster.beparams,
2521
      "candidate_pool_size": cluster.candidate_pool_size,
2522
      "default_bridge": cluster.default_bridge,
2523
      "master_netdev": cluster.master_netdev,
2524
      "volume_group_name": cluster.volume_group_name,
2525
      "file_storage_dir": cluster.file_storage_dir,
2526
      }
2527

    
2528
    return result
2529

    
2530

    
2531
class LUQueryConfigValues(NoHooksLU):
2532
  """Return configuration values.
2533

2534
  """
2535
  _OP_REQP = []
2536
  REQ_BGL = False
2537
  _FIELDS_DYNAMIC = utils.FieldSet()
2538
  _FIELDS_STATIC = utils.FieldSet("cluster_name", "master_node", "drain_flag")
2539

    
2540
  def ExpandNames(self):
2541
    self.needed_locks = {}
2542

    
2543
    _CheckOutputFields(static=self._FIELDS_STATIC,
2544
                       dynamic=self._FIELDS_DYNAMIC,
2545
                       selected=self.op.output_fields)
2546

    
2547
  def CheckPrereq(self):
2548
    """No prerequisites.
2549

2550
    """
2551
    pass
2552

    
2553
  def Exec(self, feedback_fn):
2554
    """Dump a representation of the cluster config to the standard output.
2555

2556
    """
2557
    values = []
2558
    for field in self.op.output_fields:
2559
      if field == "cluster_name":
2560
        entry = self.cfg.GetClusterName()
2561
      elif field == "master_node":
2562
        entry = self.cfg.GetMasterNode()
2563
      elif field == "drain_flag":
2564
        entry = os.path.exists(constants.JOB_QUEUE_DRAIN_FILE)
2565
      else:
2566
        raise errors.ParameterError(field)
2567
      values.append(entry)
2568
    return values
2569

    
2570

    
2571
class LUActivateInstanceDisks(NoHooksLU):
2572
  """Bring up an instance's disks.
2573

2574
  """
2575
  _OP_REQP = ["instance_name"]
2576
  REQ_BGL = False
2577

    
2578
  def ExpandNames(self):
2579
    self._ExpandAndLockInstance()
2580
    self.needed_locks[locking.LEVEL_NODE] = []
2581
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2582

    
2583
  def DeclareLocks(self, level):
2584
    if level == locking.LEVEL_NODE:
2585
      self._LockInstancesNodes()
2586

    
2587
  def CheckPrereq(self):
2588
    """Check prerequisites.
2589

2590
    This checks that the instance is in the cluster.
2591

2592
    """
2593
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2594
    assert self.instance is not None, \
2595
      "Cannot retrieve locked instance %s" % self.op.instance_name
2596
    _CheckNodeOnline(self, self.instance.primary_node)
2597

    
2598
  def Exec(self, feedback_fn):
2599
    """Activate the disks.
2600

2601
    """
2602
    disks_ok, disks_info = _AssembleInstanceDisks(self, self.instance)
2603
    if not disks_ok:
2604
      raise errors.OpExecError("Cannot activate block devices")
2605

    
2606
    return disks_info
2607

    
2608

    
2609
def _AssembleInstanceDisks(lu, instance, ignore_secondaries=False):
2610
  """Prepare the block devices for an instance.
2611

2612
  This sets up the block devices on all nodes.
2613

2614
  @type lu: L{LogicalUnit}
2615
  @param lu: the logical unit on whose behalf we execute
2616
  @type instance: L{objects.Instance}
2617
  @param instance: the instance for whose disks we assemble
2618
  @type ignore_secondaries: boolean
2619
  @param ignore_secondaries: if true, errors on secondary nodes
2620
      won't result in an error return from the function
2621
  @return: False if the operation failed, otherwise a list of
2622
      (host, instance_visible_name, node_visible_name)
2623
      with the mapping from node devices to instance devices
2624

2625
  """
2626
  device_info = []
2627
  disks_ok = True
2628
  iname = instance.name
2629
  # With the two passes mechanism we try to reduce the window of
2630
  # opportunity for the race condition of switching DRBD to primary
2631
  # before handshaking occured, but we do not eliminate it
2632

    
2633
  # The proper fix would be to wait (with some limits) until the
2634
  # connection has been made and drbd transitions from WFConnection
2635
  # into any other network-connected state (Connected, SyncTarget,
2636
  # SyncSource, etc.)
2637

    
2638
  # 1st pass, assemble on all nodes in secondary mode
2639
  for inst_disk in instance.disks:
2640
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
2641
      lu.cfg.SetDiskID(node_disk, node)
2642
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, False)
2643
      msg = result.RemoteFailMsg()
2644
      if msg:
2645
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
2646
                           " (is_primary=False, pass=1): %s",
2647
                           inst_disk.iv_name, node, msg)
2648
        if not ignore_secondaries:
2649
          disks_ok = False
2650

    
2651
  # FIXME: race condition on drbd migration to primary
2652

    
2653
  # 2nd pass, do only the primary node
2654
  for inst_disk in instance.disks:
2655
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
2656
      if node != instance.primary_node:
2657
        continue
2658
      lu.cfg.SetDiskID(node_disk, node)
2659
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, True)
2660
      msg = result.RemoteFailMsg()
2661
      if msg:
2662
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
2663
                           " (is_primary=True, pass=2): %s",
2664
                           inst_disk.iv_name, node, msg)
2665
        disks_ok = False
2666
    device_info.append((instance.primary_node, inst_disk.iv_name,
2667
                        result.payload))
2668

    
2669
  # leave the disks configured for the primary node
2670
  # this is a workaround that would be fixed better by
2671
  # improving the logical/physical id handling
2672
  for disk in instance.disks:
2673
    lu.cfg.SetDiskID(disk, instance.primary_node)
2674

    
2675
  return disks_ok, device_info
2676

    
2677

    
2678
def _StartInstanceDisks(lu, instance, force):
2679
  """Start the disks of an instance.
2680

2681
  """
2682
  disks_ok, dummy = _AssembleInstanceDisks(lu, instance,
2683
                                           ignore_secondaries=force)
2684
  if not disks_ok:
2685
    _ShutdownInstanceDisks(lu, instance)
2686
    if force is not None and not force:
2687
      lu.proc.LogWarning("", hint="If the message above refers to a"
2688
                         " secondary node,"
2689
                         " you can retry the operation using '--force'.")
2690
    raise errors.OpExecError("Disk consistency error")
2691

    
2692

    
2693
class LUDeactivateInstanceDisks(NoHooksLU):
2694
  """Shutdown an instance's disks.
2695

2696
  """
2697
  _OP_REQP = ["instance_name"]
2698
  REQ_BGL = False
2699

    
2700
  def ExpandNames(self):
2701
    self._ExpandAndLockInstance()
2702
    self.needed_locks[locking.LEVEL_NODE] = []
2703
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2704

    
2705
  def DeclareLocks(self, level):
2706
    if level == locking.LEVEL_NODE:
2707
      self._LockInstancesNodes()
2708

    
2709
  def CheckPrereq(self):
2710
    """Check prerequisites.
2711

2712
    This checks that the instance is in the cluster.
2713

2714
    """
2715
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2716
    assert self.instance is not None, \
2717
      "Cannot retrieve locked instance %s" % self.op.instance_name
2718

    
2719
  def Exec(self, feedback_fn):
2720
    """Deactivate the disks
2721

2722
    """
2723
    instance = self.instance
2724
    _SafeShutdownInstanceDisks(self, instance)
2725

    
2726

    
2727
def _SafeShutdownInstanceDisks(lu, instance):
2728
  """Shutdown block devices of an instance.
2729

2730
  This function checks if an instance is running, before calling
2731
  _ShutdownInstanceDisks.
2732

2733
  """
2734
  ins_l = lu.rpc.call_instance_list([instance.primary_node],
2735
                                      [instance.hypervisor])
2736
  ins_l = ins_l[instance.primary_node]
2737
  if ins_l.failed or not isinstance(ins_l.data, list):
2738
    raise errors.OpExecError("Can't contact node '%s'" %
2739
                             instance.primary_node)
2740

    
2741
  if instance.name in ins_l.data:
2742
    raise errors.OpExecError("Instance is running, can't shutdown"
2743
                             " block devices.")
2744

    
2745
  _ShutdownInstanceDisks(lu, instance)
2746

    
2747

    
2748
def _ShutdownInstanceDisks(lu, instance, ignore_primary=False):
2749
  """Shutdown block devices of an instance.
2750

2751
  This does the shutdown on all nodes of the instance.
2752

2753
  If the ignore_primary is false, errors on the primary node are
2754
  ignored.
2755

2756
  """
2757
  all_result = True
2758
  for disk in instance.disks:
2759
    for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
2760
      lu.cfg.SetDiskID(top_disk, node)
2761
      result = lu.rpc.call_blockdev_shutdown(node, top_disk)
2762
      msg = result.RemoteFailMsg()
2763
      if msg:
2764
        lu.LogWarning("Could not shutdown block device %s on node %s: %s",
2765
                      disk.iv_name, node, msg)
2766
        if not ignore_primary or node != instance.primary_node:
2767
          all_result = False
2768
  return all_result
2769

    
2770

    
2771
def _CheckNodeFreeMemory(lu, node, reason, requested, hypervisor_name):
2772
  """Checks if a node has enough free memory.
2773

2774
  This function check if a given node has the needed amount of free
2775
  memory. In case the node has less memory or we cannot get the
2776
  information from the node, this function raise an OpPrereqError
2777
  exception.
2778

2779
  @type lu: C{LogicalUnit}
2780
  @param lu: a logical unit from which we get configuration data
2781
  @type node: C{str}
2782
  @param node: the node to check
2783
  @type reason: C{str}
2784
  @param reason: string to use in the error message
2785
  @type requested: C{int}
2786
  @param requested: the amount of memory in MiB to check for
2787
  @type hypervisor_name: C{str}
2788
  @param hypervisor_name: the hypervisor to ask for memory stats
2789
  @raise errors.OpPrereqError: if the node doesn't have enough memory, or
2790
      we cannot check the node
2791

2792
  """
2793
  nodeinfo = lu.rpc.call_node_info([node], lu.cfg.GetVGName(), hypervisor_name)
2794
  nodeinfo[node].Raise()
2795
  free_mem = nodeinfo[node].data.get('memory_free')
2796
  if not isinstance(free_mem, int):
2797
    raise errors.OpPrereqError("Can't compute free memory on node %s, result"
2798
                             " was '%s'" % (node, free_mem))
2799
  if requested > free_mem:
2800
    raise errors.OpPrereqError("Not enough memory on node %s for %s:"
2801
                             " needed %s MiB, available %s MiB" %
2802
                             (node, reason, requested, free_mem))
2803

    
2804

    
2805
class LUStartupInstance(LogicalUnit):
2806
  """Starts an instance.
2807

2808
  """
2809
  HPATH = "instance-start"
2810
  HTYPE = constants.HTYPE_INSTANCE
2811
  _OP_REQP = ["instance_name", "force"]
2812
  REQ_BGL = False
2813

    
2814
  def ExpandNames(self):
2815
    self._ExpandAndLockInstance()
2816

    
2817
  def BuildHooksEnv(self):
2818
    """Build hooks env.
2819

2820
    This runs on master, primary and secondary nodes of the instance.
2821

2822
    """
2823
    env = {
2824
      "FORCE": self.op.force,
2825
      }
2826
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
2827
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2828
    return env, nl, nl
2829

    
2830
  def CheckPrereq(self):
2831
    """Check prerequisites.
2832

2833
    This checks that the instance is in the cluster.
2834

2835
    """
2836
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2837
    assert self.instance is not None, \
2838
      "Cannot retrieve locked instance %s" % self.op.instance_name
2839

    
2840
    # extra beparams
2841
    self.beparams = getattr(self.op, "beparams", {})
2842
    if self.beparams:
2843
      if not isinstance(self.beparams, dict):
2844
        raise errors.OpPrereqError("Invalid beparams passed: %s, expected"
2845
                                   " dict" % (type(self.beparams), ))
2846
      # fill the beparams dict
2847
      utils.ForceDictType(self.beparams, constants.BES_PARAMETER_TYPES)
2848
      self.op.beparams = self.beparams
2849

    
2850
    # extra hvparams
2851
    self.hvparams = getattr(self.op, "hvparams", {})
2852
    if self.hvparams:
2853
      if not isinstance(self.hvparams, dict):
2854
        raise errors.OpPrereqError("Invalid hvparams passed: %s, expected"
2855
                                   " dict" % (type(self.hvparams), ))
2856

    
2857
      # check hypervisor parameter syntax (locally)
2858
      cluster = self.cfg.GetClusterInfo()
2859
      utils.ForceDictType(self.hvparams, constants.HVS_PARAMETER_TYPES)
2860
      filled_hvp = objects.FillDict(cluster.hvparams[instance.hypervisor],
2861
                                    instance.hvparams)
2862
      filled_hvp.update(self.hvparams)
2863
      hv_type = hypervisor.GetHypervisor(instance.hypervisor)
2864
      hv_type.CheckParameterSyntax(filled_hvp)
2865
      _CheckHVParams(self, instance.all_nodes, instance.hypervisor, filled_hvp)
2866
      self.op.hvparams = self.hvparams
2867

    
2868
    _CheckNodeOnline(self, instance.primary_node)
2869

    
2870
    bep = self.cfg.GetClusterInfo().FillBE(instance)
2871
    # check bridges existance
2872
    _CheckInstanceBridgesExist(self, instance)
2873

    
2874
    remote_info = self.rpc.call_instance_info(instance.primary_node,
2875
                                              instance.name,
2876
                                              instance.hypervisor)
2877
    remote_info.Raise()
2878
    if not remote_info.data:
2879
      _CheckNodeFreeMemory(self, instance.primary_node,
2880
                           "starting instance %s" % instance.name,
2881
                           bep[constants.BE_MEMORY], instance.hypervisor)
2882

    
2883
  def Exec(self, feedback_fn):
2884
    """Start the instance.
2885

2886
    """
2887
    instance = self.instance
2888
    force = self.op.force
2889

    
2890
    self.cfg.MarkInstanceUp(instance.name)
2891

    
2892
    node_current = instance.primary_node
2893

    
2894
    _StartInstanceDisks(self, instance, force)
2895

    
2896
    result = self.rpc.call_instance_start(node_current, instance,
2897
                                          self.hvparams, self.beparams)
2898
    msg = result.RemoteFailMsg()
2899
    if msg:
2900
      _ShutdownInstanceDisks(self, instance)
2901
      raise errors.OpExecError("Could not start instance: %s" % msg)
2902

    
2903

    
2904
class LURebootInstance(LogicalUnit):
2905
  """Reboot an instance.
2906

2907
  """
2908
  HPATH = "instance-reboot"
2909
  HTYPE = constants.HTYPE_INSTANCE
2910
  _OP_REQP = ["instance_name", "ignore_secondaries", "reboot_type"]
2911
  REQ_BGL = False
2912

    
2913
  def ExpandNames(self):
2914
    if self.op.reboot_type not in [constants.INSTANCE_REBOOT_SOFT,
2915
                                   constants.INSTANCE_REBOOT_HARD,
2916
                                   constants.INSTANCE_REBOOT_FULL]:
2917
      raise errors.ParameterError("reboot type not in [%s, %s, %s]" %
2918
                                  (constants.INSTANCE_REBOOT_SOFT,
2919
                                   constants.INSTANCE_REBOOT_HARD,
2920
                                   constants.INSTANCE_REBOOT_FULL))
2921
    self._ExpandAndLockInstance()
2922

    
2923
  def BuildHooksEnv(self):
2924
    """Build hooks env.
2925

2926
    This runs on master, primary and secondary nodes of the instance.
2927

2928
    """
2929
    env = {
2930
      "IGNORE_SECONDARIES": self.op.ignore_secondaries,
2931
      "REBOOT_TYPE": self.op.reboot_type,
2932
      }
2933
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
2934
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2935
    return env, nl, nl
2936

    
2937
  def CheckPrereq(self):
2938
    """Check prerequisites.
2939

2940
    This checks that the instance is in the cluster.
2941

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

    
2947
    _CheckNodeOnline(self, instance.primary_node)
2948

    
2949
    # check bridges existance
2950
    _CheckInstanceBridgesExist(self, instance)
2951

    
2952
  def Exec(self, feedback_fn):
2953
    """Reboot the instance.
2954

2955
    """
2956
    instance = self.instance
2957
    ignore_secondaries = self.op.ignore_secondaries
2958
    reboot_type = self.op.reboot_type
2959

    
2960
    node_current = instance.primary_node
2961

    
2962
    if reboot_type in [constants.INSTANCE_REBOOT_SOFT,
2963
                       constants.INSTANCE_REBOOT_HARD]:
2964
      for disk in instance.disks:
2965
        self.cfg.SetDiskID(disk, node_current)
2966
      result = self.rpc.call_instance_reboot(node_current, instance,
2967
                                             reboot_type)
2968
      msg = result.RemoteFailMsg()
2969
      if msg:
2970
        raise errors.OpExecError("Could not reboot instance: %s" % msg)
2971
    else:
2972
      result = self.rpc.call_instance_shutdown(node_current, instance)
2973
      msg = result.RemoteFailMsg()
2974
      if msg:
2975
        raise errors.OpExecError("Could not shutdown instance for"
2976
                                 " full reboot: %s" % msg)
2977
      _ShutdownInstanceDisks(self, instance)
2978
      _StartInstanceDisks(self, instance, ignore_secondaries)
2979
      result = self.rpc.call_instance_start(node_current, instance, None, None)
2980
      msg = result.RemoteFailMsg()
2981
      if msg:
2982
        _ShutdownInstanceDisks(self, instance)
2983
        raise errors.OpExecError("Could not start instance for"
2984
                                 " full reboot: %s" % msg)
2985

    
2986
    self.cfg.MarkInstanceUp(instance.name)
2987

    
2988

    
2989
class LUShutdownInstance(LogicalUnit):
2990
  """Shutdown an instance.
2991

2992
  """
2993
  HPATH = "instance-stop"
2994
  HTYPE = constants.HTYPE_INSTANCE
2995
  _OP_REQP = ["instance_name"]
2996
  REQ_BGL = False
2997

    
2998
  def ExpandNames(self):
2999
    self._ExpandAndLockInstance()
3000

    
3001
  def BuildHooksEnv(self):
3002
    """Build hooks env.
3003

3004
    This runs on master, primary and secondary nodes of the instance.
3005

3006
    """
3007
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3008
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3009
    return env, nl, nl
3010

    
3011
  def CheckPrereq(self):
3012
    """Check prerequisites.
3013

3014
    This checks that the instance is in the cluster.
3015

3016
    """
3017
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3018
    assert self.instance is not None, \
3019
      "Cannot retrieve locked instance %s" % self.op.instance_name
3020
    _CheckNodeOnline(self, self.instance.primary_node)
3021

    
3022
  def Exec(self, feedback_fn):
3023
    """Shutdown the instance.
3024

3025
    """
3026
    instance = self.instance
3027
    node_current = instance.primary_node
3028
    self.cfg.MarkInstanceDown(instance.name)
3029
    result = self.rpc.call_instance_shutdown(node_current, instance)
3030
    msg = result.RemoteFailMsg()
3031
    if msg:
3032
      self.proc.LogWarning("Could not shutdown instance: %s" % msg)
3033

    
3034
    _ShutdownInstanceDisks(self, instance)
3035

    
3036

    
3037
class LUReinstallInstance(LogicalUnit):
3038
  """Reinstall an instance.
3039

3040
  """
3041
  HPATH = "instance-reinstall"
3042
  HTYPE = constants.HTYPE_INSTANCE
3043
  _OP_REQP = ["instance_name"]
3044
  REQ_BGL = False
3045

    
3046
  def ExpandNames(self):
3047
    self._ExpandAndLockInstance()
3048

    
3049
  def BuildHooksEnv(self):
3050
    """Build hooks env.
3051

3052
    This runs on master, primary and secondary nodes of the instance.
3053

3054
    """
3055
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3056
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3057
    return env, nl, nl
3058

    
3059
  def CheckPrereq(self):
3060
    """Check prerequisites.
3061

3062
    This checks that the instance is in the cluster and is not running.
3063

3064
    """
3065
    instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3066
    assert instance is not None, \
3067
      "Cannot retrieve locked instance %s" % self.op.instance_name
3068
    _CheckNodeOnline(self, instance.primary_node)
3069

    
3070
    if instance.disk_template == constants.DT_DISKLESS:
3071
      raise errors.OpPrereqError("Instance '%s' has no disks" %
3072
                                 self.op.instance_name)
3073
    if instance.admin_up:
3074
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
3075
                                 self.op.instance_name)
3076
    remote_info = self.rpc.call_instance_info(instance.primary_node,
3077
                                              instance.name,
3078
                                              instance.hypervisor)
3079
    remote_info.Raise()
3080
    if remote_info.data:
3081
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
3082
                                 (self.op.instance_name,
3083
                                  instance.primary_node))
3084

    
3085
    self.op.os_type = getattr(self.op, "os_type", None)
3086
    if self.op.os_type is not None:
3087
      # OS verification
3088
      pnode = self.cfg.GetNodeInfo(
3089
        self.cfg.ExpandNodeName(instance.primary_node))
3090
      if pnode is None:
3091
        raise errors.OpPrereqError("Primary node '%s' is unknown" %
3092
                                   self.op.pnode)
3093
      result = self.rpc.call_os_get(pnode.name, self.op.os_type)
3094
      result.Raise()
3095
      if not isinstance(result.data, objects.OS):
3096
        raise errors.OpPrereqError("OS '%s' not in supported OS list for"
3097
                                   " primary node"  % self.op.os_type)
3098

    
3099
    self.instance = instance
3100

    
3101
  def Exec(self, feedback_fn):
3102
    """Reinstall the instance.
3103

3104
    """
3105
    inst = self.instance
3106

    
3107
    if self.op.os_type is not None:
3108
      feedback_fn("Changing OS to '%s'..." % self.op.os_type)
3109
      inst.os = self.op.os_type
3110
      self.cfg.Update(inst)
3111

    
3112
    _StartInstanceDisks(self, inst, None)
3113
    try:
3114
      feedback_fn("Running the instance OS create scripts...")
3115
      result = self.rpc.call_instance_os_add(inst.primary_node, inst, True)
3116
      msg = result.RemoteFailMsg()
3117
      if msg:
3118
        raise errors.OpExecError("Could not install OS for instance %s"
3119
                                 " on node %s: %s" %
3120
                                 (inst.name, inst.primary_node, msg))
3121
    finally:
3122
      _ShutdownInstanceDisks(self, inst)
3123

    
3124

    
3125
class LURenameInstance(LogicalUnit):
3126
  """Rename an instance.
3127

3128
  """
3129
  HPATH = "instance-rename"
3130
  HTYPE = constants.HTYPE_INSTANCE
3131
  _OP_REQP = ["instance_name", "new_name"]
3132

    
3133
  def BuildHooksEnv(self):
3134
    """Build hooks env.
3135

3136
    This runs on master, primary and secondary nodes of the instance.
3137

3138
    """
3139
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3140
    env["INSTANCE_NEW_NAME"] = self.op.new_name
3141
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3142
    return env, nl, nl
3143

    
3144
  def CheckPrereq(self):
3145
    """Check prerequisites.
3146

3147
    This checks that the instance is in the cluster and is not running.
3148

3149
    """
3150
    instance = self.cfg.GetInstanceInfo(
3151
      self.cfg.ExpandInstanceName(self.op.instance_name))
3152
    if instance is None:
3153
      raise errors.OpPrereqError("Instance '%s' not known" %
3154
                                 self.op.instance_name)
3155
    _CheckNodeOnline(self, instance.primary_node)
3156

    
3157
    if instance.admin_up:
3158
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
3159
                                 self.op.instance_name)
3160
    remote_info = self.rpc.call_instance_info(instance.primary_node,
3161
                                              instance.name,
3162
                                              instance.hypervisor)
3163
    remote_info.Raise()
3164
    if remote_info.data:
3165
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
3166
                                 (self.op.instance_name,
3167
                                  instance.primary_node))
3168
    self.instance = instance
3169

    
3170
    # new name verification
3171
    name_info = utils.HostInfo(self.op.new_name)
3172

    
3173
    self.op.new_name = new_name = name_info.name
3174
    instance_list = self.cfg.GetInstanceList()
3175
    if new_name in instance_list:
3176
      raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
3177
                                 new_name)
3178

    
3179
    if not getattr(self.op, "ignore_ip", False):
3180
      if utils.TcpPing(name_info.ip, constants.DEFAULT_NODED_PORT):
3181
        raise errors.OpPrereqError("IP %s of instance %s already in use" %
3182
                                   (name_info.ip, new_name))
3183

    
3184

    
3185
  def Exec(self, feedback_fn):
3186
    """Reinstall the instance.
3187

3188
    """
3189
    inst = self.instance
3190
    old_name = inst.name
3191

    
3192
    if inst.disk_template == constants.DT_FILE:
3193
      old_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
3194

    
3195
    self.cfg.RenameInstance(inst.name, self.op.new_name)
3196
    # Change the instance lock. This is definitely safe while we hold the BGL
3197
    self.context.glm.remove(locking.LEVEL_INSTANCE, old_name)
3198
    self.context.glm.add(locking.LEVEL_INSTANCE, self.op.new_name)
3199

    
3200
    # re-read the instance from the configuration after rename
3201
    inst = self.cfg.GetInstanceInfo(self.op.new_name)
3202

    
3203
    if inst.disk_template == constants.DT_FILE:
3204
      new_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
3205
      result = self.rpc.call_file_storage_dir_rename(inst.primary_node,
3206
                                                     old_file_storage_dir,
3207
                                                     new_file_storage_dir)
3208
      result.Raise()
3209
      if not result.data:
3210
        raise errors.OpExecError("Could not connect to node '%s' to rename"
3211
                                 " directory '%s' to '%s' (but the instance"
3212
                                 " has been renamed in Ganeti)" % (
3213
                                 inst.primary_node, old_file_storage_dir,
3214
                                 new_file_storage_dir))
3215

    
3216
      if not result.data[0]:
3217
        raise errors.OpExecError("Could not rename directory '%s' to '%s'"
3218
                                 " (but the instance has been renamed in"
3219
                                 " Ganeti)" % (old_file_storage_dir,
3220
                                               new_file_storage_dir))
3221

    
3222
    _StartInstanceDisks(self, inst, None)
3223
    try:
3224
      result = self.rpc.call_instance_run_rename(inst.primary_node, inst,
3225
                                                 old_name)
3226
      msg = result.RemoteFailMsg()
3227
      if msg:
3228
        msg = ("Could not run OS rename script for instance %s on node %s"
3229
               " (but the instance has been renamed in Ganeti): %s" %
3230
               (inst.name, inst.primary_node, msg))
3231
        self.proc.LogWarning(msg)
3232
    finally:
3233
      _ShutdownInstanceDisks(self, inst)
3234

    
3235

    
3236
class LURemoveInstance(LogicalUnit):
3237
  """Remove an instance.
3238

3239
  """
3240
  HPATH = "instance-remove"
3241
  HTYPE = constants.HTYPE_INSTANCE
3242
  _OP_REQP = ["instance_name", "ignore_failures"]
3243
  REQ_BGL = False
3244

    
3245
  def ExpandNames(self):
3246
    self._ExpandAndLockInstance()
3247
    self.needed_locks[locking.LEVEL_NODE] = []
3248
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3249

    
3250
  def DeclareLocks(self, level):
3251
    if level == locking.LEVEL_NODE:
3252
      self._LockInstancesNodes()
3253

    
3254
  def BuildHooksEnv(self):
3255
    """Build hooks env.
3256

3257
    This runs on master, primary and secondary nodes of the instance.
3258

3259
    """
3260
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3261
    nl = [self.cfg.GetMasterNode()]
3262
    return env, nl, nl
3263

    
3264
  def CheckPrereq(self):
3265
    """Check prerequisites.
3266

3267
    This checks that the instance is in the cluster.
3268

3269
    """
3270
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3271
    assert self.instance is not None, \
3272
      "Cannot retrieve locked instance %s" % self.op.instance_name
3273

    
3274
  def Exec(self, feedback_fn):
3275
    """Remove the instance.
3276

3277
    """
3278
    instance = self.instance
3279
    logging.info("Shutting down instance %s on node %s",
3280
                 instance.name, instance.primary_node)
3281

    
3282
    result = self.rpc.call_instance_shutdown(instance.primary_node, instance)
3283
    msg = result.RemoteFailMsg()
3284
    if msg:
3285
      if self.op.ignore_failures:
3286
        feedback_fn("Warning: can't shutdown instance: %s" % msg)
3287
      else:
3288
        raise errors.OpExecError("Could not shutdown instance %s on"
3289
                                 " node %s: %s" %
3290
                                 (instance.name, instance.primary_node, msg))
3291

    
3292
    logging.info("Removing block devices for instance %s", instance.name)
3293

    
3294
    if not _RemoveDisks(self, instance):
3295
      if self.op.ignore_failures:
3296
        feedback_fn("Warning: can't remove instance's disks")
3297
      else:
3298
        raise errors.OpExecError("Can't remove instance's disks")
3299

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

    
3302
    self.cfg.RemoveInstance(instance.name)
3303
    self.remove_locks[locking.LEVEL_INSTANCE] = instance.name
3304

    
3305

    
3306
class LUQueryInstances(NoHooksLU):
3307
  """Logical unit for querying instances.
3308

3309
  """
3310
  _OP_REQP = ["output_fields", "names", "use_locking"]
3311
  REQ_BGL = False
3312
  _FIELDS_STATIC = utils.FieldSet(*["name", "os", "pnode", "snodes",
3313
                                    "admin_state",
3314
                                    "disk_template", "ip", "mac", "bridge",
3315
                                    "sda_size", "sdb_size", "vcpus", "tags",
3316
                                    "network_port", "beparams",
3317
                                    r"(disk)\.(size)/([0-9]+)",
3318
                                    r"(disk)\.(sizes)", "disk_usage",
3319
                                    r"(nic)\.(mac|ip|bridge)/([0-9]+)",
3320
                                    r"(nic)\.(macs|ips|bridges)",
3321
                                    r"(disk|nic)\.(count)",
3322
                                    "serial_no", "hypervisor", "hvparams",] +
3323
                                  ["hv/%s" % name
3324
                                   for name in constants.HVS_PARAMETERS] +
3325
                                  ["be/%s" % name
3326
                                   for name in constants.BES_PARAMETERS])
3327
  _FIELDS_DYNAMIC = utils.FieldSet("oper_state", "oper_ram", "status")
3328

    
3329

    
3330
  def ExpandNames(self):
3331
    _CheckOutputFields(static=self._FIELDS_STATIC,
3332
                       dynamic=self._FIELDS_DYNAMIC,
3333
                       selected=self.op.output_fields)
3334

    
3335
    self.needed_locks = {}
3336
    self.share_locks[locking.LEVEL_INSTANCE] = 1
3337
    self.share_locks[locking.LEVEL_NODE] = 1
3338

    
3339
    if self.op.names:
3340
      self.wanted = _GetWantedInstances(self, self.op.names)
3341
    else:
3342
      self.wanted = locking.ALL_SET
3343

    
3344
    self.do_node_query = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
3345
    self.do_locking = self.do_node_query and self.op.use_locking
3346
    if self.do_locking:
3347
      self.needed_locks[locking.LEVEL_INSTANCE] = self.wanted
3348
      self.needed_locks[locking.LEVEL_NODE] = []
3349
      self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3350

    
3351
  def DeclareLocks(self, level):
3352
    if level == locking.LEVEL_NODE and self.do_locking:
3353
      self._LockInstancesNodes()
3354

    
3355
  def CheckPrereq(self):
3356
    """Check prerequisites.
3357

3358
    """
3359
    pass
3360

    
3361
  def Exec(self, feedback_fn):
3362
    """Computes the list of nodes and their attributes.
3363

3364
    """
3365
    all_info = self.cfg.GetAllInstancesInfo()
3366
    if self.wanted == locking.ALL_SET:
3367
      # caller didn't specify instance names, so ordering is not important
3368
      if self.do_locking:
3369
        instance_names = self.acquired_locks[locking.LEVEL_INSTANCE]
3370
      else:
3371
        instance_names = all_info.keys()
3372
      instance_names = utils.NiceSort(instance_names)
3373
    else:
3374
      # caller did specify names, so we must keep the ordering
3375
      if self.do_locking:
3376
        tgt_set = self.acquired_locks[locking.LEVEL_INSTANCE]
3377
      else:
3378
        tgt_set = all_info.keys()
3379
      missing = set(self.wanted).difference(tgt_set)
3380
      if missing:
3381
        raise errors.OpExecError("Some instances were removed before"
3382
                                 " retrieving their data: %s" % missing)
3383
      instance_names = self.wanted
3384

    
3385
    instance_list = [all_info[iname] for iname in instance_names]
3386

    
3387
    # begin data gathering
3388

    
3389
    nodes = frozenset([inst.primary_node for inst in instance_list])
3390
    hv_list = list(set([inst.hypervisor for inst in instance_list]))
3391

    
3392
    bad_nodes = []
3393
    off_nodes = []
3394
    if self.do_node_query:
3395
      live_data = {}
3396
      node_data = self.rpc.call_all_instances_info(nodes, hv_list)
3397
      for name in nodes:
3398
        result = node_data[name]
3399
        if result.offline:
3400
          # offline nodes will be in both lists
3401
          off_nodes.append(name)
3402
        if result.failed:
3403
          bad_nodes.append(name)
3404
        else:
3405
          if result.data:
3406
            live_data.update(result.data)
3407
            # else no instance is alive
3408
    else:
3409
      live_data = dict([(name, {}) for name in instance_names])
3410

    
3411
    # end data gathering
3412

    
3413
    HVPREFIX = "hv/"
3414
    BEPREFIX = "be/"
3415
    output = []
3416
    for instance in instance_list:
3417
      iout = []
3418
      i_hv = self.cfg.GetClusterInfo().FillHV(instance)
3419
      i_be = self.cfg.GetClusterInfo().FillBE(instance)
3420
      for field in self.op.output_fields:
3421
        st_match = self._FIELDS_STATIC.Matches(field)
3422
        if field == "name":
3423
          val = instance.name
3424
        elif field == "os":
3425
          val = instance.os
3426
        elif field == "pnode":
3427
          val = instance.primary_node
3428
        elif field == "snodes":
3429
          val = list(instance.secondary_nodes)
3430
        elif field == "admin_state":
3431
          val = instance.admin_up
3432
        elif field == "oper_state":
3433
          if instance.primary_node in bad_nodes:
3434
            val = None
3435
          else:
3436
            val = bool(live_data.get(instance.name))
3437
        elif field == "status":
3438
          if instance.primary_node in off_nodes:
3439
            val = "ERROR_nodeoffline"
3440
          elif instance.primary_node in bad_nodes:
3441
            val = "ERROR_nodedown"
3442
          else:
3443
            running = bool(live_data.get(instance.name))
3444
            if running:
3445
              if instance.admin_up:
3446
                val = "running"
3447
              else:
3448
                val = "ERROR_up"
3449
            else:
3450
              if instance.admin_up:
3451
                val = "ERROR_down"
3452
              else:
3453
                val = "ADMIN_down"
3454
        elif field == "oper_ram":
3455
          if instance.primary_node in bad_nodes:
3456
            val = None
3457
          elif instance.name in live_data:
3458
            val = live_data[instance.name].get("memory", "?")
3459
          else:
3460
            val = "-"
3461
        elif field == "disk_template":
3462
          val = instance.disk_template
3463
        elif field == "ip":
3464
          val = instance.nics[0].ip
3465
        elif field == "bridge":
3466
          val = instance.nics[0].bridge
3467
        elif field == "mac":
3468
          val = instance.nics[0].mac
3469
        elif field == "sda_size" or field == "sdb_size":
3470
          idx = ord(field[2]) - ord('a')
3471
          try:
3472
            val = instance.FindDisk(idx).size
3473
          except errors.OpPrereqError:
3474
            val = None
3475
        elif field == "disk_usage": # total disk usage per node
3476
          disk_sizes = [{'size': disk.size} for disk in instance.disks]
3477
          val = _ComputeDiskSize(instance.disk_template, disk_sizes)
3478
        elif field == "tags":
3479
          val = list(instance.GetTags())
3480
        elif field == "serial_no":
3481
          val = instance.serial_no
3482
        elif field == "network_port":
3483
          val = instance.network_port
3484
        elif field == "hypervisor":
3485
          val = instance.hypervisor
3486
        elif field == "hvparams":
3487
          val = i_hv
3488
        elif (field.startswith(HVPREFIX) and
3489
              field[len(HVPREFIX):] in constants.HVS_PARAMETERS):
3490
          val = i_hv.get(field[len(HVPREFIX):], None)
3491
        elif field == "beparams":
3492
          val = i_be
3493
        elif (field.startswith(BEPREFIX) and
3494
              field[len(BEPREFIX):] in constants.BES_PARAMETERS):
3495
          val = i_be.get(field[len(BEPREFIX):], None)
3496
        elif st_match and st_match.groups():
3497
          # matches a variable list
3498
          st_groups = st_match.groups()
3499
          if st_groups and st_groups[0] == "disk":
3500
            if st_groups[1] == "count":
3501
              val = len(instance.disks)
3502
            elif st_groups[1] == "sizes":
3503
              val = [disk.size for disk in instance.disks]
3504
            elif st_groups[1] == "size":
3505
              try:
3506
                val = instance.FindDisk(st_groups[2]).size
3507
              except errors.OpPrereqError:
3508
                val = None
3509
            else:
3510
              assert False, "Unhandled disk parameter"
3511
          elif st_groups[0] == "nic":
3512
            if st_groups[1] == "count":
3513
              val = len(instance.nics)
3514
            elif st_groups[1] == "macs":
3515
              val = [nic.mac for nic in instance.nics]
3516
            elif st_groups[1] == "ips":
3517
              val = [nic.ip for nic in instance.nics]
3518
            elif st_groups[1] == "bridges":
3519
              val = [nic.bridge for nic in instance.nics]
3520
            else:
3521
              # index-based item
3522
              nic_idx = int(st_groups[2])
3523
              if nic_idx >= len(instance.nics):
3524
                val = None
3525
              else:
3526
                if st_groups[1] == "mac":
3527
                  val = instance.nics[nic_idx].mac
3528
                elif st_groups[1] == "ip":
3529
                  val = instance.nics[nic_idx].ip
3530
                elif st_groups[1] == "bridge":
3531
                  val = instance.nics[nic_idx].bridge
3532
                else:
3533
                  assert False, "Unhandled NIC parameter"
3534
          else:
3535
            assert False, "Unhandled variable parameter"
3536
        else:
3537
          raise errors.ParameterError(field)
3538
        iout.append(val)
3539
      output.append(iout)
3540

    
3541
    return output
3542

    
3543

    
3544
class LUFailoverInstance(LogicalUnit):
3545
  """Failover an instance.
3546

3547
  """
3548
  HPATH = "instance-failover"
3549
  HTYPE = constants.HTYPE_INSTANCE
3550
  _OP_REQP = ["instance_name", "ignore_consistency"]
3551
  REQ_BGL = False
3552

    
3553
  def ExpandNames(self):
3554
    self._ExpandAndLockInstance()
3555
    self.needed_locks[locking.LEVEL_NODE] = []
3556
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3557

    
3558
  def DeclareLocks(self, level):
3559
    if level == locking.LEVEL_NODE:
3560
      self._LockInstancesNodes()
3561

    
3562
  def BuildHooksEnv(self):
3563
    """Build hooks env.
3564

3565
    This runs on master, primary and secondary nodes of the instance.
3566

3567
    """
3568
    env = {
3569
      "IGNORE_CONSISTENCY": self.op.ignore_consistency,
3570
      }
3571
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3572
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3573
    return env, nl, nl
3574

    
3575
  def CheckPrereq(self):
3576
    """Check prerequisites.
3577

3578
    This checks that the instance is in the cluster.
3579

3580
    """
3581
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3582
    assert self.instance is not None, \
3583
      "Cannot retrieve locked instance %s" % self.op.instance_name
3584

    
3585
    bep = self.cfg.GetClusterInfo().FillBE(instance)
3586
    if instance.disk_template not in constants.DTS_NET_MIRROR:
3587
      raise errors.OpPrereqError("Instance's disk layout is not"
3588
                                 " network mirrored, cannot failover.")
3589

    
3590
    secondary_nodes = instance.secondary_nodes
3591
    if not secondary_nodes:
3592
      raise errors.ProgrammerError("no secondary node but using "
3593
                                   "a mirrored disk template")
3594

    
3595
    target_node = secondary_nodes[0]
3596
    _CheckNodeOnline(self, target_node)
3597
    _CheckNodeNotDrained(self, target_node)
3598
    # check memory requirements on the secondary node
3599
    _CheckNodeFreeMemory(self, target_node, "failing over instance %s" %
3600
                         instance.name, bep[constants.BE_MEMORY],
3601
                         instance.hypervisor)
3602

    
3603
    # check bridge existance
3604
    brlist = [nic.bridge for nic in instance.nics]
3605
    result = self.rpc.call_bridges_exist(target_node, brlist)
3606
    result.Raise()
3607
    if not result.data:
3608
      raise errors.OpPrereqError("One or more target bridges %s does not"
3609
                                 " exist on destination node '%s'" %
3610
                                 (brlist, target_node))
3611

    
3612
  def Exec(self, feedback_fn):
3613
    """Failover an instance.
3614

3615
    The failover is done by shutting it down on its present node and
3616
    starting it on the secondary.
3617

3618
    """
3619
    instance = self.instance
3620

    
3621
    source_node = instance.primary_node
3622
    target_node = instance.secondary_nodes[0]
3623

    
3624
    feedback_fn("* checking disk consistency between source and target")
3625
    for dev in instance.disks:
3626
      # for drbd, these are drbd over lvm
3627
      if not _CheckDiskConsistency(self, dev, target_node, False):
3628
        if instance.admin_up and not self.op.ignore_consistency:
3629
          raise errors.OpExecError("Disk %s is degraded on target node,"
3630
                                   " aborting failover." % dev.iv_name)
3631

    
3632
    feedback_fn("* shutting down instance on source node")
3633
    logging.info("Shutting down instance %s on node %s",
3634
                 instance.name, source_node)
3635

    
3636
    result = self.rpc.call_instance_shutdown(source_node, instance)
3637
    msg = result.RemoteFailMsg()
3638
    if msg:
3639
      if self.op.ignore_consistency:
3640
        self.proc.LogWarning("Could not shutdown instance %s on node %s."
3641
                             " Proceeding anyway. Please make sure node"
3642
                             " %s is down. Error details: %s",
3643
                             instance.name, source_node, source_node, msg)
3644
      else:
3645
        raise errors.OpExecError("Could not shutdown instance %s on"
3646
                                 " node %s: %s" %
3647
                                 (instance.name, source_node, msg))
3648

    
3649
    feedback_fn("* deactivating the instance's disks on source node")
3650
    if not _ShutdownInstanceDisks(self, instance, ignore_primary=True):
3651
      raise errors.OpExecError("Can't shut down the instance's disks.")
3652

    
3653
    instance.primary_node = target_node
3654
    # distribute new instance config to the other nodes
3655
    self.cfg.Update(instance)
3656

    
3657
    # Only start the instance if it's marked as up
3658
    if instance.admin_up:
3659
      feedback_fn("* activating the instance's disks on target node")
3660
      logging.info("Starting instance %s on node %s",
3661
                   instance.name, target_node)
3662

    
3663
      disks_ok, dummy = _AssembleInstanceDisks(self, instance,
3664
                                               ignore_secondaries=True)
3665
      if not disks_ok:
3666
        _ShutdownInstanceDisks(self, instance)
3667
        raise errors.OpExecError("Can't activate the instance's disks")
3668

    
3669
      feedback_fn("* starting the instance on the target node")
3670
      result = self.rpc.call_instance_start(target_node, instance, None, None)
3671
      msg = result.RemoteFailMsg()
3672
      if msg:
3673
        _ShutdownInstanceDisks(self, instance)
3674
        raise errors.OpExecError("Could not start instance %s on node %s: %s" %
3675
                                 (instance.name, target_node, msg))
3676

    
3677

    
3678
class LUMigrateInstance(LogicalUnit):
3679
  """Migrate an instance.
3680

3681
  This is migration without shutting down, compared to the failover,
3682
  which is done with shutdown.
3683

3684
  """
3685
  HPATH = "instance-migrate"
3686
  HTYPE = constants.HTYPE_INSTANCE
3687
  _OP_REQP = ["instance_name", "live", "cleanup"]
3688

    
3689
  REQ_BGL = False
3690

    
3691
  def ExpandNames(self):
3692
    self._ExpandAndLockInstance()
3693
    self.needed_locks[locking.LEVEL_NODE] = []
3694
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3695

    
3696
  def DeclareLocks(self, level):
3697
    if level == locking.LEVEL_NODE:
3698
      self._LockInstancesNodes()
3699

    
3700
  def BuildHooksEnv(self):
3701
    """Build hooks env.
3702

3703
    This runs on master, primary and secondary nodes of the instance.
3704

3705
    """
3706
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3707
    env["MIGRATE_LIVE"] = self.op.live
3708
    env["MIGRATE_CLEANUP"] = self.op.cleanup
3709
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3710
    return env, nl, nl
3711

    
3712
  def CheckPrereq(self):
3713
    """Check prerequisites.
3714

3715
    This checks that the instance is in the cluster.
3716

3717
    """
3718
    instance = self.cfg.GetInstanceInfo(
3719
      self.cfg.ExpandInstanceName(self.op.instance_name))
3720
    if instance is None:
3721
      raise errors.OpPrereqError("Instance '%s' not known" %
3722
                                 self.op.instance_name)
3723

    
3724
    if instance.disk_template != constants.DT_DRBD8:
3725
      raise errors.OpPrereqError("Instance's disk layout is not"
3726
                                 " drbd8, cannot migrate.")
3727

    
3728
    secondary_nodes = instance.secondary_nodes
3729
    if not secondary_nodes:
3730
      raise errors.ConfigurationError("No secondary node but using"
3731
                                      " drbd8 disk template")
3732

    
3733
    i_be = self.cfg.GetClusterInfo().FillBE(instance)
3734

    
3735
    target_node = secondary_nodes[0]
3736
    # check memory requirements on the secondary node
3737
    _CheckNodeFreeMemory(self, target_node, "migrating instance %s" %
3738
                         instance.name, i_be[constants.BE_MEMORY],
3739
                         instance.hypervisor)
3740

    
3741
    # check bridge existance
3742
    brlist = [nic.bridge for nic in instance.nics]
3743
    result = self.rpc.call_bridges_exist(target_node, brlist)
3744
    if result.failed or not result.data:
3745
      raise errors.OpPrereqError("One or more target bridges %s does not"
3746
                                 " exist on destination node '%s'" %
3747
                                 (brlist, target_node))
3748

    
3749
    if not self.op.cleanup:
3750
      _CheckNodeNotDrained(self, target_node)
3751
      result = self.rpc.call_instance_migratable(instance.primary_node,
3752
                                                 instance)
3753
      msg = result.RemoteFailMsg()
3754
      if msg:
3755
        raise errors.OpPrereqError("Can't migrate: %s - please use failover" %
3756
                                   msg)
3757

    
3758
    self.instance = instance
3759

    
3760
  def _WaitUntilSync(self):
3761
    """Poll with custom rpc for disk sync.
3762

3763
    This uses our own step-based rpc call.
3764

3765
    """
3766
    self.feedback_fn("* wait until resync is done")
3767
    all_done = False
3768
    while not all_done:
3769
      all_done = True
3770
      result = self.rpc.call_drbd_wait_sync(self.all_nodes,
3771
                                            self.nodes_ip,
3772
                                            self.instance.disks)
3773
      min_percent = 100
3774
      for node, nres in result.items():
3775
        msg = nres.RemoteFailMsg()
3776
        if msg:
3777
          raise errors.OpExecError("Cannot resync disks on node %s: %s" %
3778
                                   (node, msg))
3779
        node_done, node_percent = nres.payload
3780
        all_done = all_done and node_done
3781
        if node_percent is not None:
3782
          min_percent = min(min_percent, node_percent)
3783
      if not all_done:
3784
        if min_percent < 100:
3785
          self.feedback_fn("   - progress: %.1f%%" % min_percent)
3786
        time.sleep(2)
3787

    
3788
  def _EnsureSecondary(self, node):
3789
    """Demote a node to secondary.
3790

3791
    """
3792
    self.feedback_fn("* switching node %s to secondary mode" % node)
3793

    
3794
    for dev in self.instance.disks:
3795
      self.cfg.SetDiskID(dev, node)
3796

    
3797
    result = self.rpc.call_blockdev_close(node, self.instance.name,
3798
                                          self.instance.disks)
3799
    msg = result.RemoteFailMsg()
3800
    if msg:
3801
      raise errors.OpExecError("Cannot change disk to secondary on node %s,"
3802
                               " error %s" % (node, msg))
3803

    
3804
  def _GoStandalone(self):
3805
    """Disconnect from the network.
3806

3807
    """
3808
    self.feedback_fn("* changing into standalone mode")
3809
    result = self.rpc.call_drbd_disconnect_net(self.all_nodes, self.nodes_ip,
3810
                                               self.instance.disks)
3811
    for node, nres in result.items():
3812
      msg = nres.RemoteFailMsg()
3813
      if msg:
3814
        raise errors.OpExecError("Cannot disconnect disks node %s,"
3815
                                 " error %s" % (node, msg))
3816

    
3817
  def _GoReconnect(self, multimaster):
3818
    """Reconnect to the network.
3819

3820
    """
3821
    if multimaster:
3822
      msg = "dual-master"
3823
    else:
3824
      msg = "single-master"
3825
    self.feedback_fn("* changing disks into %s mode" % msg)
3826
    result = self.rpc.call_drbd_attach_net(self.all_nodes, self.nodes_ip,
3827
                                           self.instance.disks,
3828
                                           self.instance.name, multimaster)
3829
    for node, nres in result.items():
3830
      msg = nres.RemoteFailMsg()
3831
      if msg:
3832
        raise errors.OpExecError("Cannot change disks config on node %s,"
3833
                                 " error: %s" % (node, msg))
3834

    
3835
  def _ExecCleanup(self):
3836
    """Try to cleanup after a failed migration.
3837

3838
    The cleanup is done by:
3839
      - check that the instance is running only on one node
3840
        (and update the config if needed)
3841
      - change disks on its secondary node to secondary
3842
      - wait until disks are fully synchronized
3843
      - disconnect from the network
3844
      - change disks into single-master mode
3845
      - wait again until disks are fully synchronized
3846

3847
    """
3848
    instance = self.instance
3849
    target_node = self.target_node
3850
    source_node = self.source_node
3851

    
3852
    # check running on only one node
3853
    self.feedback_fn("* checking where the instance actually runs"
3854
                     " (if this hangs, the hypervisor might be in"
3855
                     " a bad state)")
3856
    ins_l = self.rpc.call_instance_list(self.all_nodes, [instance.hypervisor])
3857
    for node, result in ins_l.items():
3858
      result.Raise()
3859
      if not isinstance(result.data, list):
3860
        raise errors.OpExecError("Can't contact node '%s'" % node)
3861

    
3862
    runningon_source = instance.name in ins_l[source_node].data
3863
    runningon_target = instance.name in ins_l[target_node].data
3864

    
3865
    if runningon_source and runningon_target:
3866
      raise errors.OpExecError("Instance seems to be running on two nodes,"
3867
                               " or the hypervisor is confused. You will have"
3868
                               " to ensure manually that it runs only on one"
3869
                               " and restart this operation.")
3870

    
3871
    if not (runningon_source or runningon_target):
3872
      raise errors.OpExecError("Instance does not seem to be running at all."
3873
                               " In this case, it's safer to repair by"
3874
                               " running 'gnt-instance stop' to ensure disk"
3875
                               " shutdown, and then restarting it.")
3876

    
3877
    if runningon_target:
3878
      # the migration has actually succeeded, we need to update the config
3879
      self.feedback_fn("* instance running on secondary node (%s),"
3880
                       " updating config" % target_node)
3881
      instance.primary_node = target_node
3882
      self.cfg.Update(instance)
3883
      demoted_node = source_node
3884
    else:
3885
      self.feedback_fn("* instance confirmed to be running on its"
3886
                       " primary node (%s)" % source_node)
3887
      demoted_node = target_node
3888

    
3889
    self._EnsureSecondary(demoted_node)
3890
    try:
3891
      self._WaitUntilSync()
3892
    except errors.OpExecError:
3893
      # we ignore here errors, since if the device is standalone, it
3894
      # won't be able to sync
3895
      pass
3896
    self._GoStandalone()
3897
    self._GoReconnect(False)
3898
    self._WaitUntilSync()
3899

    
3900
    self.feedback_fn("* done")
3901

    
3902
  def _RevertDiskStatus(self):
3903
    """Try to revert the disk status after a failed migration.
3904

3905
    """
3906
    target_node = self.target_node
3907
    try:
3908
      self._EnsureSecondary(target_node)
3909
      self._GoStandalone()
3910
      self._GoReconnect(False)
3911
      self._WaitUntilSync()
3912
    except errors.OpExecError, err:
3913
      self.LogWarning("Migration failed and I can't reconnect the"
3914
                      " drives: error '%s'\n"
3915
                      "Please look and recover the instance status" %
3916
                      str(err))
3917

    
3918
  def _AbortMigration(self):
3919
    """Call the hypervisor code to abort a started migration.
3920

3921
    """
3922
    instance = self.instance
3923
    target_node = self.target_node
3924
    migration_info = self.migration_info
3925

    
3926
    abort_result = self.rpc.call_finalize_migration(target_node,
3927
                                                    instance,
3928
                                                    migration_info,
3929
                                                    False)
3930
    abort_msg = abort_result.RemoteFailMsg()
3931
    if abort_msg:
3932
      logging.error("Aborting migration failed on target node %s: %s" %
3933
                    (target_node, abort_msg))
3934
      # Don't raise an exception here, as we stil have to try to revert the
3935
      # disk status, even if this step failed.
3936

    
3937
  def _ExecMigration(self):
3938
    """Migrate an instance.
3939

3940
    The migrate is done by:
3941
      - change the disks into dual-master mode
3942
      - wait until disks are fully synchronized again
3943
      - migrate the instance
3944
      - change disks on the new secondary node (the old primary) to secondary
3945
      - wait until disks are fully synchronized
3946
      - change disks into single-master mode
3947

3948
    """
3949
    instance = self.instance
3950
    target_node = self.target_node
3951
    source_node = self.source_node
3952

    
3953
    self.feedback_fn("* checking disk consistency between source and target")
3954
    for dev in instance.disks:
3955
      if not _CheckDiskConsistency(self, dev, target_node, False):
3956
        raise errors.OpExecError("Disk %s is degraded or not fully"
3957
                                 " synchronized on target node,"
3958
                                 " aborting migrate." % dev.iv_name)
3959

    
3960
    # First get the migration information from the remote node
3961
    result = self.rpc.call_migration_info(source_node, instance)
3962
    msg = result.RemoteFailMsg()
3963
    if msg:
3964
      log_err = ("Failed fetching source migration information from %s: %s" %
3965
                 (source_node, msg))
3966
      logging.error(log_err)
3967
      raise errors.OpExecError(log_err)
3968

    
3969
    self.migration_info = migration_info = result.payload
3970

    
3971
    # Then switch the disks to master/master mode
3972
    self._EnsureSecondary(target_node)
3973
    self._GoStandalone()
3974
    self._GoReconnect(True)
3975
    self._WaitUntilSync()
3976

    
3977
    self.feedback_fn("* preparing %s to accept the instance" % target_node)
3978
    result = self.rpc.call_accept_instance(target_node,
3979
                                           instance,
3980
                                           migration_info,
3981
                                           self.nodes_ip[target_node])
3982

    
3983
    msg = result.RemoteFailMsg()
3984
    if msg:
3985
      logging.error("Instance pre-migration failed, trying to revert"
3986
                    " disk status: %s", msg)
3987
      self._AbortMigration()
3988
      self._RevertDiskStatus()
3989
      raise errors.OpExecError("Could not pre-migrate instance %s: %s" %
3990
                               (instance.name, msg))
3991

    
3992
    self.feedback_fn("* migrating instance to %s" % target_node)
3993
    time.sleep(10)
3994
    result = self.rpc.call_instance_migrate(source_node, instance,
3995
                                            self.nodes_ip[target_node],
3996
                                            self.op.live)
3997
    msg = result.RemoteFailMsg()
3998
    if msg:
3999
      logging.error("Instance migration failed, trying to revert"
4000
                    " disk status: %s", msg)
4001
      self._AbortMigration()
4002
      self._RevertDiskStatus()
4003
      raise errors.OpExecError("Could not migrate instance %s: %s" %
4004
                               (instance.name, msg))
4005
    time.sleep(10)
4006

    
4007
    instance.primary_node = target_node
4008