Statistics
| Branch: | Tag: | Revision:

root / lib / cmdlib.py @ fc8a6b8f

History | View | Annotate | Download (253.1 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=W0201
25

    
26
# W0201 since most LU attributes are defined in CheckPrereq or similar
27
# functions
28

    
29
import os
30
import os.path
31
import time
32
import re
33
import platform
34
import logging
35
import copy
36

    
37
from ganeti import ssh
38
from ganeti import utils
39
from ganeti import errors
40
from ganeti import hypervisor
41
from ganeti import locking
42
from ganeti import constants
43
from ganeti import objects
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 overridden 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 # pylint: disable-msg=C0103
91
    self.LogInfo = processor.LogInfo # pylint: disable-msg=C0103
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 acquired 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): # pylint: disable-msg=W0223
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
  def BuildHooksEnv(self):
339
    """Empty BuildHooksEnv for NoHooksLu.
340

341
    This just raises an error.
342

343
    """
344
    assert False, "BuildHooksEnv called for NoHooksLUs"
345

    
346

    
347
def _GetWantedNodes(lu, nodes):
348
  """Returns list of checked and expanded node names.
349

350
  @type lu: L{LogicalUnit}
351
  @param lu: the logical unit on whose behalf we execute
352
  @type nodes: list
353
  @param nodes: list of node names or None for all nodes
354
  @rtype: list
355
  @return: the list of nodes, sorted
356
  @raise errors.OpProgrammerError: if the nodes parameter is wrong type
357

358
  """
359
  if not isinstance(nodes, list):
360
    raise errors.OpPrereqError("Invalid argument type 'nodes'")
361

    
362
  if not nodes:
363
    raise errors.ProgrammerError("_GetWantedNodes should only be called with a"
364
      " non-empty list of nodes whose name is to be expanded.")
365

    
366
  wanted = []
367
  for name in nodes:
368
    node = lu.cfg.ExpandNodeName(name)
369
    if node is None:
370
      raise errors.OpPrereqError("No such node name '%s'" % name)
371
    wanted.append(node)
372

    
373
  return utils.NiceSort(wanted)
374

    
375

    
376
def _GetWantedInstances(lu, instances):
377
  """Returns list of checked and expanded instance names.
378

379
  @type lu: L{LogicalUnit}
380
  @param lu: the logical unit on whose behalf we execute
381
  @type instances: list
382
  @param instances: list of instance names or None for all instances
383
  @rtype: list
384
  @return: the list of instances, sorted
385
  @raise errors.OpPrereqError: if the instances parameter is wrong type
386
  @raise errors.OpPrereqError: if any of the passed instances is not found
387

388
  """
389
  if not isinstance(instances, list):
390
    raise errors.OpPrereqError("Invalid argument type 'instances'")
391

    
392
  if instances:
393
    wanted = []
394

    
395
    for name in instances:
396
      instance = lu.cfg.ExpandInstanceName(name)
397
      if instance is None:
398
        raise errors.OpPrereqError("No such instance name '%s'" % name)
399
      wanted.append(instance)
400

    
401
  else:
402
    wanted = utils.NiceSort(lu.cfg.GetInstanceList())
403
  return wanted
404

    
405

    
406
def _CheckOutputFields(static, dynamic, selected):
407
  """Checks whether all selected fields are valid.
408

409
  @type static: L{utils.FieldSet}
410
  @param static: static fields set
411
  @type dynamic: L{utils.FieldSet}
412
  @param dynamic: dynamic fields set
413

414
  """
415
  f = utils.FieldSet()
416
  f.Extend(static)
417
  f.Extend(dynamic)
418

    
419
  delta = f.NonMatching(selected)
420
  if delta:
421
    raise errors.OpPrereqError("Unknown output fields selected: %s"
422
                               % ",".join(delta))
423

    
424

    
425
def _CheckBooleanOpField(op, name):
426
  """Validates boolean opcode parameters.
427

428
  This will ensure that an opcode parameter is either a boolean value,
429
  or None (but that it always exists).
430

431
  """
432
  val = getattr(op, name, None)
433
  if not (val is None or isinstance(val, bool)):
434
    raise errors.OpPrereqError("Invalid boolean parameter '%s' (%s)" %
435
                               (name, str(val)))
436
  setattr(op, name, val)
437

    
438

    
439
def _CheckNodeOnline(lu, node):
440
  """Ensure that a given node is online.
441

442
  @param lu: the LU on behalf of which we make the check
443
  @param node: the node to check
444
  @raise errors.OpPrereqError: if the node is offline
445

446
  """
447
  if lu.cfg.GetNodeInfo(node).offline:
448
    raise errors.OpPrereqError("Can't use offline node %s" % node)
449

    
450

    
451
def _CheckNodeNotDrained(lu, node):
452
  """Ensure that a given node is not drained.
453

454
  @param lu: the LU on behalf of which we make the check
455
  @param node: the node to check
456
  @raise errors.OpPrereqError: if the node is drained
457

458
  """
459
  if lu.cfg.GetNodeInfo(node).drained:
460
    raise errors.OpPrereqError("Can't use drained node %s" % node)
461

    
462

    
463
def _BuildInstanceHookEnv(name, primary_node, secondary_nodes, os_type, status,
464
                          memory, vcpus, nics, disk_template, disks,
465
                          bep, hvp, hypervisor_name):
466
  """Builds instance related env variables for hooks
467

468
  This builds the hook environment from individual variables.
469

470
  @type name: string
471
  @param name: the name of the instance
472
  @type primary_node: string
473
  @param primary_node: the name of the instance's primary node
474
  @type secondary_nodes: list
475
  @param secondary_nodes: list of secondary nodes as strings
476
  @type os_type: string
477
  @param os_type: the name of the instance's OS
478
  @type status: boolean
479
  @param status: the should_run status of the instance
480
  @type memory: string
481
  @param memory: the memory size of the instance
482
  @type vcpus: string
483
  @param vcpus: the count of VCPUs the instance has
484
  @type nics: list
485
  @param nics: list of tuples (ip, bridge, mac) representing
486
      the NICs the instance  has
487
  @type disk_template: string
488
  @param disk_template: the disk template of the instance
489
  @type disks: list
490
  @param disks: the list of (size, mode) pairs
491
  @type bep: dict
492
  @param bep: the backend parameters for the instance
493
  @type hvp: dict
494
  @param hvp: the hypervisor parameters for the instance
495
  @type hypervisor_name: string
496
  @param hypervisor_name: the hypervisor for the instance
497
  @rtype: dict
498
  @return: the hook environment for this instance
499

500
  """
501
  if status:
502
    str_status = "up"
503
  else:
504
    str_status = "down"
505
  env = {
506
    "OP_TARGET": name,
507
    "INSTANCE_NAME": name,
508
    "INSTANCE_PRIMARY": primary_node,
509
    "INSTANCE_SECONDARIES": " ".join(secondary_nodes),
510
    "INSTANCE_OS_TYPE": os_type,
511
    "INSTANCE_STATUS": str_status,
512
    "INSTANCE_MEMORY": memory,
513
    "INSTANCE_VCPUS": vcpus,
514
    "INSTANCE_DISK_TEMPLATE": disk_template,
515
    "INSTANCE_HYPERVISOR": hypervisor_name,
516
  }
517

    
518
  if nics:
519
    nic_count = len(nics)
520
    for idx, (ip, bridge, mac) in enumerate(nics):
521
      if ip is None:
522
        ip = ""
523
      env["INSTANCE_NIC%d_IP" % idx] = ip
524
      env["INSTANCE_NIC%d_BRIDGE" % idx] = bridge
525
      env["INSTANCE_NIC%d_MAC" % idx] = mac
526
  else:
527
    nic_count = 0
528

    
529
  env["INSTANCE_NIC_COUNT"] = nic_count
530

    
531
  if disks:
532
    disk_count = len(disks)
533
    for idx, (size, mode) in enumerate(disks):
534
      env["INSTANCE_DISK%d_SIZE" % idx] = size
535
      env["INSTANCE_DISK%d_MODE" % idx] = mode
536
  else:
537
    disk_count = 0
538

    
539
  env["INSTANCE_DISK_COUNT"] = disk_count
540

    
541
  for source, kind in [(bep, "BE"), (hvp, "HV")]:
542
    for key, value in source.items():
543
      env["INSTANCE_%s_%s" % (kind, key)] = value
544

    
545
  return env
546

    
547

    
548
def _BuildInstanceHookEnvByObject(lu, instance, override=None):
549
  """Builds instance related env variables for hooks from an object.
550

551
  @type lu: L{LogicalUnit}
552
  @param lu: the logical unit on whose behalf we execute
553
  @type instance: L{objects.Instance}
554
  @param instance: the instance for which we should build the
555
      environment
556
  @type override: dict
557
  @param override: dictionary with key/values that will override
558
      our values
559
  @rtype: dict
560
  @return: the hook environment dictionary
561

562
  """
563
  cluster = lu.cfg.GetClusterInfo()
564
  bep = cluster.FillBE(instance)
565
  hvp = cluster.FillHV(instance)
566
  args = {
567
    'name': instance.name,
568
    'primary_node': instance.primary_node,
569
    'secondary_nodes': instance.secondary_nodes,
570
    'os_type': instance.os,
571
    'status': instance.admin_up,
572
    'memory': bep[constants.BE_MEMORY],
573
    'vcpus': bep[constants.BE_VCPUS],
574
    'nics': [(nic.ip, nic.bridge, nic.mac) for nic in instance.nics],
575
    'disk_template': instance.disk_template,
576
    'disks': [(disk.size, disk.mode) for disk in instance.disks],
577
    'bep': bep,
578
    'hvp': hvp,
579
    'hypervisor_name': instance.hypervisor,
580
  }
581
  if override:
582
    args.update(override)
583
  return _BuildInstanceHookEnv(**args)
584

    
585

    
586
def _AdjustCandidatePool(lu):
587
  """Adjust the candidate pool after node operations.
588

589
  """
590
  mod_list = lu.cfg.MaintainCandidatePool()
591
  if mod_list:
592
    lu.LogInfo("Promoted nodes to master candidate role: %s",
593
               ", ".join(node.name for node in mod_list))
594
    for name in mod_list:
595
      lu.context.ReaddNode(name)
596
  mc_now, mc_max = lu.cfg.GetMasterCandidateStats()
597
  if mc_now > mc_max:
598
    lu.LogInfo("Note: more nodes are candidates (%d) than desired (%d)" %
599
               (mc_now, mc_max))
600

    
601

    
602
def _CheckInstanceBridgesExist(lu, instance):
603
  """Check that the bridges needed by an instance exist.
604

605
  """
606
  # check bridges existence
607
  brlist = [nic.bridge for nic in instance.nics]
608
  result = lu.rpc.call_bridges_exist(instance.primary_node, brlist)
609
  result.Raise()
610
  if not result.data:
611
    raise errors.OpPrereqError("One or more target bridges %s does not"
612
                               " exist on destination node '%s'" %
613
                               (brlist, instance.primary_node))
614

    
615

    
616
class LUDestroyCluster(NoHooksLU):
617
  """Logical unit for destroying the cluster.
618

619
  """
620
  _OP_REQP = []
621

    
622
  def CheckPrereq(self):
623
    """Check prerequisites.
624

625
    This checks whether the cluster is empty.
626

627
    Any errors are signaled by raising errors.OpPrereqError.
628

629
    """
630
    master = self.cfg.GetMasterNode()
631

    
632
    nodelist = self.cfg.GetNodeList()
633
    if len(nodelist) != 1 or nodelist[0] != master:
634
      raise errors.OpPrereqError("There are still %d node(s) in"
635
                                 " this cluster." % (len(nodelist) - 1))
636
    instancelist = self.cfg.GetInstanceList()
637
    if instancelist:
638
      raise errors.OpPrereqError("There are still %d instance(s) in"
639
                                 " this cluster." % len(instancelist))
640

    
641
  def Exec(self, feedback_fn):
642
    """Destroys the cluster.
643

644
    """
645
    master = self.cfg.GetMasterNode()
646
    result = self.rpc.call_node_stop_master(master, False)
647
    result.Raise()
648
    if not result.data:
649
      raise errors.OpExecError("Could not disable the master role")
650
    priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
651
    utils.CreateBackup(priv_key)
652
    utils.CreateBackup(pub_key)
653
    return master
654

    
655

    
656
class LUVerifyCluster(LogicalUnit):
657
  """Verifies the cluster status.
658

659
  """
660
  HPATH = "cluster-verify"
661
  HTYPE = constants.HTYPE_CLUSTER
662
  _OP_REQP = ["skip_checks"]
663
  REQ_BGL = False
664

    
665
  def ExpandNames(self):
666
    self.needed_locks = {
667
      locking.LEVEL_NODE: locking.ALL_SET,
668
      locking.LEVEL_INSTANCE: locking.ALL_SET,
669
    }
670
    self.share_locks = dict(((i, 1) for i in locking.LEVELS))
671

    
672
  def _VerifyNode(self, nodeinfo, file_list, local_cksum,
673
                  node_result, feedback_fn, master_files,
674
                  drbd_map, vg_name):
675
    """Run multiple tests against a node.
676

677
    Test list:
678

679
      - compares ganeti version
680
      - checks vg existence and size > 20G
681
      - checks config file checksum
682
      - checks ssh to other nodes
683

684
    @type nodeinfo: L{objects.Node}
685
    @param nodeinfo: the node to check
686
    @param file_list: required list of files
687
    @param local_cksum: dictionary of local files and their checksums
688
    @param node_result: the results from the node
689
    @param feedback_fn: function used to accumulate results
690
    @param master_files: list of files that only masters should have
691
    @param drbd_map: the useddrbd minors for this node, in
692
        form of minor: (instance, must_exist) which correspond to instances
693
        and their running status
694
    @param vg_name: Ganeti Volume Group (result of self.cfg.GetVGName())
695

696
    """
697
    node = nodeinfo.name
698

    
699
    # main result, node_result should be a non-empty dict
700
    if not node_result or not isinstance(node_result, dict):
701
      feedback_fn("  - ERROR: unable to verify node %s." % (node,))
702
      return True
703

    
704
    # compares ganeti version
705
    local_version = constants.PROTOCOL_VERSION
706
    remote_version = node_result.get('version', None)
707
    if not (remote_version and isinstance(remote_version, (list, tuple)) and
708
            len(remote_version) == 2):
709
      feedback_fn("  - ERROR: connection to %s failed" % (node))
710
      return True
711

    
712
    if local_version != remote_version[0]:
713
      feedback_fn("  - ERROR: incompatible protocol versions: master %s,"
714
                  " node %s %s" % (local_version, node, remote_version[0]))
715
      return True
716

    
717
    # node seems compatible, we can actually try to look into its results
718

    
719
    bad = False
720

    
721
    # full package version
722
    if constants.RELEASE_VERSION != remote_version[1]:
723
      feedback_fn("  - WARNING: software version mismatch: master %s,"
724
                  " node %s %s" %
725
                  (constants.RELEASE_VERSION, node, remote_version[1]))
726

    
727
    # checks vg existence and size > 20G
728
    if vg_name is not None:
729
      vglist = node_result.get(constants.NV_VGLIST, None)
730
      if not vglist:
731
        feedback_fn("  - ERROR: unable to check volume groups on node %s." %
732
                        (node,))
733
        bad = True
734
      else:
735
        vgstatus = utils.CheckVolumeGroupSize(vglist, vg_name,
736
                                              constants.MIN_VG_SIZE)
737
        if vgstatus:
738
          feedback_fn("  - ERROR: %s on node %s" % (vgstatus, node))
739
          bad = True
740

    
741
    # checks config file checksum
742

    
743
    remote_cksum = node_result.get(constants.NV_FILELIST, None)
744
    if not isinstance(remote_cksum, dict):
745
      bad = True
746
      feedback_fn("  - ERROR: node hasn't returned file checksum data")
747
    else:
748
      for file_name in file_list:
749
        node_is_mc = nodeinfo.master_candidate
750
        must_have_file = file_name not in master_files
751
        if file_name not in remote_cksum:
752
          if node_is_mc or must_have_file:
753
            bad = True
754
            feedback_fn("  - ERROR: file '%s' missing" % file_name)
755
        elif remote_cksum[file_name] != local_cksum[file_name]:
756
          if node_is_mc or must_have_file:
757
            bad = True
758
            feedback_fn("  - ERROR: file '%s' has wrong checksum" % file_name)
759
          else:
760
            # not candidate and this is not a must-have file
761
            bad = True
762
            feedback_fn("  - ERROR: file '%s' should not exist on non master"
763
                        " candidates (and the file is outdated)" % file_name)
764
        else:
765
          # all good, except non-master/non-must have combination
766
          if not node_is_mc and not must_have_file:
767
            feedback_fn("  - ERROR: file '%s' should not exist on non master"
768
                        " candidates" % file_name)
769

    
770
    # checks ssh to any
771

    
772
    if constants.NV_NODELIST not in node_result:
773
      bad = True
774
      feedback_fn("  - ERROR: node hasn't returned node ssh connectivity data")
775
    else:
776
      if node_result[constants.NV_NODELIST]:
777
        bad = True
778
        for node in node_result[constants.NV_NODELIST]:
779
          feedback_fn("  - ERROR: ssh communication with node '%s': %s" %
780
                          (node, node_result[constants.NV_NODELIST][node]))
781

    
782
    if constants.NV_NODENETTEST not in node_result:
783
      bad = True
784
      feedback_fn("  - ERROR: node hasn't returned node tcp connectivity data")
785
    else:
786
      if node_result[constants.NV_NODENETTEST]:
787
        bad = True
788
        nlist = utils.NiceSort(node_result[constants.NV_NODENETTEST].keys())
789
        for node in nlist:
790
          feedback_fn("  - ERROR: tcp communication with node '%s': %s" %
791
                          (node, node_result[constants.NV_NODENETTEST][node]))
792

    
793
    hyp_result = node_result.get(constants.NV_HYPERVISOR, None)
794
    if isinstance(hyp_result, dict):
795
      for hv_name, hv_result in hyp_result.iteritems():
796
        if hv_result is not None:
797
          feedback_fn("  - ERROR: hypervisor %s verify failure: '%s'" %
798
                      (hv_name, hv_result))
799

    
800
    # check used drbd list
801
    if vg_name is not None:
802
      used_minors = node_result.get(constants.NV_DRBDLIST, [])
803
      if not isinstance(used_minors, (tuple, list)):
804
        feedback_fn("  - ERROR: cannot parse drbd status file: %s" %
805
                    str(used_minors))
806
      else:
807
        for minor, (iname, must_exist) in drbd_map.items():
808
          if minor not in used_minors and must_exist:
809
            feedback_fn("  - ERROR: drbd minor %d of instance %s is"
810
                        " not active" % (minor, iname))
811
            bad = True
812
        for minor in used_minors:
813
          if minor not in drbd_map:
814
            feedback_fn("  - ERROR: unallocated drbd minor %d is in use" %
815
                        minor)
816
            bad = True
817

    
818
    return bad
819

    
820
  def _VerifyInstance(self, instance, instanceconfig, node_vol_is,
821
                      node_instance, feedback_fn, n_offline):
822
    """Verify an instance.
823

824
    This function checks to see if the required block devices are
825
    available on the instance's node.
826

827
    """
828
    bad = False
829

    
830
    node_current = instanceconfig.primary_node
831

    
832
    node_vol_should = {}
833
    instanceconfig.MapLVsByNode(node_vol_should)
834

    
835
    for node in node_vol_should:
836
      if node in n_offline:
837
        # ignore missing volumes on offline nodes
838
        continue
839
      for volume in node_vol_should[node]:
840
        if node not in node_vol_is or volume not in node_vol_is[node]:
841
          feedback_fn("  - ERROR: volume %s missing on node %s" %
842
                          (volume, node))
843
          bad = True
844

    
845
    if instanceconfig.admin_up:
846
      if ((node_current not in node_instance or
847
          not instance in node_instance[node_current]) and
848
          node_current not in n_offline):
849
        feedback_fn("  - ERROR: instance %s not running on node %s" %
850
                        (instance, node_current))
851
        bad = True
852

    
853
    for node in node_instance:
854
      if (not node == node_current):
855
        if instance in node_instance[node]:
856
          feedback_fn("  - ERROR: instance %s should not run on node %s" %
857
                          (instance, node))
858
          bad = True
859

    
860
    return bad
861

    
862
  def _VerifyOrphanVolumes(self, node_vol_should, node_vol_is, feedback_fn):
863
    """Verify if there are any unknown volumes in the cluster.
864

865
    The .os, .swap and backup volumes are ignored. All other volumes are
866
    reported as unknown.
867

868
    """
869
    bad = False
870

    
871
    for node in node_vol_is:
872
      for volume in node_vol_is[node]:
873
        if node not in node_vol_should or volume not in node_vol_should[node]:
874
          feedback_fn("  - ERROR: volume %s on node %s should not exist" %
875
                      (volume, node))
876
          bad = True
877
    return bad
878

    
879
  def _VerifyOrphanInstances(self, instancelist, node_instance, feedback_fn):
880
    """Verify the list of running instances.
881

882
    This checks what instances are running but unknown to the cluster.
883

884
    """
885
    bad = False
886
    for node in node_instance:
887
      for runninginstance in node_instance[node]:
888
        if runninginstance not in instancelist:
889
          feedback_fn("  - ERROR: instance %s on node %s should not exist" %
890
                          (runninginstance, node))
891
          bad = True
892
    return bad
893

    
894
  def _VerifyNPlusOneMemory(self, node_info, instance_cfg, feedback_fn):
895
    """Verify N+1 Memory Resilience.
896

897
    Check that if one single node dies we can still start all the instances it
898
    was primary for.
899

900
    """
901
    bad = False
902

    
903
    for node, nodeinfo in node_info.iteritems():
904
      # This code checks that every node which is now listed as secondary has
905
      # enough memory to host all instances it is supposed to should a single
906
      # other node in the cluster fail.
907
      # FIXME: not ready for failover to an arbitrary node
908
      # FIXME: does not support file-backed instances
909
      # WARNING: we currently take into account down instances as well as up
910
      # ones, considering that even if they're down someone might want to start
911
      # them even in the event of a node failure.
912
      for prinode, instances in nodeinfo['sinst-by-pnode'].iteritems():
913
        needed_mem = 0
914
        for instance in instances:
915
          bep = self.cfg.GetClusterInfo().FillBE(instance_cfg[instance])
916
          if bep[constants.BE_AUTO_BALANCE]:
917
            needed_mem += bep[constants.BE_MEMORY]
918
        if nodeinfo['mfree'] < needed_mem:
919
          feedback_fn("  - ERROR: not enough memory on node %s to accommodate"
920
                      " failovers should node %s fail" % (node, prinode))
921
          bad = True
922
    return bad
923

    
924
  def CheckPrereq(self):
925
    """Check prerequisites.
926

927
    Transform the list of checks we're going to skip into a set and check that
928
    all its members are valid.
929

930
    """
931
    self.skip_set = frozenset(self.op.skip_checks)
932
    if not constants.VERIFY_OPTIONAL_CHECKS.issuperset(self.skip_set):
933
      raise errors.OpPrereqError("Invalid checks to be skipped specified")
934

    
935
  def BuildHooksEnv(self):
936
    """Build hooks env.
937

938
    Cluster-Verify hooks just ran in the post phase and their failure makes
939
    the output be logged in the verify output and the verification to fail.
940

941
    """
942
    all_nodes = self.cfg.GetNodeList()
943
    env = {
944
      "CLUSTER_TAGS": " ".join(self.cfg.GetClusterInfo().GetTags())
945
      }
946
    for node in self.cfg.GetAllNodesInfo().values():
947
      env["NODE_TAGS_%s" % node.name] = " ".join(node.GetTags())
948

    
949
    return env, [], all_nodes
950

    
951
  def Exec(self, feedback_fn):
952
    """Verify integrity of cluster, performing various test on nodes.
953

954
    """
955
    bad = False
956
    feedback_fn("* Verifying global settings")
957
    for msg in self.cfg.VerifyConfig():
958
      feedback_fn("  - ERROR: %s" % msg)
959

    
960
    vg_name = self.cfg.GetVGName()
961
    hypervisors = self.cfg.GetClusterInfo().enabled_hypervisors
962
    nodelist = utils.NiceSort(self.cfg.GetNodeList())
963
    nodeinfo = [self.cfg.GetNodeInfo(nname) for nname in nodelist]
964
    instancelist = utils.NiceSort(self.cfg.GetInstanceList())
965
    instanceinfo = dict((iname, self.cfg.GetInstanceInfo(iname))
966
                        for iname in instancelist)
967
    i_non_redundant = [] # Non redundant instances
968
    i_non_a_balanced = [] # Non auto-balanced instances
969
    n_offline = [] # List of offline nodes
970
    n_drained = [] # List of nodes being drained
971
    node_volume = {}
972
    node_instance = {}
973
    node_info = {}
974
    instance_cfg = {}
975

    
976
    # FIXME: verify OS list
977
    # do local checksums
978
    master_files = [constants.CLUSTER_CONF_FILE]
979

    
980
    file_names = ssconf.SimpleStore().GetFileList()
981
    file_names.append(constants.SSL_CERT_FILE)
982
    file_names.append(constants.RAPI_CERT_FILE)
983
    file_names.extend(master_files)
984

    
985
    local_checksums = utils.FingerprintFiles(file_names)
986

    
987
    feedback_fn("* Gathering data (%d nodes)" % len(nodelist))
988
    node_verify_param = {
989
      constants.NV_FILELIST: file_names,
990
      constants.NV_NODELIST: [node.name for node in nodeinfo
991
                              if not node.offline],
992
      constants.NV_HYPERVISOR: hypervisors,
993
      constants.NV_NODENETTEST: [(node.name, node.primary_ip,
994
                                  node.secondary_ip) for node in nodeinfo
995
                                 if not node.offline],
996
      constants.NV_INSTANCELIST: hypervisors,
997
      constants.NV_VERSION: None,
998
      constants.NV_HVINFO: self.cfg.GetHypervisorType(),
999
      }
1000
    if vg_name is not None:
1001
      node_verify_param[constants.NV_VGLIST] = None
1002
      node_verify_param[constants.NV_LVLIST] = vg_name
1003
      node_verify_param[constants.NV_DRBDLIST] = None
1004
    all_nvinfo = self.rpc.call_node_verify(nodelist, node_verify_param,
1005
                                           self.cfg.GetClusterName())
1006

    
1007
    cluster = self.cfg.GetClusterInfo()
1008
    master_node = self.cfg.GetMasterNode()
1009
    all_drbd_map = self.cfg.ComputeDRBDMap()
1010

    
1011
    for node_i in nodeinfo:
1012
      node = node_i.name
1013
      nresult = all_nvinfo[node].data
1014

    
1015
      if node_i.offline:
1016
        feedback_fn("* Skipping offline node %s" % (node,))
1017
        n_offline.append(node)
1018
        continue
1019

    
1020
      if node == master_node:
1021
        ntype = "master"
1022
      elif node_i.master_candidate:
1023
        ntype = "master candidate"
1024
      elif node_i.drained:
1025
        ntype = "drained"
1026
        n_drained.append(node)
1027
      else:
1028
        ntype = "regular"
1029
      feedback_fn("* Verifying node %s (%s)" % (node, ntype))
1030

    
1031
      if all_nvinfo[node].failed or not isinstance(nresult, dict):
1032
        feedback_fn("  - ERROR: connection to %s failed" % (node,))
1033
        bad = True
1034
        continue
1035

    
1036
      node_drbd = {}
1037
      for minor, instance in all_drbd_map[node].items():
1038
        if instance not in instanceinfo:
1039
          feedback_fn("  - ERROR: ghost instance '%s' in temporary DRBD map" %
1040
                      instance)
1041
          # ghost instance should not be running, but otherwise we
1042
          # don't give double warnings (both ghost instance and
1043
          # unallocated minor in use)
1044
          node_drbd[minor] = (instance, False)
1045
        else:
1046
          instance = instanceinfo[instance]
1047
          node_drbd[minor] = (instance.name, instance.admin_up)
1048
      result = self._VerifyNode(node_i, file_names, local_checksums,
1049
                                nresult, feedback_fn, master_files,
1050
                                node_drbd, vg_name)
1051
      bad = bad or result
1052

    
1053
      lvdata = nresult.get(constants.NV_LVLIST, "Missing LV data")
1054
      if vg_name is None:
1055
        node_volume[node] = {}
1056
      elif isinstance(lvdata, basestring):
1057
        feedback_fn("  - ERROR: LVM problem on node %s: %s" %
1058
                    (node, utils.SafeEncode(lvdata)))
1059
        bad = True
1060
        node_volume[node] = {}
1061
      elif not isinstance(lvdata, dict):
1062
        feedback_fn("  - ERROR: connection to %s failed (lvlist)" % (node,))
1063
        bad = True
1064
        continue
1065
      else:
1066
        node_volume[node] = lvdata
1067

    
1068
      # node_instance
1069
      idata = nresult.get(constants.NV_INSTANCELIST, None)
1070
      if not isinstance(idata, list):
1071
        feedback_fn("  - ERROR: connection to %s failed (instancelist)" %
1072
                    (node,))
1073
        bad = True
1074
        continue
1075

    
1076
      node_instance[node] = idata
1077

    
1078
      # node_info
1079
      nodeinfo = nresult.get(constants.NV_HVINFO, None)
1080
      if not isinstance(nodeinfo, dict):
1081
        feedback_fn("  - ERROR: connection to %s failed (hvinfo)" % (node,))
1082
        bad = True
1083
        continue
1084

    
1085
      try:
1086
        node_info[node] = {
1087
          "mfree": int(nodeinfo['memory_free']),
1088
          "pinst": [],
1089
          "sinst": [],
1090
          # dictionary holding all instances this node is secondary for,
1091
          # grouped by their primary node. Each key is a cluster node, and each
1092
          # value is a list of instances which have the key as primary and the
1093
          # current node as secondary.  this is handy to calculate N+1 memory
1094
          # availability if you can only failover from a primary to its
1095
          # secondary.
1096
          "sinst-by-pnode": {},
1097
        }
1098
        # FIXME: devise a free space model for file based instances as well
1099
        if vg_name is not None:
1100
          if (constants.NV_VGLIST not in nresult or
1101
              vg_name not in nresult[constants.NV_VGLIST]):
1102
            feedback_fn("  - ERROR: node %s didn't return data for the"
1103
                        " volume group '%s' - it is either missing or broken" %
1104
                        (node, vg_name))
1105
            bad = True
1106
            continue
1107
          node_info[node]["dfree"] = int(nresult[constants.NV_VGLIST][vg_name])
1108
      except (ValueError, KeyError):
1109
        feedback_fn("  - ERROR: invalid nodeinfo value returned"
1110
                    " from node %s" % (node,))
1111
        bad = True
1112
        continue
1113

    
1114
    node_vol_should = {}
1115

    
1116
    for instance in instancelist:
1117
      feedback_fn("* Verifying instance %s" % instance)
1118
      inst_config = instanceinfo[instance]
1119
      result =  self._VerifyInstance(instance, inst_config, node_volume,
1120
                                     node_instance, feedback_fn, n_offline)
1121
      bad = bad or result
1122
      inst_nodes_offline = []
1123

    
1124
      inst_config.MapLVsByNode(node_vol_should)
1125

    
1126
      instance_cfg[instance] = inst_config
1127

    
1128
      pnode = inst_config.primary_node
1129
      if pnode in node_info:
1130
        node_info[pnode]['pinst'].append(instance)
1131
      elif pnode not in n_offline:
1132
        feedback_fn("  - ERROR: instance %s, connection to primary node"
1133
                    " %s failed" % (instance, pnode))
1134
        bad = True
1135

    
1136
      if pnode in n_offline:
1137
        inst_nodes_offline.append(pnode)
1138

    
1139
      # If the instance is non-redundant we cannot survive losing its primary
1140
      # node, so we are not N+1 compliant. On the other hand we have no disk
1141
      # templates with more than one secondary so that situation is not well
1142
      # supported either.
1143
      # FIXME: does not support file-backed instances
1144
      if len(inst_config.secondary_nodes) == 0:
1145
        i_non_redundant.append(instance)
1146
      elif len(inst_config.secondary_nodes) > 1:
1147
        feedback_fn("  - WARNING: multiple secondaries for instance %s"
1148
                    % instance)
1149

    
1150
      if not cluster.FillBE(inst_config)[constants.BE_AUTO_BALANCE]:
1151
        i_non_a_balanced.append(instance)
1152

    
1153
      for snode in inst_config.secondary_nodes:
1154
        if snode in node_info:
1155
          node_info[snode]['sinst'].append(instance)
1156
          if pnode not in node_info[snode]['sinst-by-pnode']:
1157
            node_info[snode]['sinst-by-pnode'][pnode] = []
1158
          node_info[snode]['sinst-by-pnode'][pnode].append(instance)
1159
        elif snode not in n_offline:
1160
          feedback_fn("  - ERROR: instance %s, connection to secondary node"
1161
                      " %s failed" % (instance, snode))
1162
          bad = True
1163
        if snode in n_offline:
1164
          inst_nodes_offline.append(snode)
1165

    
1166
      if inst_nodes_offline:
1167
        # warn that the instance lives on offline nodes, and set bad=True
1168
        feedback_fn("  - ERROR: instance lives on offline node(s) %s" %
1169
                    ", ".join(inst_nodes_offline))
1170
        bad = True
1171

    
1172
    feedback_fn("* Verifying orphan volumes")
1173
    result = self._VerifyOrphanVolumes(node_vol_should, node_volume,
1174
                                       feedback_fn)
1175
    bad = bad or result
1176

    
1177
    feedback_fn("* Verifying remaining instances")
1178
    result = self._VerifyOrphanInstances(instancelist, node_instance,
1179
                                         feedback_fn)
1180
    bad = bad or result
1181

    
1182
    if constants.VERIFY_NPLUSONE_MEM not in self.skip_set:
1183
      feedback_fn("* Verifying N+1 Memory redundancy")
1184
      result = self._VerifyNPlusOneMemory(node_info, instance_cfg, feedback_fn)
1185
      bad = bad or result
1186

    
1187
    feedback_fn("* Other Notes")
1188
    if i_non_redundant:
1189
      feedback_fn("  - NOTICE: %d non-redundant instance(s) found."
1190
                  % len(i_non_redundant))
1191

    
1192
    if i_non_a_balanced:
1193
      feedback_fn("  - NOTICE: %d non-auto-balanced instance(s) found."
1194
                  % len(i_non_a_balanced))
1195

    
1196
    if n_offline:
1197
      feedback_fn("  - NOTICE: %d offline node(s) found." % len(n_offline))
1198

    
1199
    if n_drained:
1200
      feedback_fn("  - NOTICE: %d drained node(s) found." % len(n_drained))
1201

    
1202
    return not bad
1203

    
1204
  def HooksCallBack(self, phase, hooks_results, feedback_fn, lu_result):
1205
    """Analyze the post-hooks' result
1206

1207
    This method analyses the hook result, handles it, and sends some
1208
    nicely-formatted feedback back to the user.
1209

1210
    @param phase: one of L{constants.HOOKS_PHASE_POST} or
1211
        L{constants.HOOKS_PHASE_PRE}; it denotes the hooks phase
1212
    @param hooks_results: the results of the multi-node hooks rpc call
1213
    @param feedback_fn: function used send feedback back to the caller
1214
    @param lu_result: previous Exec result
1215
    @return: the new Exec result, based on the previous result
1216
        and hook results
1217

1218
    """
1219
    # We only really run POST phase hooks, and are only interested in
1220
    # their results
1221
    if phase == constants.HOOKS_PHASE_POST:
1222
      # Used to change hooks' output to proper indentation
1223
      indent_re = re.compile('^', re.M)
1224
      feedback_fn("* Hooks Results")
1225
      if not hooks_results:
1226
        feedback_fn("  - ERROR: general communication failure")
1227
        lu_result = 1
1228
      else:
1229
        for node_name in hooks_results:
1230
          show_node_header = True
1231
          res = hooks_results[node_name]
1232
          if res.failed or res.data is False or not isinstance(res.data, list):
1233
            if res.offline:
1234
              # no need to warn or set fail return value
1235
              continue
1236
            feedback_fn("    Communication failure in hooks execution")
1237
            lu_result = 1
1238
            continue
1239
          for script, hkr, output in res.data:
1240
            if hkr == constants.HKR_FAIL:
1241
              # The node header is only shown once, if there are
1242
              # failing hooks on that node
1243
              if show_node_header:
1244
                feedback_fn("  Node %s:" % node_name)
1245
                show_node_header = False
1246
              feedback_fn("    ERROR: Script %s failed, output:" % script)
1247
              output = indent_re.sub('      ', output)
1248
              feedback_fn("%s" % output)
1249
              lu_result = 1
1250

    
1251
      return lu_result
1252

    
1253

    
1254
class LUVerifyDisks(NoHooksLU):
1255
  """Verifies the cluster disks status.
1256

1257
  """
1258
  _OP_REQP = []
1259
  REQ_BGL = False
1260

    
1261
  def ExpandNames(self):
1262
    self.needed_locks = {
1263
      locking.LEVEL_NODE: locking.ALL_SET,
1264
      locking.LEVEL_INSTANCE: locking.ALL_SET,
1265
    }
1266
    self.share_locks = dict(((i, 1) for i in locking.LEVELS))
1267

    
1268
  def CheckPrereq(self):
1269
    """Check prerequisites.
1270

1271
    This has no prerequisites.
1272

1273
    """
1274
    pass
1275

    
1276
  def Exec(self, feedback_fn):
1277
    """Verify integrity of cluster disks.
1278

1279
    """
1280
    result = res_nodes, res_nlvm, res_instances, res_missing = [], {}, [], {}
1281

    
1282
    vg_name = self.cfg.GetVGName()
1283
    nodes = utils.NiceSort(self.cfg.GetNodeList())
1284
    instances = [self.cfg.GetInstanceInfo(name)
1285
                 for name in self.cfg.GetInstanceList()]
1286

    
1287
    nv_dict = {}
1288
    for inst in instances:
1289
      inst_lvs = {}
1290
      if (not inst.admin_up or
1291
          inst.disk_template not in constants.DTS_NET_MIRROR):
1292
        continue
1293
      inst.MapLVsByNode(inst_lvs)
1294
      # transform { iname: {node: [vol,],},} to {(node, vol): iname}
1295
      for node, vol_list in inst_lvs.iteritems():
1296
        for vol in vol_list:
1297
          nv_dict[(node, vol)] = inst
1298

    
1299
    if not nv_dict:
1300
      return result
1301

    
1302
    node_lvs = self.rpc.call_volume_list(nodes, vg_name)
1303

    
1304
    for node in nodes:
1305
      # node_volume
1306
      lvs = node_lvs[node]
1307
      if lvs.failed:
1308
        if not lvs.offline:
1309
          self.LogWarning("Connection to node %s failed: %s" %
1310
                          (node, lvs.data))
1311
        continue
1312
      lvs = lvs.data
1313
      if isinstance(lvs, basestring):
1314
        logging.warning("Error enumerating LVs on node %s: %s", node, lvs)
1315
        res_nlvm[node] = lvs
1316
        continue
1317
      elif not isinstance(lvs, dict):
1318
        logging.warning("Connection to node %s failed or invalid data"
1319
                        " returned", node)
1320
        res_nodes.append(node)
1321
        continue
1322

    
1323
      for lv_name, (_, lv_inactive, lv_online) in lvs.iteritems():
1324
        inst = nv_dict.pop((node, lv_name), None)
1325
        if (not lv_online and inst is not None
1326
            and inst.name not in res_instances):
1327
          res_instances.append(inst.name)
1328

    
1329
    # any leftover items in nv_dict are missing LVs, let's arrange the
1330
    # data better
1331
    for key, inst in nv_dict.iteritems():
1332
      if inst.name not in res_missing:
1333
        res_missing[inst.name] = []
1334
      res_missing[inst.name].append(key)
1335

    
1336
    return result
1337

    
1338

    
1339
class LURepairDiskSizes(NoHooksLU):
1340
  """Verifies the cluster disks sizes.
1341

1342
  """
1343
  _OP_REQP = ["instances"]
1344
  REQ_BGL = False
1345

    
1346
  def ExpandNames(self):
1347

    
1348
    if not isinstance(self.op.instances, list):
1349
      raise errors.OpPrereqError("Invalid argument type 'instances'")
1350

    
1351
    if self.op.instances:
1352
      self.wanted_names = []
1353
      for name in self.op.instances:
1354
        full_name = self.cfg.ExpandInstanceName(name)
1355
        if full_name is None:
1356
          raise errors.OpPrereqError("Instance '%s' not known" % name)
1357
        self.wanted_names.append(full_name)
1358
      self.needed_locks = {
1359
        locking.LEVEL_NODE: [],
1360
        locking.LEVEL_INSTANCE: self.wanted_names,
1361
        }
1362
      self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
1363
    else:
1364
      self.wanted_names = None
1365
      self.needed_locks = {
1366
        locking.LEVEL_NODE: locking.ALL_SET,
1367
        locking.LEVEL_INSTANCE: locking.ALL_SET,
1368
        }
1369
    self.share_locks = dict(((i, 1) for i in locking.LEVELS))
1370

    
1371
  def DeclareLocks(self, level):
1372
    if level == locking.LEVEL_NODE and self.wanted_names is not None:
1373
      self._LockInstancesNodes(primary_only=True)
1374

    
1375
  def CheckPrereq(self):
1376
    """Check prerequisites.
1377

1378
    This only checks the optional instance list against the existing names.
1379

1380
    """
1381
    if self.wanted_names is None:
1382
      self.wanted_names = self.acquired_locks[locking.LEVEL_INSTANCE]
1383

    
1384
    self.wanted_instances = [self.cfg.GetInstanceInfo(name) for name
1385
                             in self.wanted_names]
1386

    
1387
  def _EnsureChildSizes(self, disk):
1388
    """Ensure children of the disk have the needed disk size.
1389

1390
    This is valid mainly for DRBD8 and fixes an issue where the
1391
    children have smaller disk size.
1392

1393
    @param disk: an L{ganeti.objects.Disk} object
1394

1395
    """
1396
    if disk.dev_type == constants.LD_DRBD8:
1397
      assert disk.children, "Empty children for DRBD8?"
1398
      fchild = disk.children[0]
1399
      mismatch = fchild.size < disk.size
1400
      if mismatch:
1401
        self.LogInfo("Child disk has size %d, parent %d, fixing",
1402
                     fchild.size, disk.size)
1403
        fchild.size = disk.size
1404

    
1405
      # and we recurse on this child only, not on the metadev
1406
      return self._EnsureChildSizes(fchild) or mismatch
1407
    else:
1408
      return False
1409

    
1410
  def Exec(self, feedback_fn):
1411
    """Verify the size of cluster disks.
1412

1413
    """
1414
    # TODO: check child disks too
1415
    # TODO: check differences in size between primary/secondary nodes
1416
    per_node_disks = {}
1417
    for instance in self.wanted_instances:
1418
      pnode = instance.primary_node
1419
      if pnode not in per_node_disks:
1420
        per_node_disks[pnode] = []
1421
      for idx, disk in enumerate(instance.disks):
1422
        per_node_disks[pnode].append((instance, idx, disk))
1423

    
1424
    changed = []
1425
    for node, dskl in per_node_disks.items():
1426
      newl = [v[2].Copy() for v in dskl]
1427
      for dsk in newl:
1428
        self.cfg.SetDiskID(dsk, node)
1429
      result = self.rpc.call_blockdev_getsizes(node, newl)
1430
      if result.failed:
1431
        self.LogWarning("Failure in blockdev_getsizes call to node"
1432
                        " %s, ignoring", node)
1433
        continue
1434
      if len(result.data) != len(dskl):
1435
        self.LogWarning("Invalid result from node %s, ignoring node results",
1436
                        node)
1437
        continue
1438
      for ((instance, idx, disk), size) in zip(dskl, result.data):
1439
        if size is None:
1440
          self.LogWarning("Disk %d of instance %s did not return size"
1441
                          " information, ignoring", idx, instance.name)
1442
          continue
1443
        if not isinstance(size, (int, long)):
1444
          self.LogWarning("Disk %d of instance %s did not return valid"
1445
                          " size information, ignoring", idx, instance.name)
1446
          continue
1447
        size = size >> 20
1448
        if size != disk.size:
1449
          self.LogInfo("Disk %d of instance %s has mismatched size,"
1450
                       " correcting: recorded %d, actual %d", idx,
1451
                       instance.name, disk.size, size)
1452
          disk.size = size
1453
          self.cfg.Update(instance)
1454
          changed.append((instance.name, idx, size))
1455
        if self._EnsureChildSizes(disk):
1456
          self.cfg.Update(instance)
1457
          changed.append((instance.name, idx, disk.size))
1458
    return changed
1459

    
1460

    
1461
class LURenameCluster(LogicalUnit):
1462
  """Rename the cluster.
1463

1464
  """
1465
  HPATH = "cluster-rename"
1466
  HTYPE = constants.HTYPE_CLUSTER
1467
  _OP_REQP = ["name"]
1468

    
1469
  def BuildHooksEnv(self):
1470
    """Build hooks env.
1471

1472
    """
1473
    env = {
1474
      "OP_TARGET": self.cfg.GetClusterName(),
1475
      "NEW_NAME": self.op.name,
1476
      }
1477
    mn = self.cfg.GetMasterNode()
1478
    return env, [mn], [mn]
1479

    
1480
  def CheckPrereq(self):
1481
    """Verify that the passed name is a valid one.
1482

1483
    """
1484
    hostname = utils.HostInfo(self.op.name)
1485

    
1486
    new_name = hostname.name
1487
    self.ip = new_ip = hostname.ip
1488
    old_name = self.cfg.GetClusterName()
1489
    old_ip = self.cfg.GetMasterIP()
1490
    if new_name == old_name and new_ip == old_ip:
1491
      raise errors.OpPrereqError("Neither the name nor the IP address of the"
1492
                                 " cluster has changed")
1493
    if new_ip != old_ip:
1494
      if utils.TcpPing(new_ip, constants.DEFAULT_NODED_PORT):
1495
        raise errors.OpPrereqError("The given cluster IP address (%s) is"
1496
                                   " reachable on the network. Aborting." %
1497
                                   new_ip)
1498

    
1499
    self.op.name = new_name
1500

    
1501
  def Exec(self, feedback_fn):
1502
    """Rename the cluster.
1503

1504
    """
1505
    clustername = self.op.name
1506
    ip = self.ip
1507

    
1508
    # shutdown the master IP
1509
    master = self.cfg.GetMasterNode()
1510
    result = self.rpc.call_node_stop_master(master, False)
1511
    if result.failed or not result.data:
1512
      raise errors.OpExecError("Could not disable the master role")
1513

    
1514
    try:
1515
      cluster = self.cfg.GetClusterInfo()
1516
      cluster.cluster_name = clustername
1517
      cluster.master_ip = ip
1518
      self.cfg.Update(cluster)
1519

    
1520
      # update the known hosts file
1521
      ssh.WriteKnownHostsFile(self.cfg, constants.SSH_KNOWN_HOSTS_FILE)
1522
      node_list = self.cfg.GetNodeList()
1523
      try:
1524
        node_list.remove(master)
1525
      except ValueError:
1526
        pass
1527
      result = self.rpc.call_upload_file(node_list,
1528
                                         constants.SSH_KNOWN_HOSTS_FILE)
1529
      for to_node, to_result in result.iteritems():
1530
        if to_result.failed or not to_result.data:
1531
          logging.error("Copy of file %s to node %s failed",
1532
                        constants.SSH_KNOWN_HOSTS_FILE, to_node)
1533

    
1534
    finally:
1535
      result = self.rpc.call_node_start_master(master, False, False)
1536
      if result.failed or not result.data:
1537
        self.LogWarning("Could not re-enable the master role on"
1538
                        " the master, please restart manually.")
1539

    
1540

    
1541
def _RecursiveCheckIfLVMBased(disk):
1542
  """Check if the given disk or its children are lvm-based.
1543

1544
  @type disk: L{objects.Disk}
1545
  @param disk: the disk to check
1546
  @rtype: boolean
1547
  @return: boolean indicating whether a LD_LV dev_type was found or not
1548

1549
  """
1550
  if disk.children:
1551
    for chdisk in disk.children:
1552
      if _RecursiveCheckIfLVMBased(chdisk):
1553
        return True
1554
  return disk.dev_type == constants.LD_LV
1555

    
1556

    
1557
class LUSetClusterParams(LogicalUnit):
1558
  """Change the parameters of the cluster.
1559

1560
  """
1561
  HPATH = "cluster-modify"
1562
  HTYPE = constants.HTYPE_CLUSTER
1563
  _OP_REQP = []
1564
  REQ_BGL = False
1565

    
1566
  def CheckArguments(self):
1567
    """Check parameters
1568

1569
    """
1570
    if not hasattr(self.op, "candidate_pool_size"):
1571
      self.op.candidate_pool_size = None
1572
    if self.op.candidate_pool_size is not None:
1573
      try:
1574
        self.op.candidate_pool_size = int(self.op.candidate_pool_size)
1575
      except (ValueError, TypeError), err:
1576
        raise errors.OpPrereqError("Invalid candidate_pool_size value: %s" %
1577
                                   str(err))
1578
      if self.op.candidate_pool_size < 1:
1579
        raise errors.OpPrereqError("At least one master candidate needed")
1580

    
1581
  def ExpandNames(self):
1582
    # FIXME: in the future maybe other cluster params won't require checking on
1583
    # all nodes to be modified.
1584
    self.needed_locks = {
1585
      locking.LEVEL_NODE: locking.ALL_SET,
1586
    }
1587
    self.share_locks[locking.LEVEL_NODE] = 1
1588

    
1589
  def BuildHooksEnv(self):
1590
    """Build hooks env.
1591

1592
    """
1593
    env = {
1594
      "OP_TARGET": self.cfg.GetClusterName(),
1595
      "NEW_VG_NAME": self.op.vg_name,
1596
      }
1597
    mn = self.cfg.GetMasterNode()
1598
    return env, [mn], [mn]
1599

    
1600
  def CheckPrereq(self):
1601
    """Check prerequisites.
1602

1603
    This checks whether the given params don't conflict and
1604
    if the given volume group is valid.
1605

1606
    """
1607
    if self.op.vg_name is not None and not self.op.vg_name:
1608
      instances = self.cfg.GetAllInstancesInfo().values()
1609
      for inst in instances:
1610
        for disk in inst.disks:
1611
          if _RecursiveCheckIfLVMBased(disk):
1612
            raise errors.OpPrereqError("Cannot disable lvm storage while"
1613
                                       " lvm-based instances exist")
1614

    
1615
    node_list = self.acquired_locks[locking.LEVEL_NODE]
1616

    
1617
    # if vg_name not None, checks given volume group on all nodes
1618
    if self.op.vg_name:
1619
      vglist = self.rpc.call_vg_list(node_list)
1620
      for node in node_list:
1621
        if vglist[node].failed:
1622
          # ignoring down node
1623
          self.LogWarning("Node %s unreachable/error, ignoring" % node)
1624
          continue
1625
        vgstatus = utils.CheckVolumeGroupSize(vglist[node].data,
1626
                                              self.op.vg_name,
1627
                                              constants.MIN_VG_SIZE)
1628
        if vgstatus:
1629
          raise errors.OpPrereqError("Error on node '%s': %s" %
1630
                                     (node, vgstatus))
1631

    
1632
    self.cluster = cluster = self.cfg.GetClusterInfo()
1633
    # validate beparams changes
1634
    if self.op.beparams:
1635
      utils.ForceDictType(self.op.beparams, constants.BES_PARAMETER_TYPES)
1636
      self.new_beparams = cluster.FillDict(
1637
        cluster.beparams[constants.BEGR_DEFAULT], self.op.beparams)
1638

    
1639
    # hypervisor list/parameters
1640
    self.new_hvparams = cluster.FillDict(cluster.hvparams, {})
1641
    if self.op.hvparams:
1642
      if not isinstance(self.op.hvparams, dict):
1643
        raise errors.OpPrereqError("Invalid 'hvparams' parameter on input")
1644
      for hv_name, hv_dict in self.op.hvparams.items():
1645
        if hv_name not in self.new_hvparams:
1646
          self.new_hvparams[hv_name] = hv_dict
1647
        else:
1648
          self.new_hvparams[hv_name].update(hv_dict)
1649

    
1650
    if self.op.enabled_hypervisors is not None:
1651
      self.hv_list = self.op.enabled_hypervisors
1652
      if not self.hv_list:
1653
        raise errors.OpPrereqError("Enabled hypervisors list must contain at"
1654
                                   " least one member")
1655
      invalid_hvs = set(self.hv_list) - constants.HYPER_TYPES
1656
      if invalid_hvs:
1657
        raise errors.OpPrereqError("Enabled hypervisors contains invalid"
1658
                                   " entries: %s" % invalid_hvs)
1659
    else:
1660
      self.hv_list = cluster.enabled_hypervisors
1661

    
1662
    if self.op.hvparams or self.op.enabled_hypervisors is not None:
1663
      # either the enabled list has changed, or the parameters have, validate
1664
      for hv_name, hv_params in self.new_hvparams.items():
1665
        if ((self.op.hvparams and hv_name in self.op.hvparams) or
1666
            (self.op.enabled_hypervisors and
1667
             hv_name in self.op.enabled_hypervisors)):
1668
          # either this is a new hypervisor, or its parameters have changed
1669
          hv_class = hypervisor.GetHypervisor(hv_name)
1670
          utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
1671
          hv_class.CheckParameterSyntax(hv_params)
1672
          _CheckHVParams(self, node_list, hv_name, hv_params)
1673

    
1674
  def Exec(self, feedback_fn):
1675
    """Change the parameters of the cluster.
1676

1677
    """
1678
    if self.op.vg_name is not None:
1679
      new_volume = self.op.vg_name
1680
      if not new_volume:
1681
        new_volume = None
1682
      if new_volume != self.cfg.GetVGName():
1683
        self.cfg.SetVGName(new_volume)
1684
      else:
1685
        feedback_fn("Cluster LVM configuration already in desired"
1686
                    " state, not changing")
1687
    if self.op.hvparams:
1688
      self.cluster.hvparams = self.new_hvparams
1689
    if self.op.enabled_hypervisors is not None:
1690
      self.cluster.enabled_hypervisors = self.op.enabled_hypervisors
1691
    if self.op.beparams:
1692
      self.cluster.beparams[constants.BEGR_DEFAULT] = self.new_beparams
1693
    if self.op.candidate_pool_size is not None:
1694
      self.cluster.candidate_pool_size = self.op.candidate_pool_size
1695
      # we need to update the pool size here, otherwise the save will fail
1696
      _AdjustCandidatePool(self)
1697

    
1698
    self.cfg.Update(self.cluster)
1699

    
1700

    
1701
class LURedistributeConfig(NoHooksLU):
1702
  """Force the redistribution of cluster configuration.
1703

1704
  This is a very simple LU.
1705

1706
  """
1707
  _OP_REQP = []
1708
  REQ_BGL = False
1709

    
1710
  def ExpandNames(self):
1711
    self.needed_locks = {
1712
      locking.LEVEL_NODE: locking.ALL_SET,
1713
    }
1714
    self.share_locks[locking.LEVEL_NODE] = 1
1715

    
1716
  def CheckPrereq(self):
1717
    """Check prerequisites.
1718

1719
    """
1720

    
1721
  def Exec(self, feedback_fn):
1722
    """Redistribute the configuration.
1723

1724
    """
1725
    self.cfg.Update(self.cfg.GetClusterInfo())
1726

    
1727

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

1731
  """
1732
  if not instance.disks:
1733
    return True
1734

    
1735
  if not oneshot:
1736
    lu.proc.LogInfo("Waiting for instance %s to sync disks." % instance.name)
1737

    
1738
  node = instance.primary_node
1739

    
1740
  for dev in instance.disks:
1741
    lu.cfg.SetDiskID(dev, node)
1742

    
1743
  retries = 0
1744
  degr_retries = 10 # in seconds, as we sleep 1 second each time
1745
  while True:
1746
    max_time = 0
1747
    done = True
1748
    cumul_degraded = False
1749
    rstats = lu.rpc.call_blockdev_getmirrorstatus(node, instance.disks)
1750
    if rstats.failed or not rstats.data:
1751
      lu.LogWarning("Can't get any data from node %s", node)
1752
      retries += 1
1753
      if retries >= 10:
1754
        raise errors.RemoteError("Can't contact node %s for mirror data,"
1755
                                 " aborting." % node)
1756
      time.sleep(6)
1757
      continue
1758
    rstats = rstats.data
1759
    retries = 0
1760
    for i, mstat in enumerate(rstats):
1761
      if mstat is None:
1762
        lu.LogWarning("Can't compute data for node %s/%s",
1763
                           node, instance.disks[i].iv_name)
1764
        continue
1765
      # we ignore the ldisk parameter
1766
      perc_done, est_time, is_degraded, _ = mstat
1767
      cumul_degraded = cumul_degraded or (is_degraded and perc_done is None)
1768
      if perc_done is not None:
1769
        done = False
1770
        if est_time is not None:
1771
          rem_time = "%d estimated seconds remaining" % est_time
1772
          max_time = est_time
1773
        else:
1774
          rem_time = "no time estimate"
1775
        lu.proc.LogInfo("- device %s: %5.2f%% done, %s" %
1776
                        (instance.disks[i].iv_name, perc_done, rem_time))
1777

    
1778
    # if we're done but degraded, let's do a few small retries, to
1779
    # make sure we see a stable and not transient situation; therefore
1780
    # we force restart of the loop
1781
    if (done or oneshot) and cumul_degraded and degr_retries > 0:
1782
      logging.info("Degraded disks found, %d retries left", degr_retries)
1783
      degr_retries -= 1
1784
      time.sleep(1)
1785
      continue
1786

    
1787
    if done or oneshot:
1788
      break
1789

    
1790
    time.sleep(min(60, max_time))
1791

    
1792
  if done:
1793
    lu.proc.LogInfo("Instance %s's disks are in sync." % instance.name)
1794
  return not cumul_degraded
1795

    
1796

    
1797
def _CheckDiskConsistency(lu, dev, node, on_primary, ldisk=False):
1798
  """Check that mirrors are not degraded.
1799

1800
  The ldisk parameter, if True, will change the test from the
1801
  is_degraded attribute (which represents overall non-ok status for
1802
  the device(s)) to the ldisk (representing the local storage status).
1803

1804
  """
1805
  lu.cfg.SetDiskID(dev, node)
1806
  if ldisk:
1807
    idx = 6
1808
  else:
1809
    idx = 5
1810

    
1811
  result = True
1812
  if on_primary or dev.AssembleOnSecondary():
1813
    rstats = lu.rpc.call_blockdev_find(node, dev)
1814
    msg = rstats.RemoteFailMsg()
1815
    if msg:
1816
      lu.LogWarning("Can't find disk on node %s: %s", node, msg)
1817
      result = False
1818
    elif not rstats.payload:
1819
      lu.LogWarning("Can't find disk on node %s", node)
1820
      result = False
1821
    else:
1822
      result = result and (not rstats.payload[idx])
1823
  if dev.children:
1824
    for child in dev.children:
1825
      result = result and _CheckDiskConsistency(lu, child, node, on_primary)
1826

    
1827
  return result
1828

    
1829

    
1830
class LUDiagnoseOS(NoHooksLU):
1831
  """Logical unit for OS diagnose/query.
1832

1833
  """
1834
  _OP_REQP = ["output_fields", "names"]
1835
  REQ_BGL = False
1836
  _FIELDS_STATIC = utils.FieldSet()
1837
  _FIELDS_DYNAMIC = utils.FieldSet("name", "valid", "node_status")
1838

    
1839
  def ExpandNames(self):
1840
    if self.op.names:
1841
      raise errors.OpPrereqError("Selective OS query not supported")
1842

    
1843
    _CheckOutputFields(static=self._FIELDS_STATIC,
1844
                       dynamic=self._FIELDS_DYNAMIC,
1845
                       selected=self.op.output_fields)
1846

    
1847
    # Lock all nodes, in shared mode
1848
    # Temporary removal of locks, should be reverted later
1849
    # TODO: reintroduce locks when they are lighter-weight
1850
    self.needed_locks = {}
1851
    #self.share_locks[locking.LEVEL_NODE] = 1
1852
    #self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
1853

    
1854
  def CheckPrereq(self):
1855
    """Check prerequisites.
1856

1857
    """
1858

    
1859
  @staticmethod
1860
  def _DiagnoseByOS(node_list, rlist):
1861
    """Remaps a per-node return list into an a per-os per-node dictionary
1862

1863
    @param node_list: a list with the names of all nodes
1864
    @param rlist: a map with node names as keys and OS objects as values
1865

1866
    @rtype: dict
1867
    @return: a dictionary with osnames as keys and as value another map, with
1868
        nodes as keys and list of OS objects as values, eg::
1869

1870
          {"debian-etch": {"node1": [<object>,...],
1871
                           "node2": [<object>,]}
1872
          }
1873

1874
    """
1875
    all_os = {}
1876
    # we build here the list of nodes that didn't fail the RPC (at RPC
1877
    # level), so that nodes with a non-responding node daemon don't
1878
    # make all OSes invalid
1879
    good_nodes = [node_name for node_name in rlist
1880
                  if not rlist[node_name].failed]
1881
    for node_name, nr in rlist.iteritems():
1882
      if nr.failed or not nr.data:
1883
        continue
1884
      for os_obj in nr.data:
1885
        if os_obj.name not in all_os:
1886
          # build a list of nodes for this os containing empty lists
1887
          # for each node in node_list
1888
          all_os[os_obj.name] = {}
1889
          for nname in good_nodes:
1890
            all_os[os_obj.name][nname] = []
1891
        all_os[os_obj.name][node_name].append(os_obj)
1892
    return all_os
1893

    
1894
  def Exec(self, feedback_fn):
1895
    """Compute the list of OSes.
1896

1897
    """
1898
    valid_nodes = [node for node in self.cfg.GetOnlineNodeList()]
1899
    node_data = self.rpc.call_os_diagnose(valid_nodes)
1900
    if node_data == False:
1901
      raise errors.OpExecError("Can't gather the list of OSes")
1902
    pol = self._DiagnoseByOS(valid_nodes, node_data)
1903
    output = []
1904
    for os_name, os_data in pol.iteritems():
1905
      row = []
1906
      for field in self.op.output_fields:
1907
        if field == "name":
1908
          val = os_name
1909
        elif field == "valid":
1910
          val = utils.all([osl and osl[0] for osl in os_data.values()])
1911
        elif field == "node_status":
1912
          val = {}
1913
          for node_name, nos_list in os_data.iteritems():
1914
            val[node_name] = [(v.status, v.path) for v in nos_list]
1915
        else:
1916
          raise errors.ParameterError(field)
1917
        row.append(val)
1918
      output.append(row)
1919

    
1920
    return output
1921

    
1922

    
1923
class LURemoveNode(LogicalUnit):
1924
  """Logical unit for removing a node.
1925

1926
  """
1927
  HPATH = "node-remove"
1928
  HTYPE = constants.HTYPE_NODE
1929
  _OP_REQP = ["node_name"]
1930

    
1931
  def BuildHooksEnv(self):
1932
    """Build hooks env.
1933

1934
    This doesn't run on the target node in the pre phase as a failed
1935
    node would then be impossible to remove.
1936

1937
    """
1938
    env = {
1939
      "OP_TARGET": self.op.node_name,
1940
      "NODE_NAME": self.op.node_name,
1941
      }
1942
    all_nodes = self.cfg.GetNodeList()
1943
    all_nodes.remove(self.op.node_name)
1944
    return env, all_nodes, all_nodes
1945

    
1946
  def CheckPrereq(self):
1947
    """Check prerequisites.
1948

1949
    This checks:
1950
     - the node exists in the configuration
1951
     - it does not have primary or secondary instances
1952
     - it's not the master
1953

1954
    Any errors are signaled by raising errors.OpPrereqError.
1955

1956
    """
1957
    node = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.node_name))
1958
    if node is None:
1959
      raise errors.OpPrereqError, ("Node '%s' is unknown." % self.op.node_name)
1960

    
1961
    instance_list = self.cfg.GetInstanceList()
1962

    
1963
    masternode = self.cfg.GetMasterNode()
1964
    if node.name == masternode:
1965
      raise errors.OpPrereqError("Node is the master node,"
1966
                                 " you need to failover first.")
1967

    
1968
    for instance_name in instance_list:
1969
      instance = self.cfg.GetInstanceInfo(instance_name)
1970
      if node.name in instance.all_nodes:
1971
        raise errors.OpPrereqError("Instance %s is still running on the node,"
1972
                                   " please remove first." % instance_name)
1973
    self.op.node_name = node.name
1974
    self.node = node
1975

    
1976
  def Exec(self, feedback_fn):
1977
    """Removes the node from the cluster.
1978

1979
    """
1980
    node = self.node
1981
    logging.info("Stopping the node daemon and removing configs from node %s",
1982
                 node.name)
1983

    
1984
    self.context.RemoveNode(node.name)
1985

    
1986
    self.rpc.call_node_leave_cluster(node.name)
1987

    
1988
    # Promote nodes to master candidate as needed
1989
    _AdjustCandidatePool(self)
1990

    
1991

    
1992
class LUQueryNodes(NoHooksLU):
1993
  """Logical unit for querying nodes.
1994

1995
  """
1996
  _OP_REQP = ["output_fields", "names", "use_locking"]
1997
  REQ_BGL = False
1998
  _FIELDS_DYNAMIC = utils.FieldSet(
1999
    "dtotal", "dfree",
2000
    "mtotal", "mnode", "mfree",
2001
    "bootid",
2002
    "ctotal", "cnodes", "csockets",
2003
    )
2004

    
2005
  _FIELDS_STATIC = utils.FieldSet(
2006
    "name", "pinst_cnt", "sinst_cnt",
2007
    "pinst_list", "sinst_list",
2008
    "pip", "sip", "tags",
2009
    "serial_no",
2010
    "master_candidate",
2011
    "master",
2012
    "offline",
2013
    "drained",
2014
    "role",
2015
    )
2016

    
2017
  def ExpandNames(self):
2018
    _CheckOutputFields(static=self._FIELDS_STATIC,
2019
                       dynamic=self._FIELDS_DYNAMIC,
2020
                       selected=self.op.output_fields)
2021

    
2022
    self.needed_locks = {}
2023
    self.share_locks[locking.LEVEL_NODE] = 1
2024

    
2025
    if self.op.names:
2026
      self.wanted = _GetWantedNodes(self, self.op.names)
2027
    else:
2028
      self.wanted = locking.ALL_SET
2029

    
2030
    self.do_node_query = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
2031
    self.do_locking = self.do_node_query and self.op.use_locking
2032
    if self.do_locking:
2033
      # if we don't request only static fields, we need to lock the nodes
2034
      self.needed_locks[locking.LEVEL_NODE] = self.wanted
2035

    
2036

    
2037
  def CheckPrereq(self):
2038
    """Check prerequisites.
2039

2040
    """
2041
    # The validation of the node list is done in the _GetWantedNodes,
2042
    # if non empty, and if empty, there's no validation to do
2043
    pass
2044

    
2045
  def Exec(self, feedback_fn):
2046
    """Computes the list of nodes and their attributes.
2047

2048
    """
2049
    all_info = self.cfg.GetAllNodesInfo()
2050
    if self.do_locking:
2051
      nodenames = self.acquired_locks[locking.LEVEL_NODE]
2052
    elif self.wanted != locking.ALL_SET:
2053
      nodenames = self.wanted
2054
      missing = set(nodenames).difference(all_info.keys())
2055
      if missing:
2056
        raise errors.OpExecError(
2057
          "Some nodes were removed before retrieving their data: %s" % missing)
2058
    else:
2059
      nodenames = all_info.keys()
2060

    
2061
    nodenames = utils.NiceSort(nodenames)
2062
    nodelist = [all_info[name] for name in nodenames]
2063

    
2064
    # begin data gathering
2065

    
2066
    if self.do_node_query:
2067
      live_data = {}
2068
      node_data = self.rpc.call_node_info(nodenames, self.cfg.GetVGName(),
2069
                                          self.cfg.GetHypervisorType())
2070
      for name in nodenames:
2071
        nodeinfo = node_data[name]
2072
        if not nodeinfo.failed and nodeinfo.data:
2073
          nodeinfo = nodeinfo.data
2074
          fn = utils.TryConvert
2075
          live_data[name] = {
2076
            "mtotal": fn(int, nodeinfo.get('memory_total', None)),
2077
            "mnode": fn(int, nodeinfo.get('memory_dom0', None)),
2078
            "mfree": fn(int, nodeinfo.get('memory_free', None)),
2079
            "dtotal": fn(int, nodeinfo.get('vg_size', None)),
2080
            "dfree": fn(int, nodeinfo.get('vg_free', None)),
2081
            "ctotal": fn(int, nodeinfo.get('cpu_total', None)),
2082
            "bootid": nodeinfo.get('bootid', None),
2083
            "cnodes": fn(int, nodeinfo.get('cpu_nodes', None)),
2084
            "csockets": fn(int, nodeinfo.get('cpu_sockets', None)),
2085
            }
2086
        else:
2087
          live_data[name] = {}
2088
    else:
2089
      live_data = dict.fromkeys(nodenames, {})
2090

    
2091
    node_to_primary = dict([(name, set()) for name in nodenames])
2092
    node_to_secondary = dict([(name, set()) for name in nodenames])
2093

    
2094
    inst_fields = frozenset(("pinst_cnt", "pinst_list",
2095
                             "sinst_cnt", "sinst_list"))
2096
    if inst_fields & frozenset(self.op.output_fields):
2097
      inst_data = self.cfg.GetAllInstancesInfo()
2098

    
2099
      for instance_name, inst in inst_data.items():
2100
        if inst.primary_node in node_to_primary:
2101
          node_to_primary[inst.primary_node].add(inst.name)
2102
        for secnode in inst.secondary_nodes:
2103
          if secnode in node_to_secondary:
2104
            node_to_secondary[secnode].add(inst.name)
2105

    
2106
    master_node = self.cfg.GetMasterNode()
2107

    
2108
    # end data gathering
2109

    
2110
    output = []
2111
    for node in nodelist:
2112
      node_output = []
2113
      for field in self.op.output_fields:
2114
        if field == "name":
2115
          val = node.name
2116
        elif field == "pinst_list":
2117
          val = list(node_to_primary[node.name])
2118
        elif field == "sinst_list":
2119
          val = list(node_to_secondary[node.name])
2120
        elif field == "pinst_cnt":
2121
          val = len(node_to_primary[node.name])
2122
        elif field == "sinst_cnt":
2123
          val = len(node_to_secondary[node.name])
2124
        elif field == "pip":
2125
          val = node.primary_ip
2126
        elif field == "sip":
2127
          val = node.secondary_ip
2128
        elif field == "tags":
2129
          val = list(node.GetTags())
2130
        elif field == "serial_no":
2131
          val = node.serial_no
2132
        elif field == "master_candidate":
2133
          val = node.master_candidate
2134
        elif field == "master":
2135
          val = node.name == master_node
2136
        elif field == "offline":
2137
          val = node.offline
2138
        elif field == "drained":
2139
          val = node.drained
2140
        elif self._FIELDS_DYNAMIC.Matches(field):
2141
          val = live_data[node.name].get(field, None)
2142
        elif field == "role":
2143
          if node.name == master_node:
2144
            val = "M"
2145
          elif node.master_candidate:
2146
            val = "C"
2147
          elif node.drained:
2148
            val = "D"
2149
          elif node.offline:
2150
            val = "O"
2151
          else:
2152
            val = "R"
2153
        else:
2154
          raise errors.ParameterError(field)
2155
        node_output.append(val)
2156
      output.append(node_output)
2157

    
2158
    return output
2159

    
2160

    
2161
class LUQueryNodeVolumes(NoHooksLU):
2162
  """Logical unit for getting volumes on node(s).
2163

2164
  """
2165
  _OP_REQP = ["nodes", "output_fields"]
2166
  REQ_BGL = False
2167
  _FIELDS_DYNAMIC = utils.FieldSet("phys", "vg", "name", "size", "instance")
2168
  _FIELDS_STATIC = utils.FieldSet("node")
2169

    
2170
  def ExpandNames(self):
2171
    _CheckOutputFields(static=self._FIELDS_STATIC,
2172
                       dynamic=self._FIELDS_DYNAMIC,
2173
                       selected=self.op.output_fields)
2174

    
2175
    self.needed_locks = {}
2176
    self.share_locks[locking.LEVEL_NODE] = 1
2177
    if not self.op.nodes:
2178
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
2179
    else:
2180
      self.needed_locks[locking.LEVEL_NODE] = \
2181
        _GetWantedNodes(self, self.op.nodes)
2182

    
2183
  def CheckPrereq(self):
2184
    """Check prerequisites.
2185

2186
    This checks that the fields required are valid output fields.
2187

2188
    """
2189
    self.nodes = self.acquired_locks[locking.LEVEL_NODE]
2190

    
2191
  def Exec(self, feedback_fn):
2192
    """Computes the list of nodes and their attributes.
2193

2194
    """
2195
    nodenames = self.nodes
2196
    volumes = self.rpc.call_node_volumes(nodenames)
2197

    
2198
    ilist = [self.cfg.GetInstanceInfo(iname) for iname
2199
             in self.cfg.GetInstanceList()]
2200

    
2201
    lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
2202

    
2203
    output = []
2204
    for node in nodenames:
2205
      if node not in volumes or volumes[node].failed or not volumes[node].data:
2206
        continue
2207

    
2208
      node_vols = volumes[node].data[:]
2209
      node_vols.sort(key=lambda vol: vol['dev'])
2210

    
2211
      for vol in node_vols:
2212
        node_output = []
2213
        for field in self.op.output_fields:
2214
          if field == "node":
2215
            val = node
2216
          elif field == "phys":
2217
            val = vol['dev']
2218
          elif field == "vg":
2219
            val = vol['vg']
2220
          elif field == "name":
2221
            val = vol['name']
2222
          elif field == "size":
2223
            val = int(float(vol['size']))
2224
          elif field == "instance":
2225
            for inst in ilist:
2226
              if node not in lv_by_node[inst]:
2227
                continue
2228
              if vol['name'] in lv_by_node[inst][node]:
2229
                val = inst.name
2230
                break
2231
            else:
2232
              val = '-'
2233
          else:
2234
            raise errors.ParameterError(field)
2235
          node_output.append(str(val))
2236

    
2237
        output.append(node_output)
2238

    
2239
    return output
2240

    
2241

    
2242
class LUAddNode(LogicalUnit):
2243
  """Logical unit for adding node to the cluster.
2244

2245
  """
2246
  HPATH = "node-add"
2247
  HTYPE = constants.HTYPE_NODE
2248
  _OP_REQP = ["node_name"]
2249

    
2250
  def BuildHooksEnv(self):
2251
    """Build hooks env.
2252

2253
    This will run on all nodes before, and on all nodes + the new node after.
2254

2255
    """
2256
    env = {
2257
      "OP_TARGET": self.op.node_name,
2258
      "NODE_NAME": self.op.node_name,
2259
      "NODE_PIP": self.op.primary_ip,
2260
      "NODE_SIP": self.op.secondary_ip,
2261
      }
2262
    nodes_0 = self.cfg.GetNodeList()
2263
    nodes_1 = nodes_0 + [self.op.node_name, ]
2264
    return env, nodes_0, nodes_1
2265

    
2266
  def CheckPrereq(self):
2267
    """Check prerequisites.
2268

2269
    This checks:
2270
     - the new node is not already in the config
2271
     - it is resolvable
2272
     - its parameters (single/dual homed) matches the cluster
2273

2274
    Any errors are signaled by raising errors.OpPrereqError.
2275

2276
    """
2277
    node_name = self.op.node_name
2278
    cfg = self.cfg
2279

    
2280
    dns_data = utils.HostInfo(node_name)
2281

    
2282
    node = dns_data.name
2283
    primary_ip = self.op.primary_ip = dns_data.ip
2284
    secondary_ip = getattr(self.op, "secondary_ip", None)
2285
    if secondary_ip is None:
2286
      secondary_ip = primary_ip
2287
    if not utils.IsValidIP(secondary_ip):
2288
      raise errors.OpPrereqError("Invalid secondary IP given")
2289
    self.op.secondary_ip = secondary_ip
2290

    
2291
    node_list = cfg.GetNodeList()
2292
    if not self.op.readd and node in node_list:
2293
      raise errors.OpPrereqError("Node %s is already in the configuration" %
2294
                                 node)
2295
    elif self.op.readd and node not in node_list:
2296
      raise errors.OpPrereqError("Node %s is not in the configuration" % node)
2297

    
2298
    for existing_node_name in node_list:
2299
      existing_node = cfg.GetNodeInfo(existing_node_name)
2300

    
2301
      if self.op.readd and node == existing_node_name:
2302
        if (existing_node.primary_ip != primary_ip or
2303
            existing_node.secondary_ip != secondary_ip):
2304
          raise errors.OpPrereqError("Readded node doesn't have the same IP"
2305
                                     " address configuration as before")
2306
        continue
2307

    
2308
      if (existing_node.primary_ip == primary_ip or
2309
          existing_node.secondary_ip == primary_ip or
2310
          existing_node.primary_ip == secondary_ip or
2311
          existing_node.secondary_ip == secondary_ip):
2312
        raise errors.OpPrereqError("New node ip address(es) conflict with"
2313
                                   " existing node %s" % existing_node.name)
2314

    
2315
    # check that the type of the node (single versus dual homed) is the
2316
    # same as for the master
2317
    myself = cfg.GetNodeInfo(self.cfg.GetMasterNode())
2318
    master_singlehomed = myself.secondary_ip == myself.primary_ip
2319
    newbie_singlehomed = secondary_ip == primary_ip
2320
    if master_singlehomed != newbie_singlehomed:
2321
      if master_singlehomed:
2322
        raise errors.OpPrereqError("The master has no private ip but the"
2323
                                   " new node has one")
2324
      else:
2325
        raise errors.OpPrereqError("The master has a private ip but the"
2326
                                   " new node doesn't have one")
2327

    
2328
    # checks reachability
2329
    if not utils.TcpPing(primary_ip, constants.DEFAULT_NODED_PORT):
2330
      raise errors.OpPrereqError("Node not reachable by ping")
2331

    
2332
    if not newbie_singlehomed:
2333
      # check reachability from my secondary ip to newbie's secondary ip
2334
      if not utils.TcpPing(secondary_ip, constants.DEFAULT_NODED_PORT,
2335
                           source=myself.secondary_ip):
2336
        raise errors.OpPrereqError("Node secondary ip not reachable by TCP"
2337
                                   " based ping to noded port")
2338

    
2339
    cp_size = self.cfg.GetClusterInfo().candidate_pool_size
2340
    if self.op.readd:
2341
      exceptions = [node]
2342
    else:
2343
      exceptions = []
2344
    mc_now, mc_max = self.cfg.GetMasterCandidateStats(exceptions)
2345
    # the new node will increase mc_max with one, so:
2346
    mc_max = min(mc_max + 1, cp_size)
2347
    self.master_candidate = mc_now < mc_max
2348

    
2349
    if self.op.readd:
2350
      self.new_node = self.cfg.GetNodeInfo(node)
2351
      assert self.new_node is not None, "Can't retrieve locked node %s" % node
2352
    else:
2353
      self.new_node = objects.Node(name=node,
2354
                                   primary_ip=primary_ip,
2355
                                   secondary_ip=secondary_ip,
2356
                                   master_candidate=self.master_candidate,
2357
                                   offline=False, drained=False)
2358

    
2359
  def Exec(self, feedback_fn):
2360
    """Adds the new node to the cluster.
2361

2362
    """
2363
    new_node = self.new_node
2364
    node = new_node.name
2365

    
2366
    # for re-adds, reset the offline/drained/master-candidate flags;
2367
    # we need to reset here, otherwise offline would prevent RPC calls
2368
    # later in the procedure; this also means that if the re-add
2369
    # fails, we are left with a non-offlined, broken node
2370
    if self.op.readd:
2371
      new_node.drained = new_node.offline = False
2372
      self.LogInfo("Readding a node, the offline/drained flags were reset")
2373
      # if we demote the node, we do cleanup later in the procedure
2374
      new_node.master_candidate = self.master_candidate
2375

    
2376
    # notify the user about any possible mc promotion
2377
    if new_node.master_candidate:
2378
      self.LogInfo("Node will be a master candidate")
2379

    
2380
    # check connectivity
2381
    result = self.rpc.call_version([node])[node]
2382
    result.Raise()
2383
    if result.data:
2384
      if constants.PROTOCOL_VERSION == result.data:
2385
        logging.info("Communication to node %s fine, sw version %s match",
2386
                     node, result.data)
2387
      else:
2388
        raise errors.OpExecError("Version mismatch master version %s,"
2389
                                 " node version %s" %
2390
                                 (constants.PROTOCOL_VERSION, result.data))
2391
    else:
2392
      raise errors.OpExecError("Cannot get version from the new node")
2393

    
2394
    # setup ssh on node
2395
    logging.info("Copy ssh key to node %s", node)
2396
    priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
2397
    keyarray = []
2398
    keyfiles = [constants.SSH_HOST_DSA_PRIV, constants.SSH_HOST_DSA_PUB,
2399
                constants.SSH_HOST_RSA_PRIV, constants.SSH_HOST_RSA_PUB,
2400
                priv_key, pub_key]
2401

    
2402
    for i in keyfiles:
2403
      f = open(i, 'r')
2404
      try:
2405
        keyarray.append(f.read())
2406
      finally:
2407
        f.close()
2408

    
2409
    result = self.rpc.call_node_add(node, keyarray[0], keyarray[1],
2410
                                    keyarray[2],
2411
                                    keyarray[3], keyarray[4], keyarray[5])
2412

    
2413
    msg = result.RemoteFailMsg()
2414
    if msg:
2415
      raise errors.OpExecError("Cannot transfer ssh keys to the"
2416
                               " new node: %s" % msg)
2417

    
2418
    # Add node to our /etc/hosts, and add key to known_hosts
2419
    if self.cfg.GetClusterInfo().modify_etc_hosts:
2420
      utils.AddHostToEtcHosts(new_node.name)
2421

    
2422
    if new_node.secondary_ip != new_node.primary_ip:
2423
      result = self.rpc.call_node_has_ip_address(new_node.name,
2424
                                                 new_node.secondary_ip)
2425
      if result.failed or not result.data:
2426
        raise errors.OpExecError("Node claims it doesn't have the secondary ip"
2427
                                 " you gave (%s). Please fix and re-run this"
2428
                                 " command." % new_node.secondary_ip)
2429

    
2430
    node_verify_list = [self.cfg.GetMasterNode()]
2431
    node_verify_param = {
2432
      'nodelist': [node],
2433
      # TODO: do a node-net-test as well?
2434
    }
2435

    
2436
    result = self.rpc.call_node_verify(node_verify_list, node_verify_param,
2437
                                       self.cfg.GetClusterName())
2438
    for verifier in node_verify_list:
2439
      if result[verifier].failed or not result[verifier].data:
2440
        raise errors.OpExecError("Cannot communicate with %s's node daemon"
2441
                                 " for remote verification" % verifier)
2442
      if result[verifier].data['nodelist']:
2443
        for failed in result[verifier].data['nodelist']:
2444
          feedback_fn("ssh/hostname verification failed"
2445
                      " (checking from %s): %s" %
2446
                      (verifier, result[verifier].data['nodelist'][failed]))
2447
        raise errors.OpExecError("ssh/hostname verification failed.")
2448

    
2449
    # Distribute updated /etc/hosts and known_hosts to all nodes,
2450
    # including the node just added
2451
    myself = self.cfg.GetNodeInfo(self.cfg.GetMasterNode())
2452
    dist_nodes = self.cfg.GetNodeList()
2453
    if not self.op.readd:
2454
      dist_nodes.append(node)
2455
    if myself.name in dist_nodes:
2456
      dist_nodes.remove(myself.name)
2457

    
2458
    logging.debug("Copying hosts and known_hosts to all nodes")
2459
    for fname in (constants.ETC_HOSTS, constants.SSH_KNOWN_HOSTS_FILE):
2460
      result = self.rpc.call_upload_file(dist_nodes, fname)
2461
      for to_node, to_result in result.iteritems():
2462
        if to_result.failed or not to_result.data:
2463
          logging.error("Copy of file %s to node %s failed", fname, to_node)
2464

    
2465
    to_copy = []
2466
    enabled_hypervisors = self.cfg.GetClusterInfo().enabled_hypervisors
2467
    if constants.HTS_COPY_VNC_PASSWORD.intersection(enabled_hypervisors):
2468
      to_copy.append(constants.VNC_PASSWORD_FILE)
2469

    
2470
    for fname in to_copy:
2471
      result = self.rpc.call_upload_file([node], fname)
2472
      if result[node].failed or not result[node]:
2473
        logging.error("Could not copy file %s to node %s", fname, node)
2474

    
2475
    if self.op.readd:
2476
      self.context.ReaddNode(new_node)
2477
      # make sure we redistribute the config
2478
      self.cfg.Update(new_node)
2479
      # and make sure the new node will not have old files around
2480
      if not new_node.master_candidate:
2481
        result = self.rpc.call_node_demote_from_mc(new_node.name)
2482
        msg = result.RemoteFailMsg()
2483
        if msg:
2484
          self.LogWarning("Node failed to demote itself from master"
2485
                          " candidate status: %s" % msg)
2486
    else:
2487
      self.context.AddNode(new_node)
2488

    
2489

    
2490
class LUSetNodeParams(LogicalUnit):
2491
  """Modifies the parameters of a node.
2492

2493
  """
2494
  HPATH = "node-modify"
2495
  HTYPE = constants.HTYPE_NODE
2496
  _OP_REQP = ["node_name"]
2497
  REQ_BGL = False
2498

    
2499
  def CheckArguments(self):
2500
    node_name = self.cfg.ExpandNodeName(self.op.node_name)
2501
    if node_name is None:
2502
      raise errors.OpPrereqError("Invalid node name '%s'" % self.op.node_name)
2503
    self.op.node_name = node_name
2504
    _CheckBooleanOpField(self.op, 'master_candidate')
2505
    _CheckBooleanOpField(self.op, 'offline')
2506
    _CheckBooleanOpField(self.op, 'drained')
2507
    all_mods = [self.op.offline, self.op.master_candidate, self.op.drained]
2508
    if all_mods.count(None) == 3:
2509
      raise errors.OpPrereqError("Please pass at least one modification")
2510
    if all_mods.count(True) > 1:
2511
      raise errors.OpPrereqError("Can't set the node into more than one"
2512
                                 " state at the same time")
2513

    
2514
  def ExpandNames(self):
2515
    self.needed_locks = {locking.LEVEL_NODE: self.op.node_name}
2516

    
2517
  def BuildHooksEnv(self):
2518
    """Build hooks env.
2519

2520
    This runs on the master node.
2521

2522
    """
2523
    env = {
2524
      "OP_TARGET": self.op.node_name,
2525
      "MASTER_CANDIDATE": str(self.op.master_candidate),
2526
      "OFFLINE": str(self.op.offline),
2527
      "DRAINED": str(self.op.drained),
2528
      }
2529
    nl = [self.cfg.GetMasterNode(),
2530
          self.op.node_name]
2531
    return env, nl, nl
2532

    
2533
  def CheckPrereq(self):
2534
    """Check prerequisites.
2535

2536
    This only checks the instance list against the existing names.
2537

2538
    """
2539
    node = self.node = self.cfg.GetNodeInfo(self.op.node_name)
2540

    
2541
    if (self.op.master_candidate is not None or
2542
        self.op.drained is not None or
2543
        self.op.offline is not None):
2544
      # we can't change the master's node flags
2545
      if self.op.node_name == self.cfg.GetMasterNode():
2546
        raise errors.OpPrereqError("The master role can be changed"
2547
                                   " only via masterfailover")
2548

    
2549
    if ((self.op.master_candidate == False or self.op.offline == True or
2550
         self.op.drained == True) and node.master_candidate):
2551
      cp_size = self.cfg.GetClusterInfo().candidate_pool_size
2552
      num_candidates, _ = self.cfg.GetMasterCandidateStats()
2553
      if num_candidates <= cp_size:
2554
        msg = ("Not enough master candidates (desired"
2555
               " %d, new value will be %d)" % (cp_size, num_candidates-1))
2556
        if self.op.force:
2557
          self.LogWarning(msg)
2558
        else:
2559
          raise errors.OpPrereqError(msg)
2560

    
2561
    if (self.op.master_candidate == True and
2562
        ((node.offline and not self.op.offline == False) or
2563
         (node.drained and not self.op.drained == False))):
2564
      raise errors.OpPrereqError("Node '%s' is offline or drained, can't set"
2565
                                 " to master_candidate" % node.name)
2566

    
2567
    return
2568

    
2569
  def Exec(self, feedback_fn):
2570
    """Modifies a node.
2571

2572
    """
2573
    node = self.node
2574

    
2575
    result = []
2576
    changed_mc = False
2577

    
2578
    if self.op.offline is not None:
2579
      node.offline = self.op.offline
2580
      result.append(("offline", str(self.op.offline)))
2581
      if self.op.offline == True:
2582
        if node.master_candidate:
2583
          node.master_candidate = False
2584
          changed_mc = True
2585
          result.append(("master_candidate", "auto-demotion due to offline"))
2586
        if node.drained:
2587
          node.drained = False
2588
          result.append(("drained", "clear drained status due to offline"))
2589

    
2590
    if self.op.master_candidate is not None:
2591
      node.master_candidate = self.op.master_candidate
2592
      changed_mc = True
2593
      result.append(("master_candidate", str(self.op.master_candidate)))
2594
      if self.op.master_candidate == False:
2595
        rrc = self.rpc.call_node_demote_from_mc(node.name)
2596
        msg = rrc.RemoteFailMsg()
2597
        if msg:
2598
          self.LogWarning("Node failed to demote itself: %s" % msg)
2599

    
2600
    if self.op.drained is not None:
2601
      node.drained = self.op.drained
2602
      result.append(("drained", str(self.op.drained)))
2603
      if self.op.drained == True:
2604
        if node.master_candidate:
2605
          node.master_candidate = False
2606
          changed_mc = True
2607
          result.append(("master_candidate", "auto-demotion due to drain"))
2608
          rrc = self.rpc.call_node_demote_from_mc(node.name)
2609
          msg = rrc.RemoteFailMsg()
2610
          if msg:
2611
            self.LogWarning("Node failed to demote itself: %s" % msg)
2612
        if node.offline:
2613
          node.offline = False
2614
          result.append(("offline", "clear offline status due to drain"))
2615

    
2616
    # this will trigger configuration file update, if needed
2617
    self.cfg.Update(node)
2618
    # this will trigger job queue propagation or cleanup
2619
    if changed_mc:
2620
      self.context.ReaddNode(node)
2621

    
2622
    return result
2623

    
2624

    
2625
class LUQueryClusterInfo(NoHooksLU):
2626
  """Query cluster configuration.
2627

2628
  """
2629
  _OP_REQP = []
2630
  REQ_BGL = False
2631

    
2632
  def ExpandNames(self):
2633
    self.needed_locks = {}
2634

    
2635
  def CheckPrereq(self):
2636
    """No prerequsites needed for this LU.
2637

2638
    """
2639
    pass
2640

    
2641
  def Exec(self, feedback_fn):
2642
    """Return cluster config.
2643

2644
    """
2645
    cluster = self.cfg.GetClusterInfo()
2646
    result = {
2647
      "software_version": constants.RELEASE_VERSION,
2648
      "protocol_version": constants.PROTOCOL_VERSION,
2649
      "config_version": constants.CONFIG_VERSION,
2650
      "os_api_version": constants.OS_API_VERSION,
2651
      "export_version": constants.EXPORT_VERSION,
2652
      "architecture": (platform.architecture()[0], platform.machine()),
2653
      "name": cluster.cluster_name,
2654
      "master": cluster.master_node,
2655
      "default_hypervisor": cluster.default_hypervisor,
2656
      "enabled_hypervisors": cluster.enabled_hypervisors,
2657
      "hvparams": dict([(hypervisor_name, cluster.hvparams[hypervisor_name])
2658
                        for hypervisor_name in cluster.enabled_hypervisors]),
2659
      "beparams": cluster.beparams,
2660
      "candidate_pool_size": cluster.candidate_pool_size,
2661
      "default_bridge": cluster.default_bridge,
2662
      "master_netdev": cluster.master_netdev,
2663
      "volume_group_name": cluster.volume_group_name,
2664
      "file_storage_dir": cluster.file_storage_dir,
2665
      "tags": list(cluster.GetTags()),
2666
      }
2667

    
2668
    return result
2669

    
2670

    
2671
class LUQueryConfigValues(NoHooksLU):
2672
  """Return configuration values.
2673

2674
  """
2675
  _OP_REQP = []
2676
  REQ_BGL = False
2677
  _FIELDS_DYNAMIC = utils.FieldSet()
2678
  _FIELDS_STATIC = utils.FieldSet("cluster_name", "master_node", "drain_flag")
2679

    
2680
  def ExpandNames(self):
2681
    self.needed_locks = {}
2682

    
2683
    _CheckOutputFields(static=self._FIELDS_STATIC,
2684
                       dynamic=self._FIELDS_DYNAMIC,
2685
                       selected=self.op.output_fields)
2686

    
2687
  def CheckPrereq(self):
2688
    """No prerequisites.
2689

2690
    """
2691
    pass
2692

    
2693
  def Exec(self, feedback_fn):
2694
    """Dump a representation of the cluster config to the standard output.
2695

2696
    """
2697
    values = []
2698
    for field in self.op.output_fields:
2699
      if field == "cluster_name":
2700
        entry = self.cfg.GetClusterName()
2701
      elif field == "master_node":
2702
        entry = self.cfg.GetMasterNode()
2703
      elif field == "drain_flag":
2704
        entry = os.path.exists(constants.JOB_QUEUE_DRAIN_FILE)
2705
      else:
2706
        raise errors.ParameterError(field)
2707
      values.append(entry)
2708
    return values
2709

    
2710

    
2711
class LUActivateInstanceDisks(NoHooksLU):
2712
  """Bring up an instance's disks.
2713

2714
  """
2715
  _OP_REQP = ["instance_name"]
2716
  REQ_BGL = False
2717

    
2718
  def ExpandNames(self):
2719
    self._ExpandAndLockInstance()
2720
    self.needed_locks[locking.LEVEL_NODE] = []
2721
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2722

    
2723
  def DeclareLocks(self, level):
2724
    if level == locking.LEVEL_NODE:
2725
      self._LockInstancesNodes()
2726

    
2727
  def CheckPrereq(self):
2728
    """Check prerequisites.
2729

2730
    This checks that the instance is in the cluster.
2731

2732
    """
2733
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2734
    assert self.instance is not None, \
2735
      "Cannot retrieve locked instance %s" % self.op.instance_name
2736
    _CheckNodeOnline(self, self.instance.primary_node)
2737
    if not hasattr(self.op, "ignore_size"):
2738
      self.op.ignore_size = False
2739

    
2740
  def Exec(self, feedback_fn):
2741
    """Activate the disks.
2742

2743
    """
2744
    disks_ok, disks_info = \
2745
              _AssembleInstanceDisks(self, self.instance,
2746
                                     ignore_size=self.op.ignore_size)
2747
    if not disks_ok:
2748
      raise errors.OpExecError("Cannot activate block devices")
2749

    
2750
    return disks_info
2751

    
2752

    
2753
def _AssembleInstanceDisks(lu, instance, ignore_secondaries=False,
2754
                           ignore_size=False):
2755
  """Prepare the block devices for an instance.
2756

2757
  This sets up the block devices on all nodes.
2758

2759
  @type lu: L{LogicalUnit}
2760
  @param lu: the logical unit on whose behalf we execute
2761
  @type instance: L{objects.Instance}
2762
  @param instance: the instance for whose disks we assemble
2763
  @type ignore_secondaries: boolean
2764
  @param ignore_secondaries: if true, errors on secondary nodes
2765
      won't result in an error return from the function
2766
  @type ignore_size: boolean
2767
  @param ignore_size: if true, the current known size of the disk
2768
      will not be used during the disk activation, useful for cases
2769
      when the size is wrong
2770
  @return: False if the operation failed, otherwise a list of
2771
      (host, instance_visible_name, node_visible_name)
2772
      with the mapping from node devices to instance devices
2773

2774
  """
2775
  device_info = []
2776
  disks_ok = True
2777
  iname = instance.name
2778
  # With the two passes mechanism we try to reduce the window of
2779
  # opportunity for the race condition of switching DRBD to primary
2780
  # before handshaking occured, but we do not eliminate it
2781

    
2782
  # The proper fix would be to wait (with some limits) until the
2783
  # connection has been made and drbd transitions from WFConnection
2784
  # into any other network-connected state (Connected, SyncTarget,
2785
  # SyncSource, etc.)
2786

    
2787
  # 1st pass, assemble on all nodes in secondary mode
2788
  for inst_disk in instance.disks:
2789
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
2790
      if ignore_size:
2791
        node_disk = node_disk.Copy()
2792
        node_disk.UnsetSize()
2793
      lu.cfg.SetDiskID(node_disk, node)
2794
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, False)
2795
      msg = result.RemoteFailMsg()
2796
      if msg:
2797
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
2798
                           " (is_primary=False, pass=1): %s",
2799
                           inst_disk.iv_name, node, msg)
2800
        if not ignore_secondaries:
2801
          disks_ok = False
2802

    
2803
  # FIXME: race condition on drbd migration to primary
2804

    
2805
  # 2nd pass, do only the primary node
2806
  for inst_disk in instance.disks:
2807
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
2808
      if node != instance.primary_node:
2809
        continue
2810
      if ignore_size:
2811
        node_disk = node_disk.Copy()
2812
        node_disk.UnsetSize()
2813
      lu.cfg.SetDiskID(node_disk, node)
2814
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, True)
2815
      msg = result.RemoteFailMsg()
2816
      if msg:
2817
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
2818
                           " (is_primary=True, pass=2): %s",
2819
                           inst_disk.iv_name, node, msg)
2820
        disks_ok = False
2821
    device_info.append((instance.primary_node, inst_disk.iv_name,
2822
                        result.payload))
2823

    
2824
  # leave the disks configured for the primary node
2825
  # this is a workaround that would be fixed better by
2826
  # improving the logical/physical id handling
2827
  for disk in instance.disks:
2828
    lu.cfg.SetDiskID(disk, instance.primary_node)
2829

    
2830
  return disks_ok, device_info
2831

    
2832

    
2833
def _StartInstanceDisks(lu, instance, force):
2834
  """Start the disks of an instance.
2835

2836
  """
2837
  disks_ok, _ = _AssembleInstanceDisks(lu, instance,
2838
                                           ignore_secondaries=force)
2839
  if not disks_ok:
2840
    _ShutdownInstanceDisks(lu, instance)
2841
    if force is not None and not force:
2842
      lu.proc.LogWarning("", hint="If the message above refers to a"
2843
                         " secondary node,"
2844
                         " you can retry the operation using '--force'.")
2845
    raise errors.OpExecError("Disk consistency error")
2846

    
2847

    
2848
class LUDeactivateInstanceDisks(NoHooksLU):
2849
  """Shutdown an instance's disks.
2850

2851
  """
2852
  _OP_REQP = ["instance_name"]
2853
  REQ_BGL = False
2854

    
2855
  def ExpandNames(self):
2856
    self._ExpandAndLockInstance()
2857
    self.needed_locks[locking.LEVEL_NODE] = []
2858
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2859

    
2860
  def DeclareLocks(self, level):
2861
    if level == locking.LEVEL_NODE:
2862
      self._LockInstancesNodes()
2863

    
2864
  def CheckPrereq(self):
2865
    """Check prerequisites.
2866

2867
    This checks that the instance is in the cluster.
2868

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

    
2874
  def Exec(self, feedback_fn):
2875
    """Deactivate the disks
2876

2877
    """
2878
    instance = self.instance
2879
    _SafeShutdownInstanceDisks(self, instance)
2880

    
2881

    
2882
def _SafeShutdownInstanceDisks(lu, instance):
2883
  """Shutdown block devices of an instance.
2884

2885
  This function checks if an instance is running, before calling
2886
  _ShutdownInstanceDisks.
2887

2888
  """
2889
  ins_l = lu.rpc.call_instance_list([instance.primary_node],
2890
                                      [instance.hypervisor])
2891
  ins_l = ins_l[instance.primary_node]
2892
  if ins_l.failed or not isinstance(ins_l.data, list):
2893
    raise errors.OpExecError("Can't contact node '%s'" %
2894
                             instance.primary_node)
2895

    
2896
  if instance.name in ins_l.data:
2897
    raise errors.OpExecError("Instance is running, can't shutdown"
2898
                             " block devices.")
2899

    
2900
  _ShutdownInstanceDisks(lu, instance)
2901

    
2902

    
2903
def _ShutdownInstanceDisks(lu, instance, ignore_primary=False):
2904
  """Shutdown block devices of an instance.
2905

2906
  This does the shutdown on all nodes of the instance.
2907

2908
  If the ignore_primary is false, errors on the primary node are
2909
  ignored.
2910

2911
  """
2912
  all_result = True
2913
  for disk in instance.disks:
2914
    for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
2915
      lu.cfg.SetDiskID(top_disk, node)
2916
      result = lu.rpc.call_blockdev_shutdown(node, top_disk)
2917
      msg = result.RemoteFailMsg()
2918
      if msg:
2919
        lu.LogWarning("Could not shutdown block device %s on node %s: %s",
2920
                      disk.iv_name, node, msg)
2921
        if not ignore_primary or node != instance.primary_node:
2922
          all_result = False
2923
  return all_result
2924

    
2925

    
2926
def _CheckNodeFreeMemory(lu, node, reason, requested, hypervisor_name):
2927
  """Checks if a node has enough free memory.
2928

2929
  This function check if a given node has the needed amount of free
2930
  memory. In case the node has less memory or we cannot get the
2931
  information from the node, this function raise an OpPrereqError
2932
  exception.
2933

2934
  @type lu: C{LogicalUnit}
2935
  @param lu: a logical unit from which we get configuration data
2936
  @type node: C{str}
2937
  @param node: the node to check
2938
  @type reason: C{str}
2939
  @param reason: string to use in the error message
2940
  @type requested: C{int}
2941
  @param requested: the amount of memory in MiB to check for
2942
  @type hypervisor_name: C{str}
2943
  @param hypervisor_name: the hypervisor to ask for memory stats
2944
  @raise errors.OpPrereqError: if the node doesn't have enough memory, or
2945
      we cannot check the node
2946

2947
  """
2948
  nodeinfo = lu.rpc.call_node_info([node], lu.cfg.GetVGName(), hypervisor_name)
2949
  nodeinfo[node].Raise()
2950
  free_mem = nodeinfo[node].data.get('memory_free')
2951
  if not isinstance(free_mem, int):
2952
    raise errors.OpPrereqError("Can't compute free memory on node %s, result"
2953
                             " was '%s'" % (node, free_mem))
2954
  if requested > free_mem:
2955
    raise errors.OpPrereqError("Not enough memory on node %s for %s:"
2956
                             " needed %s MiB, available %s MiB" %
2957
                             (node, reason, requested, free_mem))
2958

    
2959

    
2960
class LUStartupInstance(LogicalUnit):
2961
  """Starts an instance.
2962

2963
  """
2964
  HPATH = "instance-start"
2965
  HTYPE = constants.HTYPE_INSTANCE
2966
  _OP_REQP = ["instance_name", "force"]
2967
  REQ_BGL = False
2968

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

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

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

2977
    """
2978
    env = {
2979
      "FORCE": self.op.force,
2980
      }
2981
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
2982
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2983
    return env, nl, nl
2984

    
2985
  def CheckPrereq(self):
2986
    """Check prerequisites.
2987

2988
    This checks that the instance is in the cluster.
2989

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

    
2995
    # extra beparams
2996
    self.beparams = getattr(self.op, "beparams", {})
2997
    if self.beparams:
2998
      if not isinstance(self.beparams, dict):
2999
        raise errors.OpPrereqError("Invalid beparams passed: %s, expected"
3000
                                   " dict" % (type(self.beparams), ))
3001
      # fill the beparams dict
3002
      utils.ForceDictType(self.beparams, constants.BES_PARAMETER_TYPES)
3003
      self.op.beparams = self.beparams
3004

    
3005
    # extra hvparams
3006
    self.hvparams = getattr(self.op, "hvparams", {})
3007
    if self.hvparams:
3008
      if not isinstance(self.hvparams, dict):
3009
        raise errors.OpPrereqError("Invalid hvparams passed: %s, expected"
3010
                                   " dict" % (type(self.hvparams), ))
3011

    
3012
      # check hypervisor parameter syntax (locally)
3013
      cluster = self.cfg.GetClusterInfo()
3014
      utils.ForceDictType(self.hvparams, constants.HVS_PARAMETER_TYPES)
3015
      filled_hvp = cluster.FillDict(cluster.hvparams[instance.hypervisor],
3016
                                    instance.hvparams)
3017
      filled_hvp.update(self.hvparams)
3018
      hv_type = hypervisor.GetHypervisor(instance.hypervisor)
3019
      hv_type.CheckParameterSyntax(filled_hvp)
3020
      _CheckHVParams(self, instance.all_nodes, instance.hypervisor, filled_hvp)
3021
      self.op.hvparams = self.hvparams
3022

    
3023
    _CheckNodeOnline(self, instance.primary_node)
3024

    
3025
    bep = self.cfg.GetClusterInfo().FillBE(instance)
3026
    # check bridges existence
3027
    _CheckInstanceBridgesExist(self, instance)
3028

    
3029
    remote_info = self.rpc.call_instance_info(instance.primary_node,
3030
                                              instance.name,
3031
                                              instance.hypervisor)
3032
    remote_info.Raise()
3033
    if not remote_info.data:
3034
      _CheckNodeFreeMemory(self, instance.primary_node,
3035
                           "starting instance %s" % instance.name,
3036
                           bep[constants.BE_MEMORY], instance.hypervisor)
3037

    
3038
  def Exec(self, feedback_fn):
3039
    """Start the instance.
3040

3041
    """
3042
    instance = self.instance
3043
    force = self.op.force
3044

    
3045
    self.cfg.MarkInstanceUp(instance.name)
3046

    
3047
    node_current = instance.primary_node
3048

    
3049
    _StartInstanceDisks(self, instance, force)
3050

    
3051
    result = self.rpc.call_instance_start(node_current, instance,
3052
                                          self.hvparams, self.beparams)
3053
    msg = result.RemoteFailMsg()
3054
    if msg:
3055
      _ShutdownInstanceDisks(self, instance)
3056
      raise errors.OpExecError("Could not start instance: %s" % msg)
3057

    
3058

    
3059
class LURebootInstance(LogicalUnit):
3060
  """Reboot an instance.
3061

3062
  """
3063
  HPATH = "instance-reboot"
3064
  HTYPE = constants.HTYPE_INSTANCE
3065
  _OP_REQP = ["instance_name", "ignore_secondaries", "reboot_type"]
3066
  REQ_BGL = False
3067

    
3068
  def ExpandNames(self):
3069
    if self.op.reboot_type not in [constants.INSTANCE_REBOOT_SOFT,
3070
                                   constants.INSTANCE_REBOOT_HARD,
3071
                                   constants.INSTANCE_REBOOT_FULL]:
3072
      raise errors.ParameterError("reboot type not in [%s, %s, %s]" %
3073
                                  (constants.INSTANCE_REBOOT_SOFT,
3074
                                   constants.INSTANCE_REBOOT_HARD,
3075
                                   constants.INSTANCE_REBOOT_FULL))
3076
    self._ExpandAndLockInstance()
3077

    
3078
  def BuildHooksEnv(self):
3079
    """Build hooks env.
3080

3081
    This runs on master, primary and secondary nodes of the instance.
3082

3083
    """
3084
    env = {
3085
      "IGNORE_SECONDARIES": self.op.ignore_secondaries,
3086
      "REBOOT_TYPE": self.op.reboot_type,
3087
      }
3088
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3089
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3090
    return env, nl, nl
3091

    
3092
  def CheckPrereq(self):
3093
    """Check prerequisites.
3094

3095
    This checks that the instance is in the cluster.
3096

3097
    """
3098
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3099
    assert self.instance is not None, \
3100
      "Cannot retrieve locked instance %s" % self.op.instance_name
3101

    
3102
    _CheckNodeOnline(self, instance.primary_node)
3103

    
3104
    # check bridges existence
3105
    _CheckInstanceBridgesExist(self, instance)
3106

    
3107
  def Exec(self, feedback_fn):
3108
    """Reboot the instance.
3109

3110
    """
3111
    instance = self.instance
3112
    ignore_secondaries = self.op.ignore_secondaries
3113
    reboot_type = self.op.reboot_type
3114

    
3115
    node_current = instance.primary_node
3116

    
3117
    if reboot_type in [constants.INSTANCE_REBOOT_SOFT,
3118
                       constants.INSTANCE_REBOOT_HARD]:
3119
      for disk in instance.disks:
3120
        self.cfg.SetDiskID(disk, node_current)
3121
      result = self.rpc.call_instance_reboot(node_current, instance,
3122
                                             reboot_type)
3123
      msg = result.RemoteFailMsg()
3124
      if msg:
3125
        raise errors.OpExecError("Could not reboot instance: %s" % msg)
3126
    else:
3127
      result = self.rpc.call_instance_shutdown(node_current, instance)
3128
      msg = result.RemoteFailMsg()
3129
      if msg:
3130
        raise errors.OpExecError("Could not shutdown instance for"
3131
                                 " full reboot: %s" % msg)
3132
      _ShutdownInstanceDisks(self, instance)
3133
      _StartInstanceDisks(self, instance, ignore_secondaries)
3134
      result = self.rpc.call_instance_start(node_current, instance, None, None)
3135
      msg = result.RemoteFailMsg()
3136
      if msg:
3137
        _ShutdownInstanceDisks(self, instance)
3138
        raise errors.OpExecError("Could not start instance for"
3139
                                 " full reboot: %s" % msg)
3140

    
3141
    self.cfg.MarkInstanceUp(instance.name)
3142

    
3143

    
3144
class LUShutdownInstance(LogicalUnit):
3145
  """Shutdown an instance.
3146

3147
  """
3148
  HPATH = "instance-stop"
3149
  HTYPE = constants.HTYPE_INSTANCE
3150
  _OP_REQP = ["instance_name"]
3151
  REQ_BGL = False
3152

    
3153
  def ExpandNames(self):
3154
    self._ExpandAndLockInstance()
3155

    
3156
  def BuildHooksEnv(self):
3157
    """Build hooks env.
3158

3159
    This runs on master, primary and secondary nodes of the instance.
3160

3161
    """
3162
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3163
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3164
    return env, nl, nl
3165

    
3166
  def CheckPrereq(self):
3167
    """Check prerequisites.
3168

3169
    This checks that the instance is in the cluster.
3170

3171
    """
3172
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3173
    assert self.instance is not None, \
3174
      "Cannot retrieve locked instance %s" % self.op.instance_name
3175
    _CheckNodeOnline(self, self.instance.primary_node)
3176

    
3177
  def Exec(self, feedback_fn):
3178
    """Shutdown the instance.
3179

3180
    """
3181
    instance = self.instance
3182
    node_current = instance.primary_node
3183
    self.cfg.MarkInstanceDown(instance.name)
3184
    result = self.rpc.call_instance_shutdown(node_current, instance)
3185
    msg = result.RemoteFailMsg()
3186
    if msg:
3187
      self.proc.LogWarning("Could not shutdown instance: %s" % msg)
3188

    
3189
    _ShutdownInstanceDisks(self, instance)
3190

    
3191

    
3192
class LUReinstallInstance(LogicalUnit):
3193
  """Reinstall an instance.
3194

3195
  """
3196
  HPATH = "instance-reinstall"
3197
  HTYPE = constants.HTYPE_INSTANCE
3198
  _OP_REQP = ["instance_name"]
3199
  REQ_BGL = False
3200

    
3201
  def ExpandNames(self):
3202
    self._ExpandAndLockInstance()
3203

    
3204
  def BuildHooksEnv(self):
3205
    """Build hooks env.
3206

3207
    This runs on master, primary and secondary nodes of the instance.
3208

3209
    """
3210
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3211
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3212
    return env, nl, nl
3213

    
3214
  def CheckPrereq(self):
3215
    """Check prerequisites.
3216

3217
    This checks that the instance is in the cluster and is not running.
3218

3219
    """
3220
    instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3221
    assert instance is not None, \
3222
      "Cannot retrieve locked instance %s" % self.op.instance_name
3223
    _CheckNodeOnline(self, instance.primary_node)
3224

    
3225
    if instance.disk_template == constants.DT_DISKLESS:
3226
      raise errors.OpPrereqError("Instance '%s' has no disks" %
3227
                                 self.op.instance_name)
3228
    if instance.admin_up:
3229
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
3230
                                 self.op.instance_name)
3231
    remote_info = self.rpc.call_instance_info(instance.primary_node,
3232
                                              instance.name,
3233
                                              instance.hypervisor)
3234
    remote_info.Raise()
3235
    if remote_info.data:
3236
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
3237
                                 (self.op.instance_name,
3238
                                  instance.primary_node))
3239

    
3240
    self.op.os_type = getattr(self.op, "os_type", None)
3241
    if self.op.os_type is not None:
3242
      # OS verification
3243
      pnode = self.cfg.GetNodeInfo(
3244
        self.cfg.ExpandNodeName(instance.primary_node))
3245
      if pnode is None:
3246
        raise errors.OpPrereqError("Primary node '%s' is unknown" %
3247
                                   self.op.pnode)
3248
      result = self.rpc.call_os_get(pnode.name, self.op.os_type)
3249
      result.Raise()
3250
      if not isinstance(result.data, objects.OS):
3251
        raise errors.OpPrereqError("OS '%s' not in supported OS list for"
3252
                                   " primary node"  % self.op.os_type)
3253

    
3254
    self.instance = instance
3255

    
3256
  def Exec(self, feedback_fn):
3257
    """Reinstall the instance.
3258

3259
    """
3260
    inst = self.instance
3261

    
3262
    if self.op.os_type is not None:
3263
      feedback_fn("Changing OS to '%s'..." % self.op.os_type)
3264
      inst.os = self.op.os_type
3265
      self.cfg.Update(inst)
3266

    
3267
    _StartInstanceDisks(self, inst, None)
3268
    try:
3269
      feedback_fn("Running the instance OS create scripts...")
3270
      result = self.rpc.call_instance_os_add(inst.primary_node, inst)
3271
      msg = result.RemoteFailMsg()
3272
      if msg:
3273
        raise errors.OpExecError("Could not install OS for instance %s"
3274
                                 " on node %s: %s" %
3275
                                 (inst.name, inst.primary_node, msg))
3276
    finally:
3277
      _ShutdownInstanceDisks(self, inst)
3278

    
3279

    
3280
class LURenameInstance(LogicalUnit):
3281
  """Rename an instance.
3282

3283
  """
3284
  HPATH = "instance-rename"
3285
  HTYPE = constants.HTYPE_INSTANCE
3286
  _OP_REQP = ["instance_name", "new_name"]
3287

    
3288
  def BuildHooksEnv(self):
3289
    """Build hooks env.
3290

3291
    This runs on master, primary and secondary nodes of the instance.
3292

3293
    """
3294
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3295
    env["INSTANCE_NEW_NAME"] = self.op.new_name
3296
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3297
    return env, nl, nl
3298

    
3299
  def CheckPrereq(self):
3300
    """Check prerequisites.
3301

3302
    This checks that the instance is in the cluster and is not running.
3303

3304
    """
3305
    instance = self.cfg.GetInstanceInfo(
3306
      self.cfg.ExpandInstanceName(self.op.instance_name))
3307
    if instance is None:
3308
      raise errors.OpPrereqError("Instance '%s' not known" %
3309
                                 self.op.instance_name)
3310
    _CheckNodeOnline(self, instance.primary_node)
3311

    
3312
    if instance.admin_up:
3313
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
3314
                                 self.op.instance_name)
3315
    remote_info = self.rpc.call_instance_info(instance.primary_node,
3316
                                              instance.name,
3317
                                              instance.hypervisor)
3318
    remote_info.Raise()
3319
    if remote_info.data:
3320
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
3321
                                 (self.op.instance_name,
3322
                                  instance.primary_node))
3323
    self.instance = instance
3324

    
3325
    # new name verification
3326
    name_info = utils.HostInfo(self.op.new_name)
3327

    
3328
    self.op.new_name = new_name = name_info.name
3329
    instance_list = self.cfg.GetInstanceList()
3330
    if new_name in instance_list:
3331
      raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
3332
                                 new_name)
3333

    
3334
    if not getattr(self.op, "ignore_ip", False):
3335
      if utils.TcpPing(name_info.ip, constants.DEFAULT_NODED_PORT):
3336
        raise errors.OpPrereqError("IP %s of instance %s already in use" %
3337
                                   (name_info.ip, new_name))
3338

    
3339

    
3340
  def Exec(self, feedback_fn):
3341
    """Reinstall the instance.
3342

3343
    """
3344
    inst = self.instance
3345
    old_name = inst.name
3346

    
3347
    if inst.disk_template == constants.DT_FILE:
3348
      old_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
3349

    
3350
    self.cfg.RenameInstance(inst.name, self.op.new_name)
3351
    # Change the instance lock. This is definitely safe while we hold the BGL
3352
    self.context.glm.remove(locking.LEVEL_INSTANCE, old_name)
3353
    self.context.glm.add(locking.LEVEL_INSTANCE, self.op.new_name)
3354

    
3355
    # re-read the instance from the configuration after rename
3356
    inst = self.cfg.GetInstanceInfo(self.op.new_name)
3357

    
3358
    if inst.disk_template == constants.DT_FILE:
3359
      new_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
3360
      result = self.rpc.call_file_storage_dir_rename(inst.primary_node,
3361
                                                     old_file_storage_dir,
3362
                                                     new_file_storage_dir)
3363
      result.Raise()
3364
      if not result.data:
3365
        raise errors.OpExecError("Could not connect to node '%s' to rename"
3366
                                 " directory '%s' to '%s' (but the instance"
3367
                                 " has been renamed in Ganeti)" % (
3368
                                 inst.primary_node, old_file_storage_dir,
3369
                                 new_file_storage_dir))
3370

    
3371
      if not result.data[0]:
3372
        raise errors.OpExecError("Could not rename directory '%s' to '%s'"
3373
                                 " (but the instance has been renamed in"
3374
                                 " Ganeti)" % (old_file_storage_dir,
3375
                                               new_file_storage_dir))
3376

    
3377
    _StartInstanceDisks(self, inst, None)
3378
    try:
3379
      result = self.rpc.call_instance_run_rename(inst.primary_node, inst,
3380
                                                 old_name)
3381
      msg = result.RemoteFailMsg()
3382
      if msg:
3383
        msg = ("Could not run OS rename script for instance %s on node %s"
3384
               " (but the instance has been renamed in Ganeti): %s" %
3385
               (inst.name, inst.primary_node, msg))
3386
        self.proc.LogWarning(msg)
3387
    finally:
3388
      _ShutdownInstanceDisks(self, inst)
3389

    
3390

    
3391
class LURemoveInstance(LogicalUnit):
3392
  """Remove an instance.
3393

3394
  """
3395
  HPATH = "instance-remove"
3396
  HTYPE = constants.HTYPE_INSTANCE
3397
  _OP_REQP = ["instance_name", "ignore_failures"]
3398
  REQ_BGL = False
3399

    
3400
  def ExpandNames(self):
3401
    self._ExpandAndLockInstance()
3402
    self.needed_locks[locking.LEVEL_NODE] = []
3403
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3404

    
3405
  def DeclareLocks(self, level):
3406
    if level == locking.LEVEL_NODE:
3407
      self._LockInstancesNodes()
3408

    
3409
  def BuildHooksEnv(self):
3410
    """Build hooks env.
3411

3412
    This runs on master, primary and secondary nodes of the instance.
3413

3414
    """
3415
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3416
    nl = [self.cfg.GetMasterNode()]
3417
    return env, nl, nl
3418

    
3419
  def CheckPrereq(self):
3420
    """Check prerequisites.
3421

3422
    This checks that the instance is in the cluster.
3423

3424
    """
3425
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3426
    assert self.instance is not None, \
3427
      "Cannot retrieve locked instance %s" % self.op.instance_name
3428

    
3429
  def Exec(self, feedback_fn):
3430
    """Remove the instance.
3431

3432
    """
3433
    instance = self.instance
3434
    logging.info("Shutting down instance %s on node %s",
3435
                 instance.name, instance.primary_node)
3436

    
3437
    result = self.rpc.call_instance_shutdown(instance.primary_node, instance)
3438
    msg = result.RemoteFailMsg()
3439
    if msg:
3440
      if self.op.ignore_failures:
3441
        feedback_fn("Warning: can't shutdown instance: %s" % msg)
3442
      else:
3443
        raise errors.OpExecError("Could not shutdown instance %s on"
3444
                                 " node %s: %s" %
3445
                                 (instance.name, instance.primary_node, msg))
3446

    
3447
    logging.info("Removing block devices for instance %s", instance.name)
3448

    
3449
    if not _RemoveDisks(self, instance):
3450
      if self.op.ignore_failures:
3451
        feedback_fn("Warning: can't remove instance's disks")
3452
      else:
3453
        raise errors.OpExecError("Can't remove instance's disks")
3454

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

    
3457
    self.cfg.RemoveInstance(instance.name)
3458
    self.remove_locks[locking.LEVEL_INSTANCE] = instance.name
3459

    
3460

    
3461
class LUQueryInstances(NoHooksLU):
3462
  """Logical unit for querying instances.
3463

3464
  """
3465
  _OP_REQP = ["output_fields", "names", "use_locking"]
3466
  REQ_BGL = False
3467
  _FIELDS_STATIC = utils.FieldSet(*["name", "os", "pnode", "snodes",
3468
                                    "admin_state",
3469
                                    "disk_template", "ip", "mac", "bridge",
3470
                                    "sda_size", "sdb_size", "vcpus", "tags",
3471
                                    "network_port", "beparams",
3472
                                    r"(disk)\.(size)/([0-9]+)",
3473
                                    r"(disk)\.(sizes)", "disk_usage",
3474
                                    r"(nic)\.(mac|ip|bridge)/([0-9]+)",
3475
                                    r"(nic)\.(macs|ips|bridges)",
3476
                                    r"(disk|nic)\.(count)",
3477
                                    "serial_no", "hypervisor", "hvparams",] +
3478
                                  ["hv/%s" % name
3479
                                   for name in constants.HVS_PARAMETERS] +
3480
                                  ["be/%s" % name
3481
                                   for name in constants.BES_PARAMETERS])
3482
  _FIELDS_DYNAMIC = utils.FieldSet("oper_state", "oper_ram", "status")
3483

    
3484

    
3485
  def ExpandNames(self):
3486
    _CheckOutputFields(static=self._FIELDS_STATIC,
3487
                       dynamic=self._FIELDS_DYNAMIC,
3488
                       selected=self.op.output_fields)
3489

    
3490
    self.needed_locks = {}
3491
    self.share_locks[locking.LEVEL_INSTANCE] = 1
3492
    self.share_locks[locking.LEVEL_NODE] = 1
3493

    
3494
    if self.op.names:
3495
      self.wanted = _GetWantedInstances(self, self.op.names)
3496
    else:
3497
      self.wanted = locking.ALL_SET
3498

    
3499
    self.do_node_query = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
3500
    self.do_locking = self.do_node_query and self.op.use_locking
3501
    if self.do_locking:
3502
      self.needed_locks[locking.LEVEL_INSTANCE] = self.wanted
3503
      self.needed_locks[locking.LEVEL_NODE] = []
3504
      self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3505

    
3506
  def DeclareLocks(self, level):
3507
    if level == locking.LEVEL_NODE and self.do_locking:
3508
      self._LockInstancesNodes()
3509

    
3510
  def CheckPrereq(self):
3511
    """Check prerequisites.
3512

3513
    """
3514
    pass
3515

    
3516
  def Exec(self, feedback_fn):
3517
    """Computes the list of nodes and their attributes.
3518

3519
    """
3520
    all_info = self.cfg.GetAllInstancesInfo()
3521
    if self.wanted == locking.ALL_SET:
3522
      # caller didn't specify instance names, so ordering is not important
3523
      if self.do_locking:
3524
        instance_names = self.acquired_locks[locking.LEVEL_INSTANCE]
3525
      else:
3526
        instance_names = all_info.keys()
3527
      instance_names = utils.NiceSort(instance_names)
3528
    else:
3529
      # caller did specify names, so we must keep the ordering
3530
      if self.do_locking:
3531
        tgt_set = self.acquired_locks[locking.LEVEL_INSTANCE]
3532
      else:
3533
        tgt_set = all_info.keys()
3534
      missing = set(self.wanted).difference(tgt_set)
3535
      if missing:
3536
        raise errors.OpExecError("Some instances were removed before"
3537
                                 " retrieving their data: %s" % missing)
3538
      instance_names = self.wanted
3539

    
3540
    instance_list = [all_info[iname] for iname in instance_names]
3541

    
3542
    # begin data gathering
3543

    
3544
    nodes = frozenset([inst.primary_node for inst in instance_list])
3545
    hv_list = list(set([inst.hypervisor for inst in instance_list]))
3546

    
3547
    bad_nodes = []
3548
    off_nodes = []
3549
    if self.do_node_query:
3550
      live_data = {}
3551
      node_data = self.rpc.call_all_instances_info(nodes, hv_list)
3552
      for name in nodes:
3553
        result = node_data[name]
3554
        if result.offline:
3555
          # offline nodes will be in both lists
3556
          off_nodes.append(name)
3557
        if result.failed:
3558
          bad_nodes.append(name)
3559
        else:
3560
          if result.data:
3561
            live_data.update(result.data)
3562
            # else no instance is alive
3563
    else:
3564
      live_data = dict([(name, {}) for name in instance_names])
3565

    
3566
    # end data gathering
3567

    
3568
    HVPREFIX = "hv/"
3569
    BEPREFIX = "be/"
3570
    output = []
3571
    for instance in instance_list:
3572
      iout = []
3573
      i_hv = self.cfg.GetClusterInfo().FillHV(instance)
3574
      i_be = self.cfg.GetClusterInfo().FillBE(instance)
3575
      for field in self.op.output_fields:
3576
        st_match = self._FIELDS_STATIC.Matches(field)
3577
        if field == "name":
3578
          val = instance.name
3579
        elif field == "os":
3580
          val = instance.os
3581
        elif field == "pnode":
3582
          val = instance.primary_node
3583
        elif field == "snodes":
3584
          val = list(instance.secondary_nodes)
3585
        elif field == "admin_state":
3586
          val = instance.admin_up
3587
        elif field == "oper_state":
3588
          if instance.primary_node in bad_nodes:
3589
            val = None
3590
          else:
3591
            val = bool(live_data.get(instance.name))
3592
        elif field == "status":
3593
          if instance.primary_node in off_nodes:
3594
            val = "ERROR_nodeoffline"
3595
          elif instance.primary_node in bad_nodes:
3596
            val = "ERROR_nodedown"
3597
          else:
3598
            running = bool(live_data.get(instance.name))
3599
            if running:
3600
              if instance.admin_up:
3601
                val = "running"
3602
              else:
3603
                val = "ERROR_up"
3604
            else:
3605
              if instance.admin_up:
3606
                val = "ERROR_down"
3607
              else:
3608
                val = "ADMIN_down"
3609
        elif field == "oper_ram":
3610
          if instance.primary_node in bad_nodes:
3611
            val = None
3612
          elif instance.name in live_data:
3613
            val = live_data[instance.name].get("memory", "?")
3614
          else:
3615
            val = "-"
3616
        elif field == "vcpus":
3617
          val = i_be[constants.BE_VCPUS]
3618
        elif field == "disk_template":
3619
          val = instance.disk_template
3620
        elif field == "ip":
3621
          if instance.nics:
3622
            val = instance.nics[0].ip
3623
          else:
3624
            val = None
3625
        elif field == "bridge":
3626
          if instance.nics:
3627
            val = instance.nics[0].bridge
3628
          else:
3629
            val = None
3630
        elif field == "mac":
3631
          if instance.nics:
3632
            val = instance.nics[0].mac
3633
          else:
3634
            val = None
3635
        elif field == "sda_size" or field == "sdb_size":
3636
          idx = ord(field[2]) - ord('a')
3637
          try:
3638
            val = instance.FindDisk(idx).size
3639
          except errors.OpPrereqError:
3640
            val = None
3641
        elif field == "disk_usage": # total disk usage per node
3642
          disk_sizes = [{'size': disk.size} for disk in instance.disks]
3643
          val = _ComputeDiskSize(instance.disk_template, disk_sizes)
3644
        elif field == "tags":
3645
          val = list(instance.GetTags())
3646
        elif field == "serial_no":
3647
          val = instance.serial_no
3648
        elif field == "network_port":
3649
          val = instance.network_port
3650
        elif field == "hypervisor":
3651
          val = instance.hypervisor
3652
        elif field == "hvparams":
3653
          val = i_hv
3654
        elif (field.startswith(HVPREFIX) and
3655
              field[len(HVPREFIX):] in constants.HVS_PARAMETERS):
3656
          val = i_hv.get(field[len(HVPREFIX):], None)
3657
        elif field == "beparams":
3658
          val = i_be
3659
        elif (field.startswith(BEPREFIX) and
3660
              field[len(BEPREFIX):] in constants.BES_PARAMETERS):
3661
          val = i_be.get(field[len(BEPREFIX):], None)
3662
        elif st_match and st_match.groups():
3663
          # matches a variable list
3664
          st_groups = st_match.groups()
3665
          if st_groups and st_groups[0] == "disk":
3666
            if st_groups[1] == "count":
3667
              val = len(instance.disks)
3668
            elif st_groups[1] == "sizes":
3669
              val = [disk.size for disk in instance.disks]
3670
            elif st_groups[1] == "size":
3671
              try:
3672
                val = instance.FindDisk(st_groups[2]).size
3673
              except errors.OpPrereqError:
3674
                val = None
3675
            else:
3676
              assert False, "Unhandled disk parameter"
3677
          elif st_groups[0] == "nic":
3678
            if st_groups[1] == "count":
3679
              val = len(instance.nics)
3680
            elif st_groups[1] == "macs":
3681
              val = [nic.mac for nic in instance.nics]
3682
            elif st_groups[1] == "ips":
3683
              val = [nic.ip for nic in instance.nics]
3684
            elif st_groups[1] == "bridges":
3685
              val = [nic.bridge for nic in instance.nics]
3686
            else:
3687
              # index-based item
3688
              nic_idx = int(st_groups[2])
3689
              if nic_idx >= len(instance.nics):
3690
                val = None
3691
              else:
3692
                if st_groups[1] == "mac":
3693
                  val = instance.nics[nic_idx].mac
3694
                elif st_groups[1] == "ip":
3695
                  val = instance.nics[nic_idx].ip
3696
                elif st_groups[1] == "bridge":
3697
                  val = instance.nics[nic_idx].bridge
3698
                else:
3699
                  assert False, "Unhandled NIC parameter"
3700
          else:
3701
            assert False, ("Declared but unhandled variable parameter '%s'" %
3702
                           field)
3703
        else:
3704
          assert False, "Declared but unhandled parameter '%s'" % field
3705
        iout.append(val)
3706
      output.append(iout)
3707

    
3708
    return output
3709

    
3710

    
3711
class LUFailoverInstance(LogicalUnit):
3712
  """Failover an instance.
3713

3714
  """
3715
  HPATH = "instance-failover"
3716
  HTYPE = constants.HTYPE_INSTANCE
3717
  _OP_REQP = ["instance_name", "ignore_consistency"]
3718
  REQ_BGL = False
3719

    
3720
  def ExpandNames(self):
3721
    self._ExpandAndLockInstance()
3722
    self.needed_locks[locking.LEVEL_NODE] = []
3723
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3724

    
3725
  def DeclareLocks(self, level):
3726
    if level == locking.LEVEL_NODE:
3727
      self._LockInstancesNodes()
3728

    
3729
  def BuildHooksEnv(self):
3730
    """Build hooks env.
3731

3732
    This runs on master, primary and secondary nodes of the instance.
3733

3734
    """
3735
    env = {
3736
      "IGNORE_CONSISTENCY": self.op.ignore_consistency,
3737
      }
3738
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3739
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3740
    return env, nl, nl
3741

    
3742
  def CheckPrereq(self):
3743
    """Check prerequisites.
3744

3745
    This checks that the instance is in the cluster.
3746

3747
    """
3748
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3749
    assert self.instance is not None, \
3750
      "Cannot retrieve locked instance %s" % self.op.instance_name
3751

    
3752
    bep = self.cfg.GetClusterInfo().FillBE(instance)
3753
    if instance.disk_template not in constants.DTS_NET_MIRROR:
3754
      raise errors.OpPrereqError("Instance's disk layout is not"
3755
                                 " network mirrored, cannot failover.")
3756

    
3757
    secondary_nodes = instance.secondary_nodes
3758
    if not secondary_nodes:
3759
      raise errors.ProgrammerError("no secondary node but using "
3760
                                   "a mirrored disk template")
3761

    
3762
    target_node = secondary_nodes[0]
3763
    _CheckNodeOnline(self, target_node)
3764
    _CheckNodeNotDrained(self, target_node)
3765

    
3766
    if instance.admin_up:
3767
      # check memory requirements on the secondary node
3768
      _CheckNodeFreeMemory(self, target_node, "failing over instance %s" %
3769
                           instance.name, bep[constants.BE_MEMORY],
3770
                           instance.hypervisor)
3771
    else:
3772
      self.LogInfo("Not checking memory on the secondary node as"
3773
                   " instance will not be started")
3774

    
3775
    # check bridge existence
3776
    brlist = [nic.bridge for nic in instance.nics]
3777
    result = self.rpc.call_bridges_exist(target_node, brlist)
3778
    result.Raise()
3779
    if not result.data:
3780
      raise errors.OpPrereqError("One or more target bridges %s does not"
3781
                                 " exist on destination node '%s'" %
3782
                                 (brlist, target_node))
3783

    
3784
  def Exec(self, feedback_fn):
3785
    """Failover an instance.
3786

3787
    The failover is done by shutting it down on its present node and
3788
    starting it on the secondary.
3789

3790
    """
3791
    instance = self.instance
3792

    
3793
    source_node = instance.primary_node
3794
    target_node = instance.secondary_nodes[0]
3795

    
3796
    feedback_fn("* checking disk consistency between source and target")
3797
    for dev in instance.disks:
3798
      # for drbd, these are drbd over lvm
3799
      if not _CheckDiskConsistency(self, dev, target_node, False):
3800
        if instance.admin_up and not self.op.ignore_consistency:
3801
          raise errors.OpExecError("Disk %s is degraded on target node,"
3802
                                   " aborting failover." % dev.iv_name)
3803

    
3804
    feedback_fn("* shutting down instance on source node")
3805
    logging.info("Shutting down instance %s on node %s",
3806
                 instance.name, source_node)
3807

    
3808
    result = self.rpc.call_instance_shutdown(source_node, instance)
3809
    msg = result.RemoteFailMsg()
3810
    if msg:
3811
      if self.op.ignore_consistency:
3812
        self.proc.LogWarning("Could not shutdown instance %s on node %s."
3813
                             " Proceeding anyway. Please make sure node"
3814
                             " %s is down. Error details: %s",
3815
                             instance.name, source_node, source_node, msg)
3816
      else:
3817
        raise errors.OpExecError("Could not shutdown instance %s on"
3818
                                 " node %s: %s" %
3819
                                 (instance.name, source_node, msg))
3820

    
3821
    feedback_fn("* deactivating the instance's disks on source node")
3822
    if not _ShutdownInstanceDisks(self, instance, ignore_primary=True):
3823
      raise errors.OpExecError("Can't shut down the instance's disks.")
3824

    
3825
    instance.primary_node = target_node
3826
    # distribute new instance config to the other nodes
3827
    self.cfg.Update(instance)
3828

    
3829
    # Only start the instance if it's marked as up
3830
    if instance.admin_up:
3831
      feedback_fn("* activating the instance's disks on target node")
3832
      logging.info("Starting instance %s on node %s",
3833
                   instance.name, target_node)
3834

    
3835
      disks_ok, _ = _AssembleInstanceDisks(self, instance,
3836
                                               ignore_secondaries=True)
3837
      if not disks_ok:
3838
        _ShutdownInstanceDisks(self, instance)
3839
        raise errors.OpExecError("Can't activate the instance's disks")
3840

    
3841
      feedback_fn("* starting the instance on the target node")
3842
      result = self.rpc.call_instance_start(target_node, instance, None, None)
3843
      msg = result.RemoteFailMsg()
3844
      if msg:
3845
        _ShutdownInstanceDisks(self, instance)
3846
        raise errors.OpExecError("Could not start instance %s on node %s: %s" %
3847
                                 (instance.name, target_node, msg))
3848

    
3849

    
3850
class LUMigrateInstance(LogicalUnit):
3851
  """Migrate an instance.
3852

3853
  This is migration without shutting down, compared to the failover,
3854
  which is done with shutdown.
3855

3856
  """
3857
  HPATH = "instance-migrate"
3858
  HTYPE = constants.HTYPE_INSTANCE
3859
  _OP_REQP = ["instance_name", "live", "cleanup"]
3860

    
3861
  REQ_BGL = False
3862

    
3863
  def ExpandNames(self):
3864
    self._ExpandAndLockInstance()
3865
    self.needed_locks[locking.LEVEL_NODE] = []
3866
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3867

    
3868
  def DeclareLocks(self, level):
3869
    if level == locking.LEVEL_NODE:
3870
      self._LockInstancesNodes()
3871

    
3872
  def BuildHooksEnv(self):
3873
    """Build hooks env.
3874

3875
    This runs on master, primary and secondary nodes of the instance.
3876

3877
    """
3878
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3879
    env["MIGRATE_LIVE"] = self.op.live
3880
    env["MIGRATE_CLEANUP"] = self.op.cleanup
3881
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3882
    return env, nl, nl
3883

    
3884
  def CheckPrereq(self):
3885
    """Check prerequisites.
3886

3887
    This checks that the instance is in the cluster.
3888

3889
    """
3890
    instance = self.cfg.GetInstanceInfo(
3891
      self.cfg.ExpandInstanceName(self.op.instance_name))
3892
    if instance is None:
3893
      raise errors.OpPrereqError("Instance '%s' not known" %
3894
                                 self.op.instance_name)
3895

    
3896
    if instance.disk_template != constants.DT_DRBD8:
3897
      raise errors.OpPrereqError("Instance's disk layout is not"
3898
                                 " drbd8, cannot migrate.")
3899

    
3900
    secondary_nodes = instance.secondary_nodes
3901
    if not secondary_nodes:
3902
      raise errors.ConfigurationError("No secondary node but using"
3903
                                      " drbd8 disk template")
3904

    
3905
    i_be = self.cfg.GetClusterInfo().FillBE(instance)
3906

    
3907
    target_node = secondary_nodes[0]
3908
    # check memory requirements on the secondary node
3909
    _CheckNodeFreeMemory(self, target_node, "migrating instance %s" %
3910
                         instance.name, i_be[constants.BE_MEMORY],
3911
                         instance.hypervisor)
3912

    
3913
    # check bridge existence
3914
    brlist = [nic.bridge for nic in instance.nics]
3915
    result = self.rpc.call_bridges_exist(target_node, brlist)
3916
    if result.failed or not result.data:
3917
      raise errors.OpPrereqError("One or more target bridges %s does not"
3918
                                 " exist on destination node '%s'" %
3919
                                 (brlist, target_node))
3920

    
3921
    if not self.op.cleanup:
3922
      _CheckNodeNotDrained(self, target_node)
3923
      result = self.rpc.call_instance_migratable(instance.primary_node,
3924
                                                 instance)
3925
      msg = result.RemoteFailMsg()
3926
      if msg:
3927
        raise errors.OpPrereqError("Can't migrate: %s - please use failover" %
3928
                                   msg)
3929

    
3930
    self.instance = instance
3931

    
3932
  def _WaitUntilSync(self):
3933
    """Poll with custom rpc for disk sync.
3934

3935
    This uses our own step-based rpc call.
3936

3937
    """
3938
    self.feedback_fn("* wait until resync is done")
3939
    all_done = False
3940
    while not all_done:
3941
      all_done = True
3942
      result = self.rpc.call_drbd_wait_sync(self.all_nodes,
3943
                                            self.nodes_ip,
3944
                                            self.instance.disks)
3945
      min_percent = 100
3946
      for node, nres in result.items():
3947
        msg = nres.RemoteFailMsg()
3948
        if msg:
3949
          raise errors.OpExecError("Cannot resync disks on node %s: %s" %
3950
                                   (node, msg))
3951
        node_done, node_percent = nres.payload
3952
        all_done = all_done and node_done
3953
        if node_percent is not None:
3954
          min_percent = min(min_percent, node_percent)
3955
      if not all_done:
3956
        if min_percent < 100:
3957
          self.feedback_fn("   - progress: %.1f%%" % min_percent)
3958
        time.sleep(2)
3959

    
3960
  def _EnsureSecondary(self, node):
3961
    """Demote a node to secondary.
3962

3963
    """
3964
    self.feedback_fn("* switching node %s to secondary mode" % node)
3965

    
3966
    for dev in self.instance.disks:
3967
      self.cfg.SetDiskID(dev, node)
3968

    
3969
    result = self.rpc.call_blockdev_close(node, self.instance.name,
3970
                                          self.instance.disks)
3971
    msg = result.RemoteFailMsg()
3972
    if msg:
3973
      raise errors.OpExecError("Cannot change disk to secondary on node %s,"
3974
                               " error %s" % (node, msg))
3975

    
3976
  def _GoStandalone(self):
3977
    """Disconnect from the network.
3978

3979
    """
3980
    self.feedback_fn("* changing into standalone mode")
3981
    result = self.rpc.call_drbd_disconnect_net(self.all_nodes, self.nodes_ip,
3982
                                               self.instance.disks)
3983
    for node, nres in result.items():
3984
      msg = nres.RemoteFailMsg()
3985
      if msg:
3986
        raise errors.OpExecError("Cannot disconnect disks node %s,"
3987
                                 " error %s" % (node, msg))
3988

    
3989
  def _GoReconnect(self, multimaster):
3990
    """Reconnect to the network.
3991

3992
    """
3993
    if multimaster:
3994
      msg = "dual-master"
3995
    else:
3996
      msg = "single-master"
3997
    self.feedback_fn("* changing disks into %s mode" % msg)
3998
    result = self.rpc.call_drbd_attach_net(self.all_nodes, self.nodes_ip,
3999
                                           self.instance.disks,
4000
                                           self.instance.name, multimaster)
4001
    for node, nres in result.items():
4002
      msg = nres.RemoteFailMsg()
4003