Statistics
| Branch: | Tag: | Revision:

root / lib / cmdlib.py @ c70d2d9b

History | View | Annotate | Download (253 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

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

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

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

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

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

    
365
  return utils.NiceSort(wanted)
366

    
367

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

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

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

    
384
  if instances:
385
    wanted = []
386

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

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

    
397

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

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

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

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

    
416

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

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

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

    
430

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

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

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

    
442

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

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

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

    
454

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

460
  This builds the hook environment from individual variables.
461

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

492
  """
493
  if status:
494
    str_status = "up"
495
  else:
496
    str_status = "down"
497
  env = {
498
    "OP_TARGET": name,
499
    "INSTANCE_NAME": name,
500
    "INSTANCE_PRIMARY": primary_node,
501
    "INSTANCE_SECONDARIES": " ".join(secondary_nodes),
502
    "INSTANCE_OS_TYPE": os_type,
503
    "INSTANCE_STATUS": str_status,
504
    "INSTANCE_MEMORY": memory,
505
    "INSTANCE_VCPUS": vcpus,
506
    "INSTANCE_DISK_TEMPLATE": disk_template,
507
    "INSTANCE_HYPERVISOR": hypervisor_name,
508
  }
509

    
510
  if nics:
511
    nic_count = len(nics)
512
    for idx, (ip, bridge, mac) in enumerate(nics):
513
      if ip is None:
514
        ip = ""
515
      env["INSTANCE_NIC%d_IP" % idx] = ip
516
      env["INSTANCE_NIC%d_BRIDGE" % idx] = bridge
517
      env["INSTANCE_NIC%d_MAC" % idx] = mac
518
  else:
519
    nic_count = 0
520

    
521
  env["INSTANCE_NIC_COUNT"] = nic_count
522

    
523
  if disks:
524
    disk_count = len(disks)
525
    for idx, (size, mode) in enumerate(disks):
526
      env["INSTANCE_DISK%d_SIZE" % idx] = size
527
      env["INSTANCE_DISK%d_MODE" % idx] = mode
528
  else:
529
    disk_count = 0
530

    
531
  env["INSTANCE_DISK_COUNT"] = disk_count
532

    
533
  for source, kind in [(bep, "BE"), (hvp, "HV")]:
534
    for key, value in source.items():
535
      env["INSTANCE_%s_%s" % (kind, key)] = value
536

    
537
  return env
538

    
539

    
540
def _BuildInstanceHookEnvByObject(lu, instance, override=None):
541
  """Builds instance related env variables for hooks from an object.
542

543
  @type lu: L{LogicalUnit}
544
  @param lu: the logical unit on whose behalf we execute
545
  @type instance: L{objects.Instance}
546
  @param instance: the instance for which we should build the
547
      environment
548
  @type override: dict
549
  @param override: dictionary with key/values that will override
550
      our values
551
  @rtype: dict
552
  @return: the hook environment dictionary
553

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

    
577

    
578
def _AdjustCandidatePool(lu):
579
  """Adjust the candidate pool after node operations.
580

581
  """
582
  mod_list = lu.cfg.MaintainCandidatePool()
583
  if mod_list:
584
    lu.LogInfo("Promoted nodes to master candidate role: %s",
585
               ", ".join(node.name for node in mod_list))
586
    for name in mod_list:
587
      lu.context.ReaddNode(name)
588
  mc_now, mc_max = lu.cfg.GetMasterCandidateStats()
589
  if mc_now > mc_max:
590
    lu.LogInfo("Note: more nodes are candidates (%d) than desired (%d)" %
591
               (mc_now, mc_max))
592

    
593

    
594
def _CheckInstanceBridgesExist(lu, instance):
595
  """Check that the bridges needed by an instance exist.
596

597
  """
598
  # check bridges existence
599
  brlist = [nic.bridge for nic in instance.nics]
600
  result = lu.rpc.call_bridges_exist(instance.primary_node, brlist)
601
  result.Raise()
602
  if not result.data:
603
    raise errors.OpPrereqError("One or more target bridges %s does not"
604
                               " exist on destination node '%s'" %
605
                               (brlist, instance.primary_node))
606

    
607

    
608
class LUDestroyCluster(NoHooksLU):
609
  """Logical unit for destroying the cluster.
610

611
  """
612
  _OP_REQP = []
613

    
614
  def CheckPrereq(self):
615
    """Check prerequisites.
616

617
    This checks whether the cluster is empty.
618

619
    Any errors are signaled by raising errors.OpPrereqError.
620

621
    """
622
    master = self.cfg.GetMasterNode()
623

    
624
    nodelist = self.cfg.GetNodeList()
625
    if len(nodelist) != 1 or nodelist[0] != master:
626
      raise errors.OpPrereqError("There are still %d node(s) in"
627
                                 " this cluster." % (len(nodelist) - 1))
628
    instancelist = self.cfg.GetInstanceList()
629
    if instancelist:
630
      raise errors.OpPrereqError("There are still %d instance(s) in"
631
                                 " this cluster." % len(instancelist))
632

    
633
  def Exec(self, feedback_fn):
634
    """Destroys the cluster.
635

636
    """
637
    master = self.cfg.GetMasterNode()
638
    result = self.rpc.call_node_stop_master(master, False)
639
    result.Raise()
640
    if not result.data:
641
      raise errors.OpExecError("Could not disable the master role")
642
    priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
643
    utils.CreateBackup(priv_key)
644
    utils.CreateBackup(pub_key)
645
    return master
646

    
647

    
648
class LUVerifyCluster(LogicalUnit):
649
  """Verifies the cluster status.
650

651
  """
652
  HPATH = "cluster-verify"
653
  HTYPE = constants.HTYPE_CLUSTER
654
  _OP_REQP = ["skip_checks"]
655
  REQ_BGL = False
656

    
657
  def ExpandNames(self):
658
    self.needed_locks = {
659
      locking.LEVEL_NODE: locking.ALL_SET,
660
      locking.LEVEL_INSTANCE: locking.ALL_SET,
661
    }
662
    self.share_locks = dict(((i, 1) for i in locking.LEVELS))
663

    
664
  def _VerifyNode(self, nodeinfo, file_list, local_cksum,
665
                  node_result, feedback_fn, master_files,
666
                  drbd_map, vg_name):
667
    """Run multiple tests against a node.
668

669
    Test list:
670

671
      - compares ganeti version
672
      - checks vg existence and size > 20G
673
      - checks config file checksum
674
      - checks ssh to other nodes
675

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

688
    """
689
    node = nodeinfo.name
690

    
691
    # main result, node_result should be a non-empty dict
692
    if not node_result or not isinstance(node_result, dict):
693
      feedback_fn("  - ERROR: unable to verify node %s." % (node,))
694
      return True
695

    
696
    # compares ganeti version
697
    local_version = constants.PROTOCOL_VERSION
698
    remote_version = node_result.get('version', None)
699
    if not (remote_version and isinstance(remote_version, (list, tuple)) and
700
            len(remote_version) == 2):
701
      feedback_fn("  - ERROR: connection to %s failed" % (node))
702
      return True
703

    
704
    if local_version != remote_version[0]:
705
      feedback_fn("  - ERROR: incompatible protocol versions: master %s,"
706
                  " node %s %s" % (local_version, node, remote_version[0]))
707
      return True
708

    
709
    # node seems compatible, we can actually try to look into its results
710

    
711
    bad = False
712

    
713
    # full package version
714
    if constants.RELEASE_VERSION != remote_version[1]:
715
      feedback_fn("  - WARNING: software version mismatch: master %s,"
716
                  " node %s %s" %
717
                  (constants.RELEASE_VERSION, node, remote_version[1]))
718

    
719
    # checks vg existence and size > 20G
720
    if vg_name is not None:
721
      vglist = node_result.get(constants.NV_VGLIST, None)
722
      if not vglist:
723
        feedback_fn("  - ERROR: unable to check volume groups on node %s." %
724
                        (node,))
725
        bad = True
726
      else:
727
        vgstatus = utils.CheckVolumeGroupSize(vglist, vg_name,
728
                                              constants.MIN_VG_SIZE)
729
        if vgstatus:
730
          feedback_fn("  - ERROR: %s on node %s" % (vgstatus, node))
731
          bad = True
732

    
733
    # checks config file checksum
734

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

    
762
    # checks ssh to any
763

    
764
    if constants.NV_NODELIST not in node_result:
765
      bad = True
766
      feedback_fn("  - ERROR: node hasn't returned node ssh connectivity data")
767
    else:
768
      if node_result[constants.NV_NODELIST]:
769
        bad = True
770
        for node in node_result[constants.NV_NODELIST]:
771
          feedback_fn("  - ERROR: ssh communication with node '%s': %s" %
772
                          (node, node_result[constants.NV_NODELIST][node]))
773

    
774
    if constants.NV_NODENETTEST not in node_result:
775
      bad = True
776
      feedback_fn("  - ERROR: node hasn't returned node tcp connectivity data")
777
    else:
778
      if node_result[constants.NV_NODENETTEST]:
779
        bad = True
780
        nlist = utils.NiceSort(node_result[constants.NV_NODENETTEST].keys())
781
        for node in nlist:
782
          feedback_fn("  - ERROR: tcp communication with node '%s': %s" %
783
                          (node, node_result[constants.NV_NODENETTEST][node]))
784

    
785
    hyp_result = node_result.get(constants.NV_HYPERVISOR, None)
786
    if isinstance(hyp_result, dict):
787
      for hv_name, hv_result in hyp_result.iteritems():
788
        if hv_result is not None:
789
          feedback_fn("  - ERROR: hypervisor %s verify failure: '%s'" %
790
                      (hv_name, hv_result))
791

    
792
    # check used drbd list
793
    if vg_name is not None:
794
      used_minors = node_result.get(constants.NV_DRBDLIST, [])
795
      if not isinstance(used_minors, (tuple, list)):
796
        feedback_fn("  - ERROR: cannot parse drbd status file: %s" %
797
                    str(used_minors))
798
      else:
799
        for minor, (iname, must_exist) in drbd_map.items():
800
          if minor not in used_minors and must_exist:
801
            feedback_fn("  - ERROR: drbd minor %d of instance %s is"
802
                        " not active" % (minor, iname))
803
            bad = True
804
        for minor in used_minors:
805
          if minor not in drbd_map:
806
            feedback_fn("  - ERROR: unallocated drbd minor %d is in use" %
807
                        minor)
808
            bad = True
809

    
810
    return bad
811

    
812
  def _VerifyInstance(self, instance, instanceconfig, node_vol_is,
813
                      node_instance, feedback_fn, n_offline):
814
    """Verify an instance.
815

816
    This function checks to see if the required block devices are
817
    available on the instance's node.
818

819
    """
820
    bad = False
821

    
822
    node_current = instanceconfig.primary_node
823

    
824
    node_vol_should = {}
825
    instanceconfig.MapLVsByNode(node_vol_should)
826

    
827
    for node in node_vol_should:
828
      if node in n_offline:
829
        # ignore missing volumes on offline nodes
830
        continue
831
      for volume in node_vol_should[node]:
832
        if node not in node_vol_is or volume not in node_vol_is[node]:
833
          feedback_fn("  - ERROR: volume %s missing on node %s" %
834
                          (volume, node))
835
          bad = True
836

    
837
    if instanceconfig.admin_up:
838
      if ((node_current not in node_instance or
839
          not instance in node_instance[node_current]) and
840
          node_current not in n_offline):
841
        feedback_fn("  - ERROR: instance %s not running on node %s" %
842
                        (instance, node_current))
843
        bad = True
844

    
845
    for node in node_instance:
846
      if (not node == node_current):
847
        if instance in node_instance[node]:
848
          feedback_fn("  - ERROR: instance %s should not run on node %s" %
849
                          (instance, node))
850
          bad = True
851

    
852
    return bad
853

    
854
  def _VerifyOrphanVolumes(self, node_vol_should, node_vol_is, feedback_fn):
855
    """Verify if there are any unknown volumes in the cluster.
856

857
    The .os, .swap and backup volumes are ignored. All other volumes are
858
    reported as unknown.
859

860
    """
861
    bad = False
862

    
863
    for node in node_vol_is:
864
      for volume in node_vol_is[node]:
865
        if node not in node_vol_should or volume not in node_vol_should[node]:
866
          feedback_fn("  - ERROR: volume %s on node %s should not exist" %
867
                      (volume, node))
868
          bad = True
869
    return bad
870

    
871
  def _VerifyOrphanInstances(self, instancelist, node_instance, feedback_fn):
872
    """Verify the list of running instances.
873

874
    This checks what instances are running but unknown to the cluster.
875

876
    """
877
    bad = False
878
    for node in node_instance:
879
      for runninginstance in node_instance[node]:
880
        if runninginstance not in instancelist:
881
          feedback_fn("  - ERROR: instance %s on node %s should not exist" %
882
                          (runninginstance, node))
883
          bad = True
884
    return bad
885

    
886
  def _VerifyNPlusOneMemory(self, node_info, instance_cfg, feedback_fn):
887
    """Verify N+1 Memory Resilience.
888

889
    Check that if one single node dies we can still start all the instances it
890
    was primary for.
891

892
    """
893
    bad = False
894

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

    
916
  def CheckPrereq(self):
917
    """Check prerequisites.
918

919
    Transform the list of checks we're going to skip into a set and check that
920
    all its members are valid.
921

922
    """
923
    self.skip_set = frozenset(self.op.skip_checks)
924
    if not constants.VERIFY_OPTIONAL_CHECKS.issuperset(self.skip_set):
925
      raise errors.OpPrereqError("Invalid checks to be skipped specified")
926

    
927
  def BuildHooksEnv(self):
928
    """Build hooks env.
929

930
    Cluster-Verify hooks just ran in the post phase and their failure makes
931
    the output be logged in the verify output and the verification to fail.
932

933
    """
934
    all_nodes = self.cfg.GetNodeList()
935
    env = {
936
      "CLUSTER_TAGS": " ".join(self.cfg.GetClusterInfo().GetTags())
937
      }
938
    for node in self.cfg.GetAllNodesInfo().values():
939
      env["NODE_TAGS_%s" % node.name] = " ".join(node.GetTags())
940

    
941
    return env, [], all_nodes
942

    
943
  def Exec(self, feedback_fn):
944
    """Verify integrity of cluster, performing various test on nodes.
945

946
    """
947
    bad = False
948
    feedback_fn("* Verifying global settings")
949
    for msg in self.cfg.VerifyConfig():
950
      feedback_fn("  - ERROR: %s" % msg)
951

    
952
    vg_name = self.cfg.GetVGName()
953
    hypervisors = self.cfg.GetClusterInfo().enabled_hypervisors
954
    nodelist = utils.NiceSort(self.cfg.GetNodeList())
955
    nodeinfo = [self.cfg.GetNodeInfo(nname) for nname in nodelist]
956
    instancelist = utils.NiceSort(self.cfg.GetInstanceList())
957
    instanceinfo = dict((iname, self.cfg.GetInstanceInfo(iname))
958
                        for iname in instancelist)
959
    i_non_redundant = [] # Non redundant instances
960
    i_non_a_balanced = [] # Non auto-balanced instances
961
    n_offline = [] # List of offline nodes
962
    n_drained = [] # List of nodes being drained
963
    node_volume = {}
964
    node_instance = {}
965
    node_info = {}
966
    instance_cfg = {}
967

    
968
    # FIXME: verify OS list
969
    # do local checksums
970
    master_files = [constants.CLUSTER_CONF_FILE]
971

    
972
    file_names = ssconf.SimpleStore().GetFileList()
973
    file_names.append(constants.SSL_CERT_FILE)
974
    file_names.append(constants.RAPI_CERT_FILE)
975
    file_names.extend(master_files)
976

    
977
    local_checksums = utils.FingerprintFiles(file_names)
978

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

    
999
    cluster = self.cfg.GetClusterInfo()
1000
    master_node = self.cfg.GetMasterNode()
1001
    all_drbd_map = self.cfg.ComputeDRBDMap()
1002

    
1003
    for node_i in nodeinfo:
1004
      node = node_i.name
1005
      nresult = all_nvinfo[node].data
1006

    
1007
      if node_i.offline:
1008
        feedback_fn("* Skipping offline node %s" % (node,))
1009
        n_offline.append(node)
1010
        continue
1011

    
1012
      if node == master_node:
1013
        ntype = "master"
1014
      elif node_i.master_candidate:
1015
        ntype = "master candidate"
1016
      elif node_i.drained:
1017
        ntype = "drained"
1018
        n_drained.append(node)
1019
      else:
1020
        ntype = "regular"
1021
      feedback_fn("* Verifying node %s (%s)" % (node, ntype))
1022

    
1023
      if all_nvinfo[node].failed or not isinstance(nresult, dict):
1024
        feedback_fn("  - ERROR: connection to %s failed" % (node,))
1025
        bad = True
1026
        continue
1027

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

    
1045
      lvdata = nresult.get(constants.NV_LVLIST, "Missing LV data")
1046
      if vg_name is None:
1047
        node_volume[node] = {}
1048
      elif isinstance(lvdata, basestring):
1049
        feedback_fn("  - ERROR: LVM problem on node %s: %s" %
1050
                    (node, utils.SafeEncode(lvdata)))
1051
        bad = True
1052
        node_volume[node] = {}
1053
      elif not isinstance(lvdata, dict):
1054
        feedback_fn("  - ERROR: connection to %s failed (lvlist)" % (node,))
1055
        bad = True
1056
        continue
1057
      else:
1058
        node_volume[node] = lvdata
1059

    
1060
      # node_instance
1061
      idata = nresult.get(constants.NV_INSTANCELIST, None)
1062
      if not isinstance(idata, list):
1063
        feedback_fn("  - ERROR: connection to %s failed (instancelist)" %
1064
                    (node,))
1065
        bad = True
1066
        continue
1067

    
1068
      node_instance[node] = idata
1069

    
1070
      # node_info
1071
      nodeinfo = nresult.get(constants.NV_HVINFO, None)
1072
      if not isinstance(nodeinfo, dict):
1073
        feedback_fn("  - ERROR: connection to %s failed (hvinfo)" % (node,))
1074
        bad = True
1075
        continue
1076

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

    
1106
    node_vol_should = {}
1107

    
1108
    for instance in instancelist:
1109
      feedback_fn("* Verifying instance %s" % instance)
1110
      inst_config = instanceinfo[instance]
1111
      result =  self._VerifyInstance(instance, inst_config, node_volume,
1112
                                     node_instance, feedback_fn, n_offline)
1113
      bad = bad or result
1114
      inst_nodes_offline = []
1115

    
1116
      inst_config.MapLVsByNode(node_vol_should)
1117

    
1118
      instance_cfg[instance] = inst_config
1119

    
1120
      pnode = inst_config.primary_node
1121
      if pnode in node_info:
1122
        node_info[pnode]['pinst'].append(instance)
1123
      elif pnode not in n_offline:
1124
        feedback_fn("  - ERROR: instance %s, connection to primary node"
1125
                    " %s failed" % (instance, pnode))
1126
        bad = True
1127

    
1128
      if pnode in n_offline:
1129
        inst_nodes_offline.append(pnode)
1130

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

    
1142
      if not cluster.FillBE(inst_config)[constants.BE_AUTO_BALANCE]:
1143
        i_non_a_balanced.append(instance)
1144

    
1145
      for snode in inst_config.secondary_nodes:
1146
        if snode in node_info:
1147
          node_info[snode]['sinst'].append(instance)
1148
          if pnode not in node_info[snode]['sinst-by-pnode']:
1149
            node_info[snode]['sinst-by-pnode'][pnode] = []
1150
          node_info[snode]['sinst-by-pnode'][pnode].append(instance)
1151
        elif snode not in n_offline:
1152
          feedback_fn("  - ERROR: instance %s, connection to secondary node"
1153
                      " %s failed" % (instance, snode))
1154
          bad = True
1155
        if snode in n_offline:
1156
          inst_nodes_offline.append(snode)
1157

    
1158
      if inst_nodes_offline:
1159
        # warn that the instance lives on offline nodes, and set bad=True
1160
        feedback_fn("  - ERROR: instance lives on offline node(s) %s" %
1161
                    ", ".join(inst_nodes_offline))
1162
        bad = True
1163

    
1164
    feedback_fn("* Verifying orphan volumes")
1165
    result = self._VerifyOrphanVolumes(node_vol_should, node_volume,
1166
                                       feedback_fn)
1167
    bad = bad or result
1168

    
1169
    feedback_fn("* Verifying remaining instances")
1170
    result = self._VerifyOrphanInstances(instancelist, node_instance,
1171
                                         feedback_fn)
1172
    bad = bad or result
1173

    
1174
    if constants.VERIFY_NPLUSONE_MEM not in self.skip_set:
1175
      feedback_fn("* Verifying N+1 Memory redundancy")
1176
      result = self._VerifyNPlusOneMemory(node_info, instance_cfg, feedback_fn)
1177
      bad = bad or result
1178

    
1179
    feedback_fn("* Other Notes")
1180
    if i_non_redundant:
1181
      feedback_fn("  - NOTICE: %d non-redundant instance(s) found."
1182
                  % len(i_non_redundant))
1183

    
1184
    if i_non_a_balanced:
1185
      feedback_fn("  - NOTICE: %d non-auto-balanced instance(s) found."
1186
                  % len(i_non_a_balanced))
1187

    
1188
    if n_offline:
1189
      feedback_fn("  - NOTICE: %d offline node(s) found." % len(n_offline))
1190

    
1191
    if n_drained:
1192
      feedback_fn("  - NOTICE: %d drained node(s) found." % len(n_drained))
1193

    
1194
    return not bad
1195

    
1196
  def HooksCallBack(self, phase, hooks_results, feedback_fn, lu_result):
1197
    """Analyze the post-hooks' result
1198

1199
    This method analyses the hook result, handles it, and sends some
1200
    nicely-formatted feedback back to the user.
1201

1202
    @param phase: one of L{constants.HOOKS_PHASE_POST} or
1203
        L{constants.HOOKS_PHASE_PRE}; it denotes the hooks phase
1204
    @param hooks_results: the results of the multi-node hooks rpc call
1205
    @param feedback_fn: function used send feedback back to the caller
1206
    @param lu_result: previous Exec result
1207
    @return: the new Exec result, based on the previous result
1208
        and hook results
1209

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

    
1243
      return lu_result
1244

    
1245

    
1246
class LUVerifyDisks(NoHooksLU):
1247
  """Verifies the cluster disks status.
1248

1249
  """
1250
  _OP_REQP = []
1251
  REQ_BGL = False
1252

    
1253
  def ExpandNames(self):
1254
    self.needed_locks = {
1255
      locking.LEVEL_NODE: locking.ALL_SET,
1256
      locking.LEVEL_INSTANCE: locking.ALL_SET,
1257
    }
1258
    self.share_locks = dict(((i, 1) for i in locking.LEVELS))
1259

    
1260
  def CheckPrereq(self):
1261
    """Check prerequisites.
1262

1263
    This has no prerequisites.
1264

1265
    """
1266
    pass
1267

    
1268
  def Exec(self, feedback_fn):
1269
    """Verify integrity of cluster disks.
1270

1271
    """
1272
    result = res_nodes, res_nlvm, res_instances, res_missing = [], {}, [], {}
1273

    
1274
    vg_name = self.cfg.GetVGName()
1275
    nodes = utils.NiceSort(self.cfg.GetNodeList())
1276
    instances = [self.cfg.GetInstanceInfo(name)
1277
                 for name in self.cfg.GetInstanceList()]
1278

    
1279
    nv_dict = {}
1280
    for inst in instances:
1281
      inst_lvs = {}
1282
      if (not inst.admin_up or
1283
          inst.disk_template not in constants.DTS_NET_MIRROR):
1284
        continue
1285
      inst.MapLVsByNode(inst_lvs)
1286
      # transform { iname: {node: [vol,],},} to {(node, vol): iname}
1287
      for node, vol_list in inst_lvs.iteritems():
1288
        for vol in vol_list:
1289
          nv_dict[(node, vol)] = inst
1290

    
1291
    if not nv_dict:
1292
      return result
1293

    
1294
    node_lvs = self.rpc.call_volume_list(nodes, vg_name)
1295

    
1296
    for node in nodes:
1297
      # node_volume
1298
      lvs = node_lvs[node]
1299
      if lvs.failed:
1300
        if not lvs.offline:
1301
          self.LogWarning("Connection to node %s failed: %s" %
1302
                          (node, lvs.data))
1303
        continue
1304
      lvs = lvs.data
1305
      if isinstance(lvs, basestring):
1306
        logging.warning("Error enumerating LVs on node %s: %s", node, lvs)
1307
        res_nlvm[node] = lvs
1308
        continue
1309
      elif not isinstance(lvs, dict):
1310
        logging.warning("Connection to node %s failed or invalid data"
1311
                        " returned", node)
1312
        res_nodes.append(node)
1313
        continue
1314

    
1315
      for lv_name, (_, lv_inactive, lv_online) in lvs.iteritems():
1316
        inst = nv_dict.pop((node, lv_name), None)
1317
        if (not lv_online and inst is not None
1318
            and inst.name not in res_instances):
1319
          res_instances.append(inst.name)
1320

    
1321
    # any leftover items in nv_dict are missing LVs, let's arrange the
1322
    # data better
1323
    for key, inst in nv_dict.iteritems():
1324
      if inst.name not in res_missing:
1325
        res_missing[inst.name] = []
1326
      res_missing[inst.name].append(key)
1327

    
1328
    return result
1329

    
1330

    
1331
class LURepairDiskSizes(NoHooksLU):
1332
  """Verifies the cluster disks sizes.
1333

1334
  """
1335
  _OP_REQP = ["instances"]
1336
  REQ_BGL = False
1337

    
1338
  def ExpandNames(self):
1339

    
1340
    if not isinstance(self.op.instances, list):
1341
      raise errors.OpPrereqError("Invalid argument type 'instances'")
1342

    
1343
    if self.op.instances:
1344
      self.wanted_names = []
1345
      for name in self.op.instances:
1346
        full_name = self.cfg.ExpandInstanceName(name)
1347
        if full_name is None:
1348
          raise errors.OpPrereqError("Instance '%s' not known" % name)
1349
        self.wanted_names.append(full_name)
1350
      self.needed_locks = {
1351
        locking.LEVEL_NODE: [],
1352
        locking.LEVEL_INSTANCE: self.wanted_names,
1353
        }
1354
      self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
1355
    else:
1356
      self.wanted_names = None
1357
      self.needed_locks = {
1358
        locking.LEVEL_NODE: locking.ALL_SET,
1359
        locking.LEVEL_INSTANCE: locking.ALL_SET,
1360
        }
1361
    self.share_locks = dict(((i, 1) for i in locking.LEVELS))
1362

    
1363
  def DeclareLocks(self, level):
1364
    if level == locking.LEVEL_NODE and self.wanted_names is not None:
1365
      self._LockInstancesNodes(primary_only=True)
1366

    
1367
  def CheckPrereq(self):
1368
    """Check prerequisites.
1369

1370
    This only checks the optional instance list against the existing names.
1371

1372
    """
1373
    if self.wanted_names is None:
1374
      self.wanted_names = self.acquired_locks[locking.LEVEL_INSTANCE]
1375

    
1376
    self.wanted_instances = [self.cfg.GetInstanceInfo(name) for name
1377
                             in self.wanted_names]
1378

    
1379
  def _EnsureChildSizes(self, disk):
1380
    """Ensure children of the disk have the needed disk size.
1381

1382
    This is valid mainly for DRBD8 and fixes an issue where the
1383
    children have smaller disk size.
1384

1385
    @param disk: an L{ganeti.objects.Disk} object
1386

1387
    """
1388
    if disk.dev_type == constants.LD_DRBD8:
1389
      assert disk.children, "Empty children for DRBD8?"
1390
      fchild = disk.children[0]
1391
      mismatch = fchild.size < disk.size
1392
      if mismatch:
1393
        self.LogInfo("Child disk has size %d, parent %d, fixing",
1394
                     fchild.size, disk.size)
1395
        fchild.size = disk.size
1396

    
1397
      # and we recurse on this child only, not on the metadev
1398
      return self._EnsureChildSizes(fchild) or mismatch
1399
    else:
1400
      return False
1401

    
1402
  def Exec(self, feedback_fn):
1403
    """Verify the size of cluster disks.
1404

1405
    """
1406
    # TODO: check child disks too
1407
    # TODO: check differences in size between primary/secondary nodes
1408
    per_node_disks = {}
1409
    for instance in self.wanted_instances:
1410
      pnode = instance.primary_node
1411
      if pnode not in per_node_disks:
1412
        per_node_disks[pnode] = []
1413
      for idx, disk in enumerate(instance.disks):
1414
        per_node_disks[pnode].append((instance, idx, disk))
1415

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

    
1452

    
1453
class LURenameCluster(LogicalUnit):
1454
  """Rename the cluster.
1455

1456
  """
1457
  HPATH = "cluster-rename"
1458
  HTYPE = constants.HTYPE_CLUSTER
1459
  _OP_REQP = ["name"]
1460

    
1461
  def BuildHooksEnv(self):
1462
    """Build hooks env.
1463

1464
    """
1465
    env = {
1466
      "OP_TARGET": self.cfg.GetClusterName(),
1467
      "NEW_NAME": self.op.name,
1468
      }
1469
    mn = self.cfg.GetMasterNode()
1470
    return env, [mn], [mn]
1471

    
1472
  def CheckPrereq(self):
1473
    """Verify that the passed name is a valid one.
1474

1475
    """
1476
    hostname = utils.HostInfo(self.op.name)
1477

    
1478
    new_name = hostname.name
1479
    self.ip = new_ip = hostname.ip
1480
    old_name = self.cfg.GetClusterName()
1481
    old_ip = self.cfg.GetMasterIP()
1482
    if new_name == old_name and new_ip == old_ip:
1483
      raise errors.OpPrereqError("Neither the name nor the IP address of the"
1484
                                 " cluster has changed")
1485
    if new_ip != old_ip:
1486
      if utils.TcpPing(new_ip, constants.DEFAULT_NODED_PORT):
1487
        raise errors.OpPrereqError("The given cluster IP address (%s) is"
1488
                                   " reachable on the network. Aborting." %
1489
                                   new_ip)
1490

    
1491
    self.op.name = new_name
1492

    
1493
  def Exec(self, feedback_fn):
1494
    """Rename the cluster.
1495

1496
    """
1497
    clustername = self.op.name
1498
    ip = self.ip
1499

    
1500
    # shutdown the master IP
1501
    master = self.cfg.GetMasterNode()
1502
    result = self.rpc.call_node_stop_master(master, False)
1503
    if result.failed or not result.data:
1504
      raise errors.OpExecError("Could not disable the master role")
1505

    
1506
    try:
1507
      cluster = self.cfg.GetClusterInfo()
1508
      cluster.cluster_name = clustername
1509
      cluster.master_ip = ip
1510
      self.cfg.Update(cluster)
1511

    
1512
      # update the known hosts file
1513
      ssh.WriteKnownHostsFile(self.cfg, constants.SSH_KNOWN_HOSTS_FILE)
1514
      node_list = self.cfg.GetNodeList()
1515
      try:
1516
        node_list.remove(master)
1517
      except ValueError:
1518
        pass
1519
      result = self.rpc.call_upload_file(node_list,
1520
                                         constants.SSH_KNOWN_HOSTS_FILE)
1521
      for to_node, to_result in result.iteritems():
1522
        if to_result.failed or not to_result.data:
1523
          logging.error("Copy of file %s to node %s failed",
1524
                        constants.SSH_KNOWN_HOSTS_FILE, to_node)
1525

    
1526
    finally:
1527
      result = self.rpc.call_node_start_master(master, False, False)
1528
      if result.failed or not result.data:
1529
        self.LogWarning("Could not re-enable the master role on"
1530
                        " the master, please restart manually.")
1531

    
1532

    
1533
def _RecursiveCheckIfLVMBased(disk):
1534
  """Check if the given disk or its children are lvm-based.
1535

1536
  @type disk: L{objects.Disk}
1537
  @param disk: the disk to check
1538
  @rtype: boolean
1539
  @return: boolean indicating whether a LD_LV dev_type was found or not
1540

1541
  """
1542
  if disk.children:
1543
    for chdisk in disk.children:
1544
      if _RecursiveCheckIfLVMBased(chdisk):
1545
        return True
1546
  return disk.dev_type == constants.LD_LV
1547

    
1548

    
1549
class LUSetClusterParams(LogicalUnit):
1550
  """Change the parameters of the cluster.
1551

1552
  """
1553
  HPATH = "cluster-modify"
1554
  HTYPE = constants.HTYPE_CLUSTER
1555
  _OP_REQP = []
1556
  REQ_BGL = False
1557

    
1558
  def CheckArguments(self):
1559
    """Check parameters
1560

1561
    """
1562
    if not hasattr(self.op, "candidate_pool_size"):
1563
      self.op.candidate_pool_size = None
1564
    if self.op.candidate_pool_size is not None:
1565
      try:
1566
        self.op.candidate_pool_size = int(self.op.candidate_pool_size)
1567
      except (ValueError, TypeError), err:
1568
        raise errors.OpPrereqError("Invalid candidate_pool_size value: %s" %
1569
                                   str(err))
1570
      if self.op.candidate_pool_size < 1:
1571
        raise errors.OpPrereqError("At least one master candidate needed")
1572

    
1573
  def ExpandNames(self):
1574
    # FIXME: in the future maybe other cluster params won't require checking on
1575
    # all nodes to be modified.
1576
    self.needed_locks = {
1577
      locking.LEVEL_NODE: locking.ALL_SET,
1578
    }
1579
    self.share_locks[locking.LEVEL_NODE] = 1
1580

    
1581
  def BuildHooksEnv(self):
1582
    """Build hooks env.
1583

1584
    """
1585
    env = {
1586
      "OP_TARGET": self.cfg.GetClusterName(),
1587
      "NEW_VG_NAME": self.op.vg_name,
1588
      }
1589
    mn = self.cfg.GetMasterNode()
1590
    return env, [mn], [mn]
1591

    
1592
  def CheckPrereq(self):
1593
    """Check prerequisites.
1594

1595
    This checks whether the given params don't conflict and
1596
    if the given volume group is valid.
1597

1598
    """
1599
    if self.op.vg_name is not None and not self.op.vg_name:
1600
      instances = self.cfg.GetAllInstancesInfo().values()
1601
      for inst in instances:
1602
        for disk in inst.disks:
1603
          if _RecursiveCheckIfLVMBased(disk):
1604
            raise errors.OpPrereqError("Cannot disable lvm storage while"
1605
                                       " lvm-based instances exist")
1606

    
1607
    node_list = self.acquired_locks[locking.LEVEL_NODE]
1608

    
1609
    # if vg_name not None, checks given volume group on all nodes
1610
    if self.op.vg_name:
1611
      vglist = self.rpc.call_vg_list(node_list)
1612
      for node in node_list:
1613
        if vglist[node].failed:
1614
          # ignoring down node
1615
          self.LogWarning("Node %s unreachable/error, ignoring" % node)
1616
          continue
1617
        vgstatus = utils.CheckVolumeGroupSize(vglist[node].data,
1618
                                              self.op.vg_name,
1619
                                              constants.MIN_VG_SIZE)
1620
        if vgstatus:
1621
          raise errors.OpPrereqError("Error on node '%s': %s" %
1622
                                     (node, vgstatus))
1623

    
1624
    self.cluster = cluster = self.cfg.GetClusterInfo()
1625
    # validate beparams changes
1626
    if self.op.beparams:
1627
      utils.ForceDictType(self.op.beparams, constants.BES_PARAMETER_TYPES)
1628
      self.new_beparams = cluster.FillDict(
1629
        cluster.beparams[constants.BEGR_DEFAULT], self.op.beparams)
1630

    
1631
    # hypervisor list/parameters
1632
    self.new_hvparams = cluster.FillDict(cluster.hvparams, {})
1633
    if self.op.hvparams:
1634
      if not isinstance(self.op.hvparams, dict):
1635
        raise errors.OpPrereqError("Invalid 'hvparams' parameter on input")
1636
      for hv_name, hv_dict in self.op.hvparams.items():
1637
        if hv_name not in self.new_hvparams:
1638
          self.new_hvparams[hv_name] = hv_dict
1639
        else:
1640
          self.new_hvparams[hv_name].update(hv_dict)
1641

    
1642
    if self.op.enabled_hypervisors is not None:
1643
      self.hv_list = self.op.enabled_hypervisors
1644
      if not self.hv_list:
1645
        raise errors.OpPrereqError("Enabled hypervisors list must contain at"
1646
                                   " least one member")
1647
      invalid_hvs = set(self.hv_list) - constants.HYPER_TYPES
1648
      if invalid_hvs:
1649
        raise errors.OpPrereqError("Enabled hypervisors contains invalid"
1650
                                   " entries: %s" % invalid_hvs)
1651
    else:
1652
      self.hv_list = cluster.enabled_hypervisors
1653

    
1654
    if self.op.hvparams or self.op.enabled_hypervisors is not None:
1655
      # either the enabled list has changed, or the parameters have, validate
1656
      for hv_name, hv_params in self.new_hvparams.items():
1657
        if ((self.op.hvparams and hv_name in self.op.hvparams) or
1658
            (self.op.enabled_hypervisors and
1659
             hv_name in self.op.enabled_hypervisors)):
1660
          # either this is a new hypervisor, or its parameters have changed
1661
          hv_class = hypervisor.GetHypervisor(hv_name)
1662
          utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
1663
          hv_class.CheckParameterSyntax(hv_params)
1664
          _CheckHVParams(self, node_list, hv_name, hv_params)
1665

    
1666
  def Exec(self, feedback_fn):
1667
    """Change the parameters of the cluster.
1668

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

    
1690
    self.cfg.Update(self.cluster)
1691

    
1692

    
1693
class LURedistributeConfig(NoHooksLU):
1694
  """Force the redistribution of cluster configuration.
1695

1696
  This is a very simple LU.
1697

1698
  """
1699
  _OP_REQP = []
1700
  REQ_BGL = False
1701

    
1702
  def ExpandNames(self):
1703
    self.needed_locks = {
1704
      locking.LEVEL_NODE: locking.ALL_SET,
1705
    }
1706
    self.share_locks[locking.LEVEL_NODE] = 1
1707

    
1708
  def CheckPrereq(self):
1709
    """Check prerequisites.
1710

1711
    """
1712

    
1713
  def Exec(self, feedback_fn):
1714
    """Redistribute the configuration.
1715

1716
    """
1717
    self.cfg.Update(self.cfg.GetClusterInfo())
1718

    
1719

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

1723
  """
1724
  if not instance.disks:
1725
    return True
1726

    
1727
  if not oneshot:
1728
    lu.proc.LogInfo("Waiting for instance %s to sync disks." % instance.name)
1729

    
1730
  node = instance.primary_node
1731

    
1732
  for dev in instance.disks:
1733
    lu.cfg.SetDiskID(dev, node)
1734

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

    
1770
    # if we're done but degraded, let's do a few small retries, to
1771
    # make sure we see a stable and not transient situation; therefore
1772
    # we force restart of the loop
1773
    if (done or oneshot) and cumul_degraded and degr_retries > 0:
1774
      logging.info("Degraded disks found, %d retries left", degr_retries)
1775
      degr_retries -= 1
1776
      time.sleep(1)
1777
      continue
1778

    
1779
    if done or oneshot:
1780
      break
1781

    
1782
    time.sleep(min(60, max_time))
1783

    
1784
  if done:
1785
    lu.proc.LogInfo("Instance %s's disks are in sync." % instance.name)
1786
  return not cumul_degraded
1787

    
1788

    
1789
def _CheckDiskConsistency(lu, dev, node, on_primary, ldisk=False):
1790
  """Check that mirrors are not degraded.
1791

1792
  The ldisk parameter, if True, will change the test from the
1793
  is_degraded attribute (which represents overall non-ok status for
1794
  the device(s)) to the ldisk (representing the local storage status).
1795

1796
  """
1797
  lu.cfg.SetDiskID(dev, node)
1798
  if ldisk:
1799
    idx = 6
1800
  else:
1801
    idx = 5
1802

    
1803
  result = True
1804
  if on_primary or dev.AssembleOnSecondary():
1805
    rstats = lu.rpc.call_blockdev_find(node, dev)
1806
    msg = rstats.RemoteFailMsg()
1807
    if msg:
1808
      lu.LogWarning("Can't find disk on node %s: %s", node, msg)
1809
      result = False
1810
    elif not rstats.payload:
1811
      lu.LogWarning("Can't find disk on node %s", node)
1812
      result = False
1813
    else:
1814
      result = result and (not rstats.payload[idx])
1815
  if dev.children:
1816
    for child in dev.children:
1817
      result = result and _CheckDiskConsistency(lu, child, node, on_primary)
1818

    
1819
  return result
1820

    
1821

    
1822
class LUDiagnoseOS(NoHooksLU):
1823
  """Logical unit for OS diagnose/query.
1824

1825
  """
1826
  _OP_REQP = ["output_fields", "names"]
1827
  REQ_BGL = False
1828
  _FIELDS_STATIC = utils.FieldSet()
1829
  _FIELDS_DYNAMIC = utils.FieldSet("name", "valid", "node_status")
1830

    
1831
  def ExpandNames(self):
1832
    if self.op.names:
1833
      raise errors.OpPrereqError("Selective OS query not supported")
1834

    
1835
    _CheckOutputFields(static=self._FIELDS_STATIC,
1836
                       dynamic=self._FIELDS_DYNAMIC,
1837
                       selected=self.op.output_fields)
1838

    
1839
    # Lock all nodes, in shared mode
1840
    # Temporary removal of locks, should be reverted later
1841
    # TODO: reintroduce locks when they are lighter-weight
1842
    self.needed_locks = {}
1843
    #self.share_locks[locking.LEVEL_NODE] = 1
1844
    #self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
1845

    
1846
  def CheckPrereq(self):
1847
    """Check prerequisites.
1848

1849
    """
1850

    
1851
  @staticmethod
1852
  def _DiagnoseByOS(node_list, rlist):
1853
    """Remaps a per-node return list into an a per-os per-node dictionary
1854

1855
    @param node_list: a list with the names of all nodes
1856
    @param rlist: a map with node names as keys and OS objects as values
1857

1858
    @rtype: dict
1859
    @return: a dictionary with osnames as keys and as value another map, with
1860
        nodes as keys and list of OS objects as values, eg::
1861

1862
          {"debian-etch": {"node1": [<object>,...],
1863
                           "node2": [<object>,]}
1864
          }
1865

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

    
1886
  def Exec(self, feedback_fn):
1887
    """Compute the list of OSes.
1888

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

    
1912
    return output
1913

    
1914

    
1915
class LURemoveNode(LogicalUnit):
1916
  """Logical unit for removing a node.
1917

1918
  """
1919
  HPATH = "node-remove"
1920
  HTYPE = constants.HTYPE_NODE
1921
  _OP_REQP = ["node_name"]
1922

    
1923
  def BuildHooksEnv(self):
1924
    """Build hooks env.
1925

1926
    This doesn't run on the target node in the pre phase as a failed
1927
    node would then be impossible to remove.
1928

1929
    """
1930
    env = {
1931
      "OP_TARGET": self.op.node_name,
1932
      "NODE_NAME": self.op.node_name,
1933
      }
1934
    all_nodes = self.cfg.GetNodeList()
1935
    all_nodes.remove(self.op.node_name)
1936
    return env, all_nodes, all_nodes
1937

    
1938
  def CheckPrereq(self):
1939
    """Check prerequisites.
1940

1941
    This checks:
1942
     - the node exists in the configuration
1943
     - it does not have primary or secondary instances
1944
     - it's not the master
1945

1946
    Any errors are signaled by raising errors.OpPrereqError.
1947

1948
    """
1949
    node = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.node_name))
1950
    if node is None:
1951
      raise errors.OpPrereqError, ("Node '%s' is unknown." % self.op.node_name)
1952

    
1953
    instance_list = self.cfg.GetInstanceList()
1954

    
1955
    masternode = self.cfg.GetMasterNode()
1956
    if node.name == masternode:
1957
      raise errors.OpPrereqError("Node is the master node,"
1958
                                 " you need to failover first.")
1959

    
1960
    for instance_name in instance_list:
1961
      instance = self.cfg.GetInstanceInfo(instance_name)
1962
      if node.name in instance.all_nodes:
1963
        raise errors.OpPrereqError("Instance %s is still running on the node,"
1964
                                   " please remove first." % instance_name)
1965
    self.op.node_name = node.name
1966
    self.node = node
1967

    
1968
  def Exec(self, feedback_fn):
1969
    """Removes the node from the cluster.
1970

1971
    """
1972
    node = self.node
1973
    logging.info("Stopping the node daemon and removing configs from node %s",
1974
                 node.name)
1975

    
1976
    self.context.RemoveNode(node.name)
1977

    
1978
    self.rpc.call_node_leave_cluster(node.name)
1979

    
1980
    # Promote nodes to master candidate as needed
1981
    _AdjustCandidatePool(self)
1982

    
1983

    
1984
class LUQueryNodes(NoHooksLU):
1985
  """Logical unit for querying nodes.
1986

1987
  """
1988
  _OP_REQP = ["output_fields", "names", "use_locking"]
1989
  REQ_BGL = False
1990
  _FIELDS_DYNAMIC = utils.FieldSet(
1991
    "dtotal", "dfree",
1992
    "mtotal", "mnode", "mfree",
1993
    "bootid",
1994
    "ctotal", "cnodes", "csockets",
1995
    )
1996

    
1997
  _FIELDS_STATIC = utils.FieldSet(
1998
    "name", "pinst_cnt", "sinst_cnt",
1999
    "pinst_list", "sinst_list",
2000
    "pip", "sip", "tags",
2001
    "serial_no",
2002
    "master_candidate",
2003
    "master",
2004
    "offline",
2005
    "drained",
2006
    "role",
2007
    )
2008

    
2009
  def ExpandNames(self):
2010
    _CheckOutputFields(static=self._FIELDS_STATIC,
2011
                       dynamic=self._FIELDS_DYNAMIC,
2012
                       selected=self.op.output_fields)
2013

    
2014
    self.needed_locks = {}
2015
    self.share_locks[locking.LEVEL_NODE] = 1
2016

    
2017
    if self.op.names:
2018
      self.wanted = _GetWantedNodes(self, self.op.names)
2019
    else:
2020
      self.wanted = locking.ALL_SET
2021

    
2022
    self.do_node_query = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
2023
    self.do_locking = self.do_node_query and self.op.use_locking
2024
    if self.do_locking:
2025
      # if we don't request only static fields, we need to lock the nodes
2026
      self.needed_locks[locking.LEVEL_NODE] = self.wanted
2027

    
2028

    
2029
  def CheckPrereq(self):
2030
    """Check prerequisites.
2031

2032
    """
2033
    # The validation of the node list is done in the _GetWantedNodes,
2034
    # if non empty, and if empty, there's no validation to do
2035
    pass
2036

    
2037
  def Exec(self, feedback_fn):
2038
    """Computes the list of nodes and their attributes.
2039

2040
    """
2041
    all_info = self.cfg.GetAllNodesInfo()
2042
    if self.do_locking:
2043
      nodenames = self.acquired_locks[locking.LEVEL_NODE]
2044
    elif self.wanted != locking.ALL_SET:
2045
      nodenames = self.wanted
2046
      missing = set(nodenames).difference(all_info.keys())
2047
      if missing:
2048
        raise errors.OpExecError(
2049
          "Some nodes were removed before retrieving their data: %s" % missing)
2050
    else:
2051
      nodenames = all_info.keys()
2052

    
2053
    nodenames = utils.NiceSort(nodenames)
2054
    nodelist = [all_info[name] for name in nodenames]
2055

    
2056
    # begin data gathering
2057

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

    
2083
    node_to_primary = dict([(name, set()) for name in nodenames])
2084
    node_to_secondary = dict([(name, set()) for name in nodenames])
2085

    
2086
    inst_fields = frozenset(("pinst_cnt", "pinst_list",
2087
                             "sinst_cnt", "sinst_list"))
2088
    if inst_fields & frozenset(self.op.output_fields):
2089
      inst_data = self.cfg.GetAllInstancesInfo()
2090

    
2091
      for instance_name, inst in inst_data.items():
2092
        if inst.primary_node in node_to_primary:
2093
          node_to_primary[inst.primary_node].add(inst.name)
2094
        for secnode in inst.secondary_nodes:
2095
          if secnode in node_to_secondary:
2096
            node_to_secondary[secnode].add(inst.name)
2097

    
2098
    master_node = self.cfg.GetMasterNode()
2099

    
2100
    # end data gathering
2101

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

    
2150
    return output
2151

    
2152

    
2153
class LUQueryNodeVolumes(NoHooksLU):
2154
  """Logical unit for getting volumes on node(s).
2155

2156
  """
2157
  _OP_REQP = ["nodes", "output_fields"]
2158
  REQ_BGL = False
2159
  _FIELDS_DYNAMIC = utils.FieldSet("phys", "vg", "name", "size", "instance")
2160
  _FIELDS_STATIC = utils.FieldSet("node")
2161

    
2162
  def ExpandNames(self):
2163
    _CheckOutputFields(static=self._FIELDS_STATIC,
2164
                       dynamic=self._FIELDS_DYNAMIC,
2165
                       selected=self.op.output_fields)
2166

    
2167
    self.needed_locks = {}
2168
    self.share_locks[locking.LEVEL_NODE] = 1
2169
    if not self.op.nodes:
2170
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
2171
    else:
2172
      self.needed_locks[locking.LEVEL_NODE] = \
2173
        _GetWantedNodes(self, self.op.nodes)
2174

    
2175
  def CheckPrereq(self):
2176
    """Check prerequisites.
2177

2178
    This checks that the fields required are valid output fields.
2179

2180
    """
2181
    self.nodes = self.acquired_locks[locking.LEVEL_NODE]
2182

    
2183
  def Exec(self, feedback_fn):
2184
    """Computes the list of nodes and their attributes.
2185

2186
    """
2187
    nodenames = self.nodes
2188
    volumes = self.rpc.call_node_volumes(nodenames)
2189

    
2190
    ilist = [self.cfg.GetInstanceInfo(iname) for iname
2191
             in self.cfg.GetInstanceList()]
2192

    
2193
    lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
2194

    
2195
    output = []
2196
    for node in nodenames:
2197
      if node not in volumes or volumes[node].failed or not volumes[node].data:
2198
        continue
2199

    
2200
      node_vols = volumes[node].data[:]
2201
      node_vols.sort(key=lambda vol: vol['dev'])
2202

    
2203
      for vol in node_vols:
2204
        node_output = []
2205
        for field in self.op.output_fields:
2206
          if field == "node":
2207
            val = node
2208
          elif field == "phys":
2209
            val = vol['dev']
2210
          elif field == "vg":
2211
            val = vol['vg']
2212
          elif field == "name":
2213
            val = vol['name']
2214
          elif field == "size":
2215
            val = int(float(vol['size']))
2216
          elif field == "instance":
2217
            for inst in ilist:
2218
              if node not in lv_by_node[inst]:
2219
                continue
2220
              if vol['name'] in lv_by_node[inst][node]:
2221
                val = inst.name
2222
                break
2223
            else:
2224
              val = '-'
2225
          else:
2226
            raise errors.ParameterError(field)
2227
          node_output.append(str(val))
2228

    
2229
        output.append(node_output)
2230

    
2231
    return output
2232

    
2233

    
2234
class LUAddNode(LogicalUnit):
2235
  """Logical unit for adding node to the cluster.
2236

2237
  """
2238
  HPATH = "node-add"
2239
  HTYPE = constants.HTYPE_NODE
2240
  _OP_REQP = ["node_name"]
2241

    
2242
  def BuildHooksEnv(self):
2243
    """Build hooks env.
2244

2245
    This will run on all nodes before, and on all nodes + the new node after.
2246

2247
    """
2248
    env = {
2249
      "OP_TARGET": self.op.node_name,
2250
      "NODE_NAME": self.op.node_name,
2251
      "NODE_PIP": self.op.primary_ip,
2252
      "NODE_SIP": self.op.secondary_ip,
2253
      }
2254
    nodes_0 = self.cfg.GetNodeList()
2255
    nodes_1 = nodes_0 + [self.op.node_name, ]
2256
    return env, nodes_0, nodes_1
2257

    
2258
  def CheckPrereq(self):
2259
    """Check prerequisites.
2260

2261
    This checks:
2262
     - the new node is not already in the config
2263
     - it is resolvable
2264
     - its parameters (single/dual homed) matches the cluster
2265

2266
    Any errors are signaled by raising errors.OpPrereqError.
2267

2268
    """
2269
    node_name = self.op.node_name
2270
    cfg = self.cfg
2271

    
2272
    dns_data = utils.HostInfo(node_name)
2273

    
2274
    node = dns_data.name
2275
    primary_ip = self.op.primary_ip = dns_data.ip
2276
    secondary_ip = getattr(self.op, "secondary_ip", None)
2277
    if secondary_ip is None:
2278
      secondary_ip = primary_ip
2279
    if not utils.IsValidIP(secondary_ip):
2280
      raise errors.OpPrereqError("Invalid secondary IP given")
2281
    self.op.secondary_ip = secondary_ip
2282

    
2283
    node_list = cfg.GetNodeList()
2284
    if not self.op.readd and node in node_list:
2285
      raise errors.OpPrereqError("Node %s is already in the configuration" %
2286
                                 node)
2287
    elif self.op.readd and node not in node_list:
2288
      raise errors.OpPrereqError("Node %s is not in the configuration" % node)
2289

    
2290
    for existing_node_name in node_list:
2291
      existing_node = cfg.GetNodeInfo(existing_node_name)
2292

    
2293
      if self.op.readd and node == existing_node_name:
2294
        if (existing_node.primary_ip != primary_ip or
2295
            existing_node.secondary_ip != secondary_ip):
2296
          raise errors.OpPrereqError("Readded node doesn't have the same IP"
2297
                                     " address configuration as before")
2298
        continue
2299

    
2300
      if (existing_node.primary_ip == primary_ip or
2301
          existing_node.secondary_ip == primary_ip or
2302
          existing_node.primary_ip == secondary_ip or
2303
          existing_node.secondary_ip == secondary_ip):
2304
        raise errors.OpPrereqError("New node ip address(es) conflict with"
2305
                                   " existing node %s" % existing_node.name)
2306

    
2307
    # check that the type of the node (single versus dual homed) is the
2308
    # same as for the master
2309
    myself = cfg.GetNodeInfo(self.cfg.GetMasterNode())
2310
    master_singlehomed = myself.secondary_ip == myself.primary_ip
2311
    newbie_singlehomed = secondary_ip == primary_ip
2312
    if master_singlehomed != newbie_singlehomed:
2313
      if master_singlehomed:
2314
        raise errors.OpPrereqError("The master has no private ip but the"
2315
                                   " new node has one")
2316
      else:
2317
        raise errors.OpPrereqError("The master has a private ip but the"
2318
                                   " new node doesn't have one")
2319

    
2320
    # checks reachability
2321
    if not utils.TcpPing(primary_ip, constants.DEFAULT_NODED_PORT):
2322
      raise errors.OpPrereqError("Node not reachable by ping")
2323

    
2324
    if not newbie_singlehomed:
2325
      # check reachability from my secondary ip to newbie's secondary ip
2326
      if not utils.TcpPing(secondary_ip, constants.DEFAULT_NODED_PORT,
2327
                           source=myself.secondary_ip):
2328
        raise errors.OpPrereqError("Node secondary ip not reachable by TCP"
2329
                                   " based ping to noded port")
2330

    
2331
    cp_size = self.cfg.GetClusterInfo().candidate_pool_size
2332
    if self.op.readd:
2333
      exceptions = [node]
2334
    else:
2335
      exceptions = []
2336
    mc_now, mc_max = self.cfg.GetMasterCandidateStats(exceptions)
2337
    # the new node will increase mc_max with one, so:
2338
    mc_max = min(mc_max + 1, cp_size)
2339
    self.master_candidate = mc_now < mc_max
2340

    
2341
    if self.op.readd:
2342
      self.new_node = self.cfg.GetNodeInfo(node)
2343
      assert self.new_node is not None, "Can't retrieve locked node %s" % node
2344
    else:
2345
      self.new_node = objects.Node(name=node,
2346
                                   primary_ip=primary_ip,
2347
                                   secondary_ip=secondary_ip,
2348
                                   master_candidate=self.master_candidate,
2349
                                   offline=False, drained=False)
2350

    
2351
  def Exec(self, feedback_fn):
2352
    """Adds the new node to the cluster.
2353

2354
    """
2355
    new_node = self.new_node
2356
    node = new_node.name
2357

    
2358
    # for re-adds, reset the offline/drained/master-candidate flags;
2359
    # we need to reset here, otherwise offline would prevent RPC calls
2360
    # later in the procedure; this also means that if the re-add
2361
    # fails, we are left with a non-offlined, broken node
2362
    if self.op.readd:
2363
      new_node.drained = new_node.offline = False
2364
      self.LogInfo("Readding a node, the offline/drained flags were reset")
2365
      # if we demote the node, we do cleanup later in the procedure
2366
      new_node.master_candidate = self.master_candidate
2367

    
2368
    # notify the user about any possible mc promotion
2369
    if new_node.master_candidate:
2370
      self.LogInfo("Node will be a master candidate")
2371

    
2372
    # check connectivity
2373
    result = self.rpc.call_version([node])[node]
2374
    result.Raise()
2375
    if result.data:
2376
      if constants.PROTOCOL_VERSION == result.data:
2377
        logging.info("Communication to node %s fine, sw version %s match",
2378
                     node, result.data)
2379
      else:
2380
        raise errors.OpExecError("Version mismatch master version %s,"
2381
                                 " node version %s" %
2382
                                 (constants.PROTOCOL_VERSION, result.data))
2383
    else:
2384
      raise errors.OpExecError("Cannot get version from the new node")
2385

    
2386
    # setup ssh on node
2387
    logging.info("Copy ssh key to node %s", node)
2388
    priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
2389
    keyarray = []
2390
    keyfiles = [constants.SSH_HOST_DSA_PRIV, constants.SSH_HOST_DSA_PUB,
2391
                constants.SSH_HOST_RSA_PRIV, constants.SSH_HOST_RSA_PUB,
2392
                priv_key, pub_key]
2393

    
2394
    for i in keyfiles:
2395
      f = open(i, 'r')
2396
      try:
2397
        keyarray.append(f.read())
2398
      finally:
2399
        f.close()
2400

    
2401
    result = self.rpc.call_node_add(node, keyarray[0], keyarray[1],
2402
                                    keyarray[2],
2403
                                    keyarray[3], keyarray[4], keyarray[5])
2404

    
2405
    msg = result.RemoteFailMsg()
2406
    if msg:
2407
      raise errors.OpExecError("Cannot transfer ssh keys to the"
2408
                               " new node: %s" % msg)
2409

    
2410
    # Add node to our /etc/hosts, and add key to known_hosts
2411
    if self.cfg.GetClusterInfo().modify_etc_hosts:
2412
      utils.AddHostToEtcHosts(new_node.name)
2413

    
2414
    if new_node.secondary_ip != new_node.primary_ip:
2415
      result = self.rpc.call_node_has_ip_address(new_node.name,
2416
                                                 new_node.secondary_ip)
2417
      if result.failed or not result.data:
2418
        raise errors.OpExecError("Node claims it doesn't have the secondary ip"
2419
                                 " you gave (%s). Please fix and re-run this"
2420
                                 " command." % new_node.secondary_ip)
2421

    
2422
    node_verify_list = [self.cfg.GetMasterNode()]
2423
    node_verify_param = {
2424
      'nodelist': [node],
2425
      # TODO: do a node-net-test as well?
2426
    }
2427

    
2428
    result = self.rpc.call_node_verify(node_verify_list, node_verify_param,
2429
                                       self.cfg.GetClusterName())
2430
    for verifier in node_verify_list:
2431
      if result[verifier].failed or not result[verifier].data:
2432
        raise errors.OpExecError("Cannot communicate with %s's node daemon"
2433
                                 " for remote verification" % verifier)
2434
      if result[verifier].data['nodelist']:
2435
        for failed in result[verifier].data['nodelist']:
2436
          feedback_fn("ssh/hostname verification failed"
2437
                      " (checking from %s): %s" %
2438
                      (verifier, result[verifier].data['nodelist'][failed]))
2439
        raise errors.OpExecError("ssh/hostname verification failed.")
2440

    
2441
    # Distribute updated /etc/hosts and known_hosts to all nodes,
2442
    # including the node just added
2443
    myself = self.cfg.GetNodeInfo(self.cfg.GetMasterNode())
2444
    dist_nodes = self.cfg.GetNodeList()
2445
    if not self.op.readd:
2446
      dist_nodes.append(node)
2447
    if myself.name in dist_nodes:
2448
      dist_nodes.remove(myself.name)
2449

    
2450
    logging.debug("Copying hosts and known_hosts to all nodes")
2451
    for fname in (constants.ETC_HOSTS, constants.SSH_KNOWN_HOSTS_FILE):
2452
      result = self.rpc.call_upload_file(dist_nodes, fname)
2453
      for to_node, to_result in result.iteritems():
2454
        if to_result.failed or not to_result.data:
2455
          logging.error("Copy of file %s to node %s failed", fname, to_node)
2456

    
2457
    to_copy = []
2458
    enabled_hypervisors = self.cfg.GetClusterInfo().enabled_hypervisors
2459
    if constants.HTS_COPY_VNC_PASSWORD.intersection(enabled_hypervisors):
2460
      to_copy.append(constants.VNC_PASSWORD_FILE)
2461

    
2462
    for fname in to_copy:
2463
      result = self.rpc.call_upload_file([node], fname)
2464
      if result[node].failed or not result[node]:
2465
        logging.error("Could not copy file %s to node %s", fname, node)
2466

    
2467
    if self.op.readd:
2468
      self.context.ReaddNode(new_node)
2469
      # make sure we redistribute the config
2470
      self.cfg.Update(new_node)
2471
      # and make sure the new node will not have old files around
2472
      if not new_node.master_candidate:
2473
        result = self.rpc.call_node_demote_from_mc(new_node.name)
2474
        msg = result.RemoteFailMsg()
2475
        if msg:
2476
          self.LogWarning("Node failed to demote itself from master"
2477
                          " candidate status: %s" % msg)
2478
    else:
2479
      self.context.AddNode(new_node)
2480

    
2481

    
2482
class LUSetNodeParams(LogicalUnit):
2483
  """Modifies the parameters of a node.
2484

2485
  """
2486
  HPATH = "node-modify"
2487
  HTYPE = constants.HTYPE_NODE
2488
  _OP_REQP = ["node_name"]
2489
  REQ_BGL = False
2490

    
2491
  def CheckArguments(self):
2492
    node_name = self.cfg.ExpandNodeName(self.op.node_name)
2493
    if node_name is None:
2494
      raise errors.OpPrereqError("Invalid node name '%s'" % self.op.node_name)
2495
    self.op.node_name = node_name
2496
    _CheckBooleanOpField(self.op, 'master_candidate')
2497
    _CheckBooleanOpField(self.op, 'offline')
2498
    _CheckBooleanOpField(self.op, 'drained')
2499
    all_mods = [self.op.offline, self.op.master_candidate, self.op.drained]
2500
    if all_mods.count(None) == 3:
2501
      raise errors.OpPrereqError("Please pass at least one modification")
2502
    if all_mods.count(True) > 1:
2503
      raise errors.OpPrereqError("Can't set the node into more than one"
2504
                                 " state at the same time")
2505

    
2506
  def ExpandNames(self):
2507
    self.needed_locks = {locking.LEVEL_NODE: self.op.node_name}
2508

    
2509
  def BuildHooksEnv(self):
2510
    """Build hooks env.
2511

2512
    This runs on the master node.
2513

2514
    """
2515
    env = {
2516
      "OP_TARGET": self.op.node_name,
2517
      "MASTER_CANDIDATE": str(self.op.master_candidate),
2518
      "OFFLINE": str(self.op.offline),
2519
      "DRAINED": str(self.op.drained),
2520
      }
2521
    nl = [self.cfg.GetMasterNode(),
2522
          self.op.node_name]
2523
    return env, nl, nl
2524

    
2525
  def CheckPrereq(self):
2526
    """Check prerequisites.
2527

2528
    This only checks the instance list against the existing names.
2529

2530
    """
2531
    node = self.node = self.cfg.GetNodeInfo(self.op.node_name)
2532

    
2533
    if (self.op.master_candidate is not None or
2534
        self.op.drained is not None or
2535
        self.op.offline is not None):
2536
      # we can't change the master's node flags
2537
      if self.op.node_name == self.cfg.GetMasterNode():
2538
        raise errors.OpPrereqError("The master role can be changed"
2539
                                   " only via masterfailover")
2540

    
2541
    if ((self.op.master_candidate == False or self.op.offline == True or
2542
         self.op.drained == True) and node.master_candidate):
2543
      cp_size = self.cfg.GetClusterInfo().candidate_pool_size
2544
      num_candidates, _ = self.cfg.GetMasterCandidateStats()
2545
      if num_candidates <= cp_size:
2546
        msg = ("Not enough master candidates (desired"
2547
               " %d, new value will be %d)" % (cp_size, num_candidates-1))
2548
        if self.op.force:
2549
          self.LogWarning(msg)
2550
        else:
2551
          raise errors.OpPrereqError(msg)
2552

    
2553
    if (self.op.master_candidate == True and
2554
        ((node.offline and not self.op.offline == False) or
2555
         (node.drained and not self.op.drained == False))):
2556
      raise errors.OpPrereqError("Node '%s' is offline or drained, can't set"
2557
                                 " to master_candidate" % node.name)
2558

    
2559
    return
2560

    
2561
  def Exec(self, feedback_fn):
2562
    """Modifies a node.
2563

2564
    """
2565
    node = self.node
2566

    
2567
    result = []
2568
    changed_mc = False
2569

    
2570
    if self.op.offline is not None:
2571
      node.offline = self.op.offline
2572
      result.append(("offline", str(self.op.offline)))
2573
      if self.op.offline == True:
2574
        if node.master_candidate:
2575
          node.master_candidate = False
2576
          changed_mc = True
2577
          result.append(("master_candidate", "auto-demotion due to offline"))
2578
        if node.drained:
2579
          node.drained = False
2580
          result.append(("drained", "clear drained status due to offline"))
2581

    
2582
    if self.op.master_candidate is not None:
2583
      node.master_candidate = self.op.master_candidate
2584
      changed_mc = True
2585
      result.append(("master_candidate", str(self.op.master_candidate)))
2586
      if self.op.master_candidate == False:
2587
        rrc = self.rpc.call_node_demote_from_mc(node.name)
2588
        msg = rrc.RemoteFailMsg()
2589
        if msg:
2590
          self.LogWarning("Node failed to demote itself: %s" % msg)
2591

    
2592
    if self.op.drained is not None:
2593
      node.drained = self.op.drained
2594
      result.append(("drained", str(self.op.drained)))
2595
      if self.op.drained == True:
2596
        if node.master_candidate:
2597
          node.master_candidate = False
2598
          changed_mc = True
2599
          result.append(("master_candidate", "auto-demotion due to drain"))
2600
          rrc = self.rpc.call_node_demote_from_mc(node.name)
2601
          msg = rrc.RemoteFailMsg()
2602
          if msg:
2603
            self.LogWarning("Node failed to demote itself: %s" % msg)
2604
        if node.offline:
2605
          node.offline = False
2606
          result.append(("offline", "clear offline status due to drain"))
2607

    
2608
    # this will trigger configuration file update, if needed
2609
    self.cfg.Update(node)
2610
    # this will trigger job queue propagation or cleanup
2611
    if changed_mc:
2612
      self.context.ReaddNode(node)
2613

    
2614
    return result
2615

    
2616

    
2617
class LUQueryClusterInfo(NoHooksLU):
2618
  """Query cluster configuration.
2619

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

    
2624
  def ExpandNames(self):
2625
    self.needed_locks = {}
2626

    
2627
  def CheckPrereq(self):
2628
    """No prerequsites needed for this LU.
2629

2630
    """
2631
    pass
2632

    
2633
  def Exec(self, feedback_fn):
2634
    """Return cluster config.
2635

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

    
2660
    return result
2661

    
2662

    
2663
class LUQueryConfigValues(NoHooksLU):
2664
  """Return configuration values.
2665

2666
  """
2667
  _OP_REQP = []
2668
  REQ_BGL = False
2669
  _FIELDS_DYNAMIC = utils.FieldSet()
2670
  _FIELDS_STATIC = utils.FieldSet("cluster_name", "master_node", "drain_flag")
2671

    
2672
  def ExpandNames(self):
2673
    self.needed_locks = {}
2674

    
2675
    _CheckOutputFields(static=self._FIELDS_STATIC,
2676
                       dynamic=self._FIELDS_DYNAMIC,
2677
                       selected=self.op.output_fields)
2678

    
2679
  def CheckPrereq(self):
2680
    """No prerequisites.
2681

2682
    """
2683
    pass
2684

    
2685
  def Exec(self, feedback_fn):
2686
    """Dump a representation of the cluster config to the standard output.
2687

2688
    """
2689
    values = []
2690
    for field in self.op.output_fields:
2691
      if field == "cluster_name":
2692
        entry = self.cfg.GetClusterName()
2693
      elif field == "master_node":
2694
        entry = self.cfg.GetMasterNode()
2695
      elif field == "drain_flag":
2696
        entry = os.path.exists(constants.JOB_QUEUE_DRAIN_FILE)
2697
      else:
2698
        raise errors.ParameterError(field)
2699
      values.append(entry)
2700
    return values
2701

    
2702

    
2703
class LUActivateInstanceDisks(NoHooksLU):
2704
  """Bring up an instance's disks.
2705

2706
  """
2707
  _OP_REQP = ["instance_name"]
2708
  REQ_BGL = False
2709

    
2710
  def ExpandNames(self):
2711
    self._ExpandAndLockInstance()
2712
    self.needed_locks[locking.LEVEL_NODE] = []
2713
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2714

    
2715
  def DeclareLocks(self, level):
2716
    if level == locking.LEVEL_NODE:
2717
      self._LockInstancesNodes()
2718

    
2719
  def CheckPrereq(self):
2720
    """Check prerequisites.
2721

2722
    This checks that the instance is in the cluster.
2723

2724
    """
2725
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2726
    assert self.instance is not None, \
2727
      "Cannot retrieve locked instance %s" % self.op.instance_name
2728
    _CheckNodeOnline(self, self.instance.primary_node)
2729
    if not hasattr(self.op, "ignore_size"):
2730
      self.op.ignore_size = False
2731

    
2732
  def Exec(self, feedback_fn):
2733
    """Activate the disks.
2734

2735
    """
2736
    disks_ok, disks_info = \
2737
              _AssembleInstanceDisks(self, self.instance,
2738
                                     ignore_size=self.op.ignore_size)
2739
    if not disks_ok:
2740
      raise errors.OpExecError("Cannot activate block devices")
2741

    
2742
    return disks_info
2743

    
2744

    
2745
def _AssembleInstanceDisks(lu, instance, ignore_secondaries=False,
2746
                           ignore_size=False):
2747
  """Prepare the block devices for an instance.
2748

2749
  This sets up the block devices on all nodes.
2750

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

2766
  """
2767
  device_info = []
2768
  disks_ok = True
2769
  iname = instance.name
2770
  # With the two passes mechanism we try to reduce the window of
2771
  # opportunity for the race condition of switching DRBD to primary
2772
  # before handshaking occured, but we do not eliminate it
2773

    
2774
  # The proper fix would be to wait (with some limits) until the
2775
  # connection has been made and drbd transitions from WFConnection
2776
  # into any other network-connected state (Connected, SyncTarget,
2777
  # SyncSource, etc.)
2778

    
2779
  # 1st pass, assemble on all nodes in secondary mode
2780
  for inst_disk in instance.disks:
2781
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
2782
      if ignore_size:
2783
        node_disk = node_disk.Copy()
2784
        node_disk.UnsetSize()
2785
      lu.cfg.SetDiskID(node_disk, node)
2786
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, False)
2787
      msg = result.RemoteFailMsg()
2788
      if msg:
2789
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
2790
                           " (is_primary=False, pass=1): %s",
2791
                           inst_disk.iv_name, node, msg)
2792
        if not ignore_secondaries:
2793
          disks_ok = False
2794

    
2795
  # FIXME: race condition on drbd migration to primary
2796

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

    
2816
  # leave the disks configured for the primary node
2817
  # this is a workaround that would be fixed better by
2818
  # improving the logical/physical id handling
2819
  for disk in instance.disks:
2820
    lu.cfg.SetDiskID(disk, instance.primary_node)
2821

    
2822
  return disks_ok, device_info
2823

    
2824

    
2825
def _StartInstanceDisks(lu, instance, force):
2826
  """Start the disks of an instance.
2827

2828
  """
2829
  disks_ok, _ = _AssembleInstanceDisks(lu, instance,
2830
                                           ignore_secondaries=force)
2831
  if not disks_ok:
2832
    _ShutdownInstanceDisks(lu, instance)
2833
    if force is not None and not force:
2834
      lu.proc.LogWarning("", hint="If the message above refers to a"
2835
                         " secondary node,"
2836
                         " you can retry the operation using '--force'.")
2837
    raise errors.OpExecError("Disk consistency error")
2838

    
2839

    
2840
class LUDeactivateInstanceDisks(NoHooksLU):
2841
  """Shutdown an instance's disks.
2842

2843
  """
2844
  _OP_REQP = ["instance_name"]
2845
  REQ_BGL = False
2846

    
2847
  def ExpandNames(self):
2848
    self._ExpandAndLockInstance()
2849
    self.needed_locks[locking.LEVEL_NODE] = []
2850
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2851

    
2852
  def DeclareLocks(self, level):
2853
    if level == locking.LEVEL_NODE:
2854
      self._LockInstancesNodes()
2855

    
2856
  def CheckPrereq(self):
2857
    """Check prerequisites.
2858

2859
    This checks that the instance is in the cluster.
2860

2861
    """
2862
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2863
    assert self.instance is not None, \
2864
      "Cannot retrieve locked instance %s" % self.op.instance_name
2865

    
2866
  def Exec(self, feedback_fn):
2867
    """Deactivate the disks
2868

2869
    """
2870
    instance = self.instance
2871
    _SafeShutdownInstanceDisks(self, instance)
2872

    
2873

    
2874
def _SafeShutdownInstanceDisks(lu, instance):
2875
  """Shutdown block devices of an instance.
2876

2877
  This function checks if an instance is running, before calling
2878
  _ShutdownInstanceDisks.
2879

2880
  """
2881
  ins_l = lu.rpc.call_instance_list([instance.primary_node],
2882
                                      [instance.hypervisor])
2883
  ins_l = ins_l[instance.primary_node]
2884
  if ins_l.failed or not isinstance(ins_l.data, list):
2885
    raise errors.OpExecError("Can't contact node '%s'" %
2886
                             instance.primary_node)
2887

    
2888
  if instance.name in ins_l.data:
2889
    raise errors.OpExecError("Instance is running, can't shutdown"
2890
                             " block devices.")
2891

    
2892
  _ShutdownInstanceDisks(lu, instance)
2893

    
2894

    
2895
def _ShutdownInstanceDisks(lu, instance, ignore_primary=False):
2896
  """Shutdown block devices of an instance.
2897

2898
  This does the shutdown on all nodes of the instance.
2899

2900
  If the ignore_primary is false, errors on the primary node are
2901
  ignored.
2902

2903
  """
2904
  all_result = True
2905
  for disk in instance.disks:
2906
    for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
2907
      lu.cfg.SetDiskID(top_disk, node)
2908
      result = lu.rpc.call_blockdev_shutdown(node, top_disk)
2909
      msg = result.RemoteFailMsg()
2910
      if msg:
2911
        lu.LogWarning("Could not shutdown block device %s on node %s: %s",
2912
                      disk.iv_name, node, msg)
2913
        if not ignore_primary or node != instance.primary_node:
2914
          all_result = False
2915
  return all_result
2916

    
2917

    
2918
def _CheckNodeFreeMemory(lu, node, reason, requested, hypervisor_name):
2919
  """Checks if a node has enough free memory.
2920

2921
  This function check if a given node has the needed amount of free
2922
  memory. In case the node has less memory or we cannot get the
2923
  information from the node, this function raise an OpPrereqError
2924
  exception.
2925

2926
  @type lu: C{LogicalUnit}
2927
  @param lu: a logical unit from which we get configuration data
2928
  @type node: C{str}
2929
  @param node: the node to check
2930
  @type reason: C{str}
2931
  @param reason: string to use in the error message
2932
  @type requested: C{int}
2933
  @param requested: the amount of memory in MiB to check for
2934
  @type hypervisor_name: C{str}
2935
  @param hypervisor_name: the hypervisor to ask for memory stats
2936
  @raise errors.OpPrereqError: if the node doesn't have enough memory, or
2937
      we cannot check the node
2938

2939
  """
2940
  nodeinfo = lu.rpc.call_node_info([node], lu.cfg.GetVGName(), hypervisor_name)
2941
  nodeinfo[node].Raise()
2942
  free_mem = nodeinfo[node].data.get('memory_free')
2943
  if not isinstance(free_mem, int):
2944
    raise errors.OpPrereqError("Can't compute free memory on node %s, result"
2945
                             " was '%s'" % (node, free_mem))
2946
  if requested > free_mem:
2947
    raise errors.OpPrereqError("Not enough memory on node %s for %s:"
2948
                             " needed %s MiB, available %s MiB" %
2949
                             (node, reason, requested, free_mem))
2950

    
2951

    
2952
class LUStartupInstance(LogicalUnit):
2953
  """Starts an instance.
2954

2955
  """
2956
  HPATH = "instance-start"
2957
  HTYPE = constants.HTYPE_INSTANCE
2958
  _OP_REQP = ["instance_name", "force"]
2959
  REQ_BGL = False
2960

    
2961
  def ExpandNames(self):
2962
    self._ExpandAndLockInstance()
2963

    
2964
  def BuildHooksEnv(self):
2965
    """Build hooks env.
2966

2967
    This runs on master, primary and secondary nodes of the instance.
2968

2969
    """
2970
    env = {
2971
      "FORCE": self.op.force,
2972
      }
2973
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
2974
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2975
    return env, nl, nl
2976

    
2977
  def CheckPrereq(self):
2978
    """Check prerequisites.
2979

2980
    This checks that the instance is in the cluster.
2981

2982
    """
2983
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2984
    assert self.instance is not None, \
2985
      "Cannot retrieve locked instance %s" % self.op.instance_name
2986

    
2987
    # extra beparams
2988
    self.beparams = getattr(self.op, "beparams", {})
2989
    if self.beparams:
2990
      if not isinstance(self.beparams, dict):
2991
        raise errors.OpPrereqError("Invalid beparams passed: %s, expected"
2992
                                   " dict" % (type(self.beparams), ))
2993
      # fill the beparams dict
2994
      utils.ForceDictType(self.beparams, constants.BES_PARAMETER_TYPES)
2995
      self.op.beparams = self.beparams
2996

    
2997
    # extra hvparams
2998
    self.hvparams = getattr(self.op, "hvparams", {})
2999
    if self.hvparams:
3000
      if not isinstance(self.hvparams, dict):
3001
        raise errors.OpPrereqError("Invalid hvparams passed: %s, expected"
3002
                                   " dict" % (type(self.hvparams), ))
3003

    
3004
      # check hypervisor parameter syntax (locally)
3005
      cluster = self.cfg.GetClusterInfo()
3006
      utils.ForceDictType(self.hvparams, constants.HVS_PARAMETER_TYPES)
3007
      filled_hvp = cluster.FillDict(cluster.hvparams[instance.hypervisor],
3008
                                    instance.hvparams)
3009
      filled_hvp.update(self.hvparams)
3010
      hv_type = hypervisor.GetHypervisor(instance.hypervisor)
3011
      hv_type.CheckParameterSyntax(filled_hvp)
3012
      _CheckHVParams(self, instance.all_nodes, instance.hypervisor, filled_hvp)
3013
      self.op.hvparams = self.hvparams
3014

    
3015
    _CheckNodeOnline(self, instance.primary_node)
3016

    
3017
    bep = self.cfg.GetClusterInfo().FillBE(instance)
3018
    # check bridges existence
3019
    _CheckInstanceBridgesExist(self, instance)
3020

    
3021
    remote_info = self.rpc.call_instance_info(instance.primary_node,
3022
                                              instance.name,
3023
                                              instance.hypervisor)
3024
    remote_info.Raise()
3025
    if not remote_info.data:
3026
      _CheckNodeFreeMemory(self, instance.primary_node,
3027
                           "starting instance %s" % instance.name,
3028
                           bep[constants.BE_MEMORY], instance.hypervisor)
3029

    
3030
  def Exec(self, feedback_fn):
3031
    """Start the instance.
3032

3033
    """
3034
    instance = self.instance
3035
    force = self.op.force
3036

    
3037
    self.cfg.MarkInstanceUp(instance.name)
3038

    
3039
    node_current = instance.primary_node
3040

    
3041
    _StartInstanceDisks(self, instance, force)
3042

    
3043
    result = self.rpc.call_instance_start(node_current, instance,
3044
                                          self.hvparams, self.beparams)
3045
    msg = result.RemoteFailMsg()
3046
    if msg:
3047
      _ShutdownInstanceDisks(self, instance)
3048
      raise errors.OpExecError("Could not start instance: %s" % msg)
3049

    
3050

    
3051
class LURebootInstance(LogicalUnit):
3052
  """Reboot an instance.
3053

3054
  """
3055
  HPATH = "instance-reboot"
3056
  HTYPE = constants.HTYPE_INSTANCE
3057
  _OP_REQP = ["instance_name", "ignore_secondaries", "reboot_type"]
3058
  REQ_BGL = False
3059

    
3060
  def ExpandNames(self):
3061
    if self.op.reboot_type not in [constants.INSTANCE_REBOOT_SOFT,
3062
                                   constants.INSTANCE_REBOOT_HARD,
3063
                                   constants.INSTANCE_REBOOT_FULL]:
3064
      raise errors.ParameterError("reboot type not in [%s, %s, %s]" %
3065
                                  (constants.INSTANCE_REBOOT_SOFT,
3066
                                   constants.INSTANCE_REBOOT_HARD,
3067
                                   constants.INSTANCE_REBOOT_FULL))
3068
    self._ExpandAndLockInstance()
3069

    
3070
  def BuildHooksEnv(self):
3071
    """Build hooks env.
3072

3073
    This runs on master, primary and secondary nodes of the instance.
3074

3075
    """
3076
    env = {
3077
      "IGNORE_SECONDARIES": self.op.ignore_secondaries,
3078
      "REBOOT_TYPE": self.op.reboot_type,
3079
      }
3080
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3081
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3082
    return env, nl, nl
3083

    
3084
  def CheckPrereq(self):
3085
    """Check prerequisites.
3086

3087
    This checks that the instance is in the cluster.
3088

3089
    """
3090
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3091
    assert self.instance is not None, \
3092
      "Cannot retrieve locked instance %s" % self.op.instance_name
3093

    
3094
    _CheckNodeOnline(self, instance.primary_node)
3095

    
3096
    # check bridges existence
3097
    _CheckInstanceBridgesExist(self, instance)
3098

    
3099
  def Exec(self, feedback_fn):
3100
    """Reboot the instance.
3101

3102
    """
3103
    instance = self.instance
3104
    ignore_secondaries = self.op.ignore_secondaries
3105
    reboot_type = self.op.reboot_type
3106

    
3107
    node_current = instance.primary_node
3108

    
3109
    if reboot_type in [constants.INSTANCE_REBOOT_SOFT,
3110
                       constants.INSTANCE_REBOOT_HARD]:
3111
      for disk in instance.disks:
3112
        self.cfg.SetDiskID(disk, node_current)
3113
      result = self.rpc.call_instance_reboot(node_current, instance,
3114
                                             reboot_type)
3115
      msg = result.RemoteFailMsg()
3116
      if msg:
3117
        raise errors.OpExecError("Could not reboot instance: %s" % msg)
3118
    else:
3119
      result = self.rpc.call_instance_shutdown(node_current, instance)
3120
      msg = result.RemoteFailMsg()
3121
      if msg:
3122
        raise errors.OpExecError("Could not shutdown instance for"
3123
                                 " full reboot: %s" % msg)
3124
      _ShutdownInstanceDisks(self, instance)
3125
      _StartInstanceDisks(self, instance, ignore_secondaries)
3126
      result = self.rpc.call_instance_start(node_current, instance, None, None)
3127
      msg = result.RemoteFailMsg()
3128
      if msg:
3129
        _ShutdownInstanceDisks(self, instance)
3130
        raise errors.OpExecError("Could not start instance for"
3131
                                 " full reboot: %s" % msg)
3132

    
3133
    self.cfg.MarkInstanceUp(instance.name)
3134

    
3135

    
3136
class LUShutdownInstance(LogicalUnit):
3137
  """Shutdown an instance.
3138

3139
  """
3140
  HPATH = "instance-stop"
3141
  HTYPE = constants.HTYPE_INSTANCE
3142
  _OP_REQP = ["instance_name"]
3143
  REQ_BGL = False
3144

    
3145
  def ExpandNames(self):
3146
    self._ExpandAndLockInstance()
3147

    
3148
  def BuildHooksEnv(self):
3149
    """Build hooks env.
3150

3151
    This runs on master, primary and secondary nodes of the instance.
3152

3153
    """
3154
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3155
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3156
    return env, nl, nl
3157

    
3158
  def CheckPrereq(self):
3159
    """Check prerequisites.
3160

3161
    This checks that the instance is in the cluster.
3162

3163
    """
3164
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3165
    assert self.instance is not None, \
3166
      "Cannot retrieve locked instance %s" % self.op.instance_name
3167
    _CheckNodeOnline(self, self.instance.primary_node)
3168

    
3169
  def Exec(self, feedback_fn):
3170
    """Shutdown the instance.
3171

3172
    """
3173
    instance = self.instance
3174
    node_current = instance.primary_node
3175
    self.cfg.MarkInstanceDown(instance.name)
3176
    result = self.rpc.call_instance_shutdown(node_current, instance)
3177
    msg = result.RemoteFailMsg()
3178
    if msg:
3179
      self.proc.LogWarning("Could not shutdown instance: %s" % msg)
3180

    
3181
    _ShutdownInstanceDisks(self, instance)
3182

    
3183

    
3184
class LUReinstallInstance(LogicalUnit):
3185
  """Reinstall an instance.
3186

3187
  """
3188
  HPATH = "instance-reinstall"
3189
  HTYPE = constants.HTYPE_INSTANCE
3190
  _OP_REQP = ["instance_name"]
3191
  REQ_BGL = False
3192

    
3193
  def ExpandNames(self):
3194
    self._ExpandAndLockInstance()
3195

    
3196
  def BuildHooksEnv(self):
3197
    """Build hooks env.
3198

3199
    This runs on master, primary and secondary nodes of the instance.
3200

3201
    """
3202
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3203
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3204
    return env, nl, nl
3205

    
3206
  def CheckPrereq(self):
3207
    """Check prerequisites.
3208

3209
    This checks that the instance is in the cluster and is not running.
3210

3211
    """
3212
    instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3213
    assert instance is not None, \
3214
      "Cannot retrieve locked instance %s" % self.op.instance_name
3215
    _CheckNodeOnline(self, instance.primary_node)
3216

    
3217
    if instance.disk_template == constants.DT_DISKLESS:
3218
      raise errors.OpPrereqError("Instance '%s' has no disks" %
3219
                                 self.op.instance_name)
3220
    if instance.admin_up:
3221
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
3222
                                 self.op.instance_name)
3223
    remote_info = self.rpc.call_instance_info(instance.primary_node,
3224
                                              instance.name,
3225
                                              instance.hypervisor)
3226
    remote_info.Raise()
3227
    if remote_info.data:
3228
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
3229
                                 (self.op.instance_name,
3230
                                  instance.primary_node))
3231

    
3232
    self.op.os_type = getattr(self.op, "os_type", None)
3233
    if self.op.os_type is not None:
3234
      # OS verification
3235
      pnode = self.cfg.GetNodeInfo(
3236
        self.cfg.ExpandNodeName(instance.primary_node))
3237
      if pnode is None:
3238
        raise errors.OpPrereqError("Primary node '%s' is unknown" %
3239
                                   self.op.pnode)
3240
      result = self.rpc.call_os_get(pnode.name, self.op.os_type)
3241
      result.Raise()
3242
      if not isinstance(result.data, objects.OS):
3243
        raise errors.OpPrereqError("OS '%s' not in supported OS list for"
3244
                                   " primary node"  % self.op.os_type)
3245

    
3246
    self.instance = instance
3247

    
3248
  def Exec(self, feedback_fn):
3249
    """Reinstall the instance.
3250

3251
    """
3252
    inst = self.instance
3253

    
3254
    if self.op.os_type is not None:
3255
      feedback_fn("Changing OS to '%s'..." % self.op.os_type)
3256
      inst.os = self.op.os_type
3257
      self.cfg.Update(inst)
3258

    
3259
    _StartInstanceDisks(self, inst, None)
3260
    try:
3261
      feedback_fn("Running the instance OS create scripts...")
3262
      result = self.rpc.call_instance_os_add(inst.primary_node, inst)
3263
      msg = result.RemoteFailMsg()
3264
      if msg:
3265
        raise errors.OpExecError("Could not install OS for instance %s"
3266
                                 " on node %s: %s" %
3267
                                 (inst.name, inst.primary_node, msg))
3268
    finally:
3269
      _ShutdownInstanceDisks(self, inst)
3270

    
3271

    
3272
class LURenameInstance(LogicalUnit):
3273
  """Rename an instance.
3274

3275
  """
3276
  HPATH = "instance-rename"
3277
  HTYPE = constants.HTYPE_INSTANCE
3278
  _OP_REQP = ["instance_name", "new_name"]
3279

    
3280
  def BuildHooksEnv(self):
3281
    """Build hooks env.
3282

3283
    This runs on master, primary and secondary nodes of the instance.
3284

3285
    """
3286
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3287
    env["INSTANCE_NEW_NAME"] = self.op.new_name
3288
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3289
    return env, nl, nl
3290

    
3291
  def CheckPrereq(self):
3292
    """Check prerequisites.
3293

3294
    This checks that the instance is in the cluster and is not running.
3295

3296
    """
3297
    instance = self.cfg.GetInstanceInfo(
3298
      self.cfg.ExpandInstanceName(self.op.instance_name))
3299
    if instance is None:
3300
      raise errors.OpPrereqError("Instance '%s' not known" %
3301
                                 self.op.instance_name)
3302
    _CheckNodeOnline(self, instance.primary_node)
3303

    
3304
    if instance.admin_up:
3305
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
3306
                                 self.op.instance_name)
3307
    remote_info = self.rpc.call_instance_info(instance.primary_node,
3308
                                              instance.name,
3309
                                              instance.hypervisor)
3310
    remote_info.Raise()
3311
    if remote_info.data:
3312
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
3313
                                 (self.op.instance_name,
3314
                                  instance.primary_node))
3315
    self.instance = instance
3316

    
3317
    # new name verification
3318
    name_info = utils.HostInfo(self.op.new_name)
3319

    
3320
    self.op.new_name = new_name = name_info.name
3321
    instance_list = self.cfg.GetInstanceList()
3322
    if new_name in instance_list:
3323
      raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
3324
                                 new_name)
3325

    
3326
    if not getattr(self.op, "ignore_ip", False):
3327
      if utils.TcpPing(name_info.ip, constants.DEFAULT_NODED_PORT):
3328
        raise errors.OpPrereqError("IP %s of instance %s already in use" %
3329
                                   (name_info.ip, new_name))
3330

    
3331

    
3332
  def Exec(self, feedback_fn):
3333
    """Reinstall the instance.
3334

3335
    """
3336
    inst = self.instance
3337
    old_name = inst.name
3338

    
3339
    if inst.disk_template == constants.DT_FILE:
3340
      old_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
3341

    
3342
    self.cfg.RenameInstance(inst.name, self.op.new_name)
3343
    # Change the instance lock. This is definitely safe while we hold the BGL
3344
    self.context.glm.remove(locking.LEVEL_INSTANCE, old_name)
3345
    self.context.glm.add(locking.LEVEL_INSTANCE, self.op.new_name)
3346

    
3347
    # re-read the instance from the configuration after rename
3348
    inst = self.cfg.GetInstanceInfo(self.op.new_name)
3349

    
3350
    if inst.disk_template == constants.DT_FILE:
3351
      new_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
3352
      result = self.rpc.call_file_storage_dir_rename(inst.primary_node,
3353
                                                     old_file_storage_dir,
3354
                                                     new_file_storage_dir)
3355
      result.Raise()
3356
      if not result.data:
3357
        raise errors.OpExecError("Could not connect to node '%s' to rename"
3358
                                 " directory '%s' to '%s' (but the instance"
3359
                                 " has been renamed in Ganeti)" % (
3360
                                 inst.primary_node, old_file_storage_dir,
3361
                                 new_file_storage_dir))
3362

    
3363
      if not result.data[0]:
3364
        raise errors.OpExecError("Could not rename directory '%s' to '%s'"
3365
                                 " (but the instance has been renamed in"
3366
                                 " Ganeti)" % (old_file_storage_dir,
3367
                                               new_file_storage_dir))
3368

    
3369
    _StartInstanceDisks(self, inst, None)
3370
    try:
3371
      result = self.rpc.call_instance_run_rename(inst.primary_node, inst,
3372
                                                 old_name)
3373
      msg = result.RemoteFailMsg()
3374
      if msg:
3375
        msg = ("Could not run OS rename script for instance %s on node %s"
3376
               " (but the instance has been renamed in Ganeti): %s" %
3377
               (inst.name, inst.primary_node, msg))
3378
        self.proc.LogWarning(msg)
3379
    finally:
3380
      _ShutdownInstanceDisks(self, inst)
3381

    
3382

    
3383
class LURemoveInstance(LogicalUnit):
3384
  """Remove an instance.
3385

3386
  """
3387
  HPATH = "instance-remove"
3388
  HTYPE = constants.HTYPE_INSTANCE
3389
  _OP_REQP = ["instance_name", "ignore_failures"]
3390
  REQ_BGL = False
3391

    
3392
  def ExpandNames(self):
3393
    self._ExpandAndLockInstance()
3394
    self.needed_locks[locking.LEVEL_NODE] = []
3395
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3396

    
3397
  def DeclareLocks(self, level):
3398
    if level == locking.LEVEL_NODE:
3399
      self._LockInstancesNodes()
3400

    
3401
  def BuildHooksEnv(self):
3402
    """Build hooks env.
3403

3404
    This runs on master, primary and secondary nodes of the instance.
3405

3406
    """
3407
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3408
    nl = [self.cfg.GetMasterNode()]
3409
    return env, nl, nl
3410

    
3411
  def CheckPrereq(self):
3412
    """Check prerequisites.
3413

3414
    This checks that the instance is in the cluster.
3415

3416
    """
3417
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3418
    assert self.instance is not None, \
3419
      "Cannot retrieve locked instance %s" % self.op.instance_name
3420

    
3421
  def Exec(self, feedback_fn):
3422
    """Remove the instance.
3423

3424
    """
3425
    instance = self.instance
3426
    logging.info("Shutting down instance %s on node %s",
3427
                 instance.name, instance.primary_node)
3428

    
3429
    result = self.rpc.call_instance_shutdown(instance.primary_node, instance)
3430
    msg = result.RemoteFailMsg()
3431
    if msg:
3432
      if self.op.ignore_failures:
3433
        feedback_fn("Warning: can't shutdown instance: %s" % msg)
3434
      else:
3435
        raise errors.OpExecError("Could not shutdown instance %s on"
3436
                                 " node %s: %s" %
3437
                                 (instance.name, instance.primary_node, msg))
3438

    
3439
    logging.info("Removing block devices for instance %s", instance.name)
3440

    
3441
    if not _RemoveDisks(self, instance):
3442
      if self.op.ignore_failures:
3443
        feedback_fn("Warning: can't remove instance's disks")
3444
      else:
3445
        raise errors.OpExecError("Can't remove instance's disks")
3446

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

    
3449
    self.cfg.RemoveInstance(instance.name)
3450
    self.remove_locks[locking.LEVEL_INSTANCE] = instance.name
3451

    
3452

    
3453
class LUQueryInstances(NoHooksLU):
3454
  """Logical unit for querying instances.
3455

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

    
3476

    
3477
  def ExpandNames(self):
3478
    _CheckOutputFields(static=self._FIELDS_STATIC,
3479
                       dynamic=self._FIELDS_DYNAMIC,
3480
                       selected=self.op.output_fields)
3481

    
3482
    self.needed_locks = {}
3483
    self.share_locks[locking.LEVEL_INSTANCE] = 1
3484
    self.share_locks[locking.LEVEL_NODE] = 1
3485

    
3486
    if self.op.names:
3487
      self.wanted = _GetWantedInstances(self, self.op.names)
3488
    else:
3489
      self.wanted = locking.ALL_SET
3490

    
3491
    self.do_node_query = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
3492
    self.do_locking = self.do_node_query and self.op.use_locking
3493
    if self.do_locking:
3494
      self.needed_locks[locking.LEVEL_INSTANCE] = self.wanted
3495
      self.needed_locks[locking.LEVEL_NODE] = []
3496
      self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3497

    
3498
  def DeclareLocks(self, level):
3499
    if level == locking.LEVEL_NODE and self.do_locking:
3500
      self._LockInstancesNodes()
3501

    
3502
  def CheckPrereq(self):
3503
    """Check prerequisites.
3504

3505
    """
3506
    pass
3507

    
3508
  def Exec(self, feedback_fn):
3509
    """Computes the list of nodes and their attributes.
3510

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

    
3532
    instance_list = [all_info[iname] for iname in instance_names]
3533

    
3534
    # begin data gathering
3535

    
3536
    nodes = frozenset([inst.primary_node for inst in instance_list])
3537
    hv_list = list(set([inst.hypervisor for inst in instance_list]))
3538

    
3539
    bad_nodes = []
3540
    off_nodes = []
3541
    if self.do_node_query:
3542
      live_data = {}
3543
      node_data = self.rpc.call_all_instances_info(nodes, hv_list)
3544
      for name in nodes:
3545
        result = node_data[name]
3546
        if result.offline:
3547
          # offline nodes will be in both lists
3548
          off_nodes.append(name)
3549
        if result.failed:
3550
          bad_nodes.append(name)
3551
        else:
3552
          if result.data:
3553
            live_data.update(result.data)
3554
            # else no instance is alive
3555
    else:
3556
      live_data = dict([(name, {}) for name in instance_names])
3557

    
3558
    # end data gathering
3559

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

    
3700
    return output
3701

    
3702

    
3703
class LUFailoverInstance(LogicalUnit):
3704
  """Failover an instance.
3705

3706
  """
3707
  HPATH = "instance-failover"
3708
  HTYPE = constants.HTYPE_INSTANCE
3709
  _OP_REQP = ["instance_name", "ignore_consistency"]
3710
  REQ_BGL = False
3711

    
3712
  def ExpandNames(self):
3713
    self._ExpandAndLockInstance()
3714
    self.needed_locks[locking.LEVEL_NODE] = []
3715
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3716

    
3717
  def DeclareLocks(self, level):
3718
    if level == locking.LEVEL_NODE:
3719
      self._LockInstancesNodes()
3720

    
3721
  def BuildHooksEnv(self):
3722
    """Build hooks env.
3723

3724
    This runs on master, primary and secondary nodes of the instance.
3725

3726
    """
3727
    env = {
3728
      "IGNORE_CONSISTENCY": self.op.ignore_consistency,
3729
      }
3730
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3731
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3732
    return env, nl, nl
3733

    
3734
  def CheckPrereq(self):
3735
    """Check prerequisites.
3736

3737
    This checks that the instance is in the cluster.
3738

3739
    """
3740
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3741
    assert self.instance is not None, \
3742
      "Cannot retrieve locked instance %s" % self.op.instance_name
3743

    
3744
    bep = self.cfg.GetClusterInfo().FillBE(instance)
3745
    if instance.disk_template not in constants.DTS_NET_MIRROR:
3746
      raise errors.OpPrereqError("Instance's disk layout is not"
3747
                                 " network mirrored, cannot failover.")
3748

    
3749
    secondary_nodes = instance.secondary_nodes
3750
    if not secondary_nodes:
3751
      raise errors.ProgrammerError("no secondary node but using "
3752
                                   "a mirrored disk template")
3753

    
3754
    target_node = secondary_nodes[0]
3755
    _CheckNodeOnline(self, target_node)
3756
    _CheckNodeNotDrained(self, target_node)
3757

    
3758
    if instance.admin_up:
3759
      # check memory requirements on the secondary node
3760
      _CheckNodeFreeMemory(self, target_node, "failing over instance %s" %
3761
                           instance.name, bep[constants.BE_MEMORY],
3762
                           instance.hypervisor)
3763
    else:
3764
      self.LogInfo("Not checking memory on the secondary node as"
3765
                   " instance will not be started")
3766

    
3767
    # check bridge existence
3768
    brlist = [nic.bridge for nic in instance.nics]
3769
    result = self.rpc.call_bridges_exist(target_node, brlist)
3770
    result.Raise()
3771
    if not result.data:
3772
      raise errors.OpPrereqError("One or more target bridges %s does not"
3773
                                 " exist on destination node '%s'" %
3774
                                 (brlist, target_node))
3775

    
3776
  def Exec(self, feedback_fn):
3777
    """Failover an instance.
3778

3779
    The failover is done by shutting it down on its present node and
3780
    starting it on the secondary.
3781

3782
    """
3783
    instance = self.instance
3784

    
3785
    source_node = instance.primary_node
3786
    target_node = instance.secondary_nodes[0]
3787

    
3788
    feedback_fn("* checking disk consistency between source and target")
3789
    for dev in instance.disks:
3790
      # for drbd, these are drbd over lvm
3791
      if not _CheckDiskConsistency(self, dev, target_node, False):
3792
        if instance.admin_up and not self.op.ignore_consistency:
3793
          raise errors.OpExecError("Disk %s is degraded on target node,"
3794
                                   " aborting failover." % dev.iv_name)
3795

    
3796
    feedback_fn("* shutting down instance on source node")
3797
    logging.info("Shutting down instance %s on node %s",
3798
                 instance.name, source_node)
3799

    
3800
    result = self.rpc.call_instance_shutdown(source_node, instance)
3801
    msg = result.RemoteFailMsg()
3802
    if msg:
3803
      if self.op.ignore_consistency:
3804
        self.proc.LogWarning("Could not shutdown instance %s on node %s."
3805
                             " Proceeding anyway. Please make sure node"
3806
                             " %s is down. Error details: %s",
3807
                             instance.name, source_node, source_node, msg)
3808
      else:
3809
        raise errors.OpExecError("Could not shutdown instance %s on"
3810
                                 " node %s: %s" %
3811
                                 (instance.name, source_node, msg))
3812

    
3813
    feedback_fn("* deactivating the instance's disks on source node")
3814
    if not _ShutdownInstanceDisks(self, instance, ignore_primary=True):
3815
      raise errors.OpExecError("Can't shut down the instance's disks.")
3816

    
3817
    instance.primary_node = target_node
3818
    # distribute new instance config to the other nodes
3819
    self.cfg.Update(instance)
3820

    
3821
    # Only start the instance if it's marked as up
3822
    if instance.admin_up:
3823
      feedback_fn("* activating the instance's disks on target node")
3824
      logging.info("Starting instance %s on node %s",
3825
                   instance.name, target_node)
3826

    
3827
      disks_ok, _ = _AssembleInstanceDisks(self, instance,
3828
                                               ignore_secondaries=True)
3829
      if not disks_ok:
3830
        _ShutdownInstanceDisks(self, instance)
3831
        raise errors.OpExecError("Can't activate the instance's disks")
3832

    
3833
      feedback_fn("* starting the instance on the target node")
3834
      result = self.rpc.call_instance_start(target_node, instance, None, None)
3835
      msg = result.RemoteFailMsg()
3836
      if msg:
3837
        _ShutdownInstanceDisks(self, instance)
3838
        raise errors.OpExecError("Could not start instance %s on node %s: %s" %
3839
                                 (instance.name, target_node, msg))
3840

    
3841

    
3842
class LUMigrateInstance(LogicalUnit):
3843
  """Migrate an instance.
3844

3845
  This is migration without shutting down, compared to the failover,
3846
  which is done with shutdown.
3847

3848
  """
3849
  HPATH = "instance-migrate"
3850
  HTYPE = constants.HTYPE_INSTANCE
3851
  _OP_REQP = ["instance_name", "live", "cleanup"]
3852

    
3853
  REQ_BGL = False
3854

    
3855
  def ExpandNames(self):
3856
    self._ExpandAndLockInstance()
3857
    self.needed_locks[locking.LEVEL_NODE] = []
3858
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3859

    
3860
  def DeclareLocks(self, level):
3861
    if level == locking.LEVEL_NODE:
3862
      self._LockInstancesNodes()
3863

    
3864
  def BuildHooksEnv(self):
3865
    """Build hooks env.
3866

3867
    This runs on master, primary and secondary nodes of the instance.
3868

3869
    """
3870
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3871
    env["MIGRATE_LIVE"] = self.op.live
3872
    env["MIGRATE_CLEANUP"] = self.op.cleanup
3873
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3874
    return env, nl, nl
3875

    
3876
  def CheckPrereq(self):
3877
    """Check prerequisites.
3878

3879
    This checks that the instance is in the cluster.
3880

3881
    """
3882
    instance = self.cfg.GetInstanceInfo(
3883
      self.cfg.ExpandInstanceName(self.op.instance_name))
3884
    if instance is None:
3885
      raise errors.OpPrereqError("Instance '%s' not known" %
3886
                                 self.op.instance_name)
3887

    
3888
    if instance.disk_template != constants.DT_DRBD8:
3889
      raise errors.OpPrereqError("Instance's disk layout is not"
3890
                                 " drbd8, cannot migrate.")
3891

    
3892
    secondary_nodes = instance.secondary_nodes
3893
    if not secondary_nodes:
3894
      raise errors.ConfigurationError("No secondary node but using"
3895
                                      " drbd8 disk template")
3896

    
3897
    i_be = self.cfg.GetClusterInfo().FillBE(instance)
3898

    
3899
    target_node = secondary_nodes[0]
3900
    # check memory requirements on the secondary node
3901
    _CheckNodeFreeMemory(self, target_node, "migrating instance %s" %
3902
                         instance.name, i_be[constants.BE_MEMORY],
3903
                         instance.hypervisor)
3904

    
3905
    # check bridge existence
3906
    brlist = [nic.bridge for nic in instance.nics]
3907
    result = self.rpc.call_bridges_exist(target_node, brlist)
3908
    if result.failed or not result.data:
3909
      raise errors.OpPrereqError("One or more target bridges %s does not"
3910
                                 " exist on destination node '%s'" %
3911
                                 (brlist, target_node))
3912

    
3913
    if not self.op.cleanup:
3914
      _CheckNodeNotDrained(self, target_node)
3915
      result = self.rpc.call_instance_migratable(instance.primary_node,
3916
                                                 instance)
3917
      msg = result.RemoteFailMsg()
3918
      if msg:
3919
        raise errors.OpPrereqError("Can't migrate: %s - please use failover" %
3920
                                   msg)
3921

    
3922
    self.instance = instance
3923

    
3924
  def _WaitUntilSync(self):
3925
    """Poll with custom rpc for disk sync.
3926

3927
    This uses our own step-based rpc call.
3928

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

    
3952
  def _EnsureSecondary(self, node):
3953
    """Demote a node to secondary.
3954

3955
    """
3956
    self.feedback_fn("* switching node %s to secondary mode" % node)
3957

    
3958
    for dev in self.instance.disks:
3959
      self.cfg.SetDiskID(dev, node)
3960

    
3961
    result = self.rpc.call_blockdev_close(node, self.instance.name,
3962
                                          self.instance.disks)
3963
    msg = result.RemoteFailMsg()
3964
    if msg:
3965
      raise errors.OpExecError("Cannot change disk to secondary on node %s,"
3966
                               " error %s" % (node, msg))
3967

    
3968
  def _GoStandalone(self):
3969
    """Disconnect from the network.
3970

3971
    """
3972
    self.feedback_fn("* changing into standalone mode")
3973
    result = self.rpc.call_drbd_disconnect_net(self.all_nodes, self.nodes_ip,
3974
                                               self.instance.disks)
3975
    for node, nres in result.items():
3976
      msg = nres.RemoteFailMsg()
3977
      if msg:
3978
        raise errors.OpExecError("Cannot disconnect disks node %s,"
3979
                                 " error %s" % (node, msg))
3980

    
3981
  def _GoReconnect(self, multimaster):
3982
    """Reconnect to the network.
3983