Statistics
| Branch: | Tag: | Revision:

root / lib / cmdlib.py @ e631cb25

History | View | Annotate | Download (245.4 kB)

1
#
2
#
3

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

    
21

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

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

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

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

    
47

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

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

60
  Note that all commands require root permissions.
61

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

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

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

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

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

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

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

    
108
  ssh = property(fget=__GetSSH)
109

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

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

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

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

125
    """
126
    pass
127

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

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

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

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

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

149
    Examples::
150

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

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

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

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

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

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

188
    """
189

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

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

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

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

204
    """
205
    raise NotImplementedError
206

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

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

214
    """
215
    raise NotImplementedError
216

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

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

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

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

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

236
    """
237
    raise NotImplementedError
238

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

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

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

257
    """
258
    return lu_result
259

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

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

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

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

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

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

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

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

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

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

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

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

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

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

    
325
    del self.recalculate_locks[locking.LEVEL_NODE]
326

    
327

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

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

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

    
338

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

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

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

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

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

    
365
  return utils.NiceSort(wanted)
366

    
367

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

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

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

    
384
  if instances:
385
    wanted = []
386

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

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

    
397

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

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

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

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

    
416

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

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

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

    
430

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

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

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

    
442

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

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

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

    
454

    
455
def _BuildInstanceHookEnv(name, primary_node, secondary_nodes, os_type, status,
456
                          memory, vcpus, nics, disk_template, disks,
457
                          bep, hvp, hypervisor):
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 distk 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: string
488
  @param hypervisor: 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,
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': 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 brigdes needed by an instance exist.
596

597
  """
598
  # check bridges existance
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 signalled 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 existance 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 accomodate"
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 rone 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
    """Analize 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
    to_act = set()
1297
    for node in nodes:
1298
      # node_volume
1299
      lvs = node_lvs[node]
1300
      if lvs.failed:
1301
        if not lvs.offline:
1302
          self.LogWarning("Connection to node %s failed: %s" %
1303
                          (node, lvs.data))
1304
        continue
1305
      lvs = lvs.data
1306
      if isinstance(lvs, basestring):
1307
        logging.warning("Error enumerating LVs on node %s: %s", node, lvs)
1308
        res_nlvm[node] = lvs
1309
        continue
1310
      elif not isinstance(lvs, dict):
1311
        logging.warning("Connection to node %s failed or invalid data"
1312
                        " returned", node)
1313
        res_nodes.append(node)
1314
        continue
1315

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

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

    
1329
    return result
1330

    
1331

    
1332
class LURenameCluster(LogicalUnit):
1333
  """Rename the cluster.
1334

1335
  """
1336
  HPATH = "cluster-rename"
1337
  HTYPE = constants.HTYPE_CLUSTER
1338
  _OP_REQP = ["name"]
1339

    
1340
  def BuildHooksEnv(self):
1341
    """Build hooks env.
1342

1343
    """
1344
    env = {
1345
      "OP_TARGET": self.cfg.GetClusterName(),
1346
      "NEW_NAME": self.op.name,
1347
      }
1348
    mn = self.cfg.GetMasterNode()
1349
    return env, [mn], [mn]
1350

    
1351
  def CheckPrereq(self):
1352
    """Verify that the passed name is a valid one.
1353

1354
    """
1355
    hostname = utils.HostInfo(self.op.name)
1356

    
1357
    new_name = hostname.name
1358
    self.ip = new_ip = hostname.ip
1359
    old_name = self.cfg.GetClusterName()
1360
    old_ip = self.cfg.GetMasterIP()
1361
    if new_name == old_name and new_ip == old_ip:
1362
      raise errors.OpPrereqError("Neither the name nor the IP address of the"
1363
                                 " cluster has changed")
1364
    if new_ip != old_ip:
1365
      if utils.TcpPing(new_ip, constants.DEFAULT_NODED_PORT):
1366
        raise errors.OpPrereqError("The given cluster IP address (%s) is"
1367
                                   " reachable on the network. Aborting." %
1368
                                   new_ip)
1369

    
1370
    self.op.name = new_name
1371

    
1372
  def Exec(self, feedback_fn):
1373
    """Rename the cluster.
1374

1375
    """
1376
    clustername = self.op.name
1377
    ip = self.ip
1378

    
1379
    # shutdown the master IP
1380
    master = self.cfg.GetMasterNode()
1381
    result = self.rpc.call_node_stop_master(master, False)
1382
    if result.failed or not result.data:
1383
      raise errors.OpExecError("Could not disable the master role")
1384

    
1385
    try:
1386
      cluster = self.cfg.GetClusterInfo()
1387
      cluster.cluster_name = clustername
1388
      cluster.master_ip = ip
1389
      self.cfg.Update(cluster)
1390

    
1391
      # update the known hosts file
1392
      ssh.WriteKnownHostsFile(self.cfg, constants.SSH_KNOWN_HOSTS_FILE)
1393
      node_list = self.cfg.GetNodeList()
1394
      try:
1395
        node_list.remove(master)
1396
      except ValueError:
1397
        pass
1398
      result = self.rpc.call_upload_file(node_list,
1399
                                         constants.SSH_KNOWN_HOSTS_FILE)
1400
      for to_node, to_result in result.iteritems():
1401
        if to_result.failed or not to_result.data:
1402
          logging.error("Copy of file %s to node %s failed",
1403
                        constants.SSH_KNOWN_HOSTS_FILE, to_node)
1404

    
1405
    finally:
1406
      result = self.rpc.call_node_start_master(master, False)
1407
      if result.failed or not result.data:
1408
        self.LogWarning("Could not re-enable the master role on"
1409
                        " the master, please restart manually.")
1410

    
1411

    
1412
def _RecursiveCheckIfLVMBased(disk):
1413
  """Check if the given disk or its children are lvm-based.
1414

1415
  @type disk: L{objects.Disk}
1416
  @param disk: the disk to check
1417
  @rtype: booleean
1418
  @return: boolean indicating whether a LD_LV dev_type was found or not
1419

1420
  """
1421
  if disk.children:
1422
    for chdisk in disk.children:
1423
      if _RecursiveCheckIfLVMBased(chdisk):
1424
        return True
1425
  return disk.dev_type == constants.LD_LV
1426

    
1427

    
1428
class LUSetClusterParams(LogicalUnit):
1429
  """Change the parameters of the cluster.
1430

1431
  """
1432
  HPATH = "cluster-modify"
1433
  HTYPE = constants.HTYPE_CLUSTER
1434
  _OP_REQP = []
1435
  REQ_BGL = False
1436

    
1437
  def CheckArguments(self):
1438
    """Check parameters
1439

1440
    """
1441
    if not hasattr(self.op, "candidate_pool_size"):
1442
      self.op.candidate_pool_size = None
1443
    if self.op.candidate_pool_size is not None:
1444
      try:
1445
        self.op.candidate_pool_size = int(self.op.candidate_pool_size)
1446
      except (ValueError, TypeError), err:
1447
        raise errors.OpPrereqError("Invalid candidate_pool_size value: %s" %
1448
                                   str(err))
1449
      if self.op.candidate_pool_size < 1:
1450
        raise errors.OpPrereqError("At least one master candidate needed")
1451

    
1452
  def ExpandNames(self):
1453
    # FIXME: in the future maybe other cluster params won't require checking on
1454
    # all nodes to be modified.
1455
    self.needed_locks = {
1456
      locking.LEVEL_NODE: locking.ALL_SET,
1457
    }
1458
    self.share_locks[locking.LEVEL_NODE] = 1
1459

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

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

    
1471
  def CheckPrereq(self):
1472
    """Check prerequisites.
1473

1474
    This checks whether the given params don't conflict and
1475
    if the given volume group is valid.
1476

1477
    """
1478
    if self.op.vg_name is not None and not self.op.vg_name:
1479
      instances = self.cfg.GetAllInstancesInfo().values()
1480
      for inst in instances:
1481
        for disk in inst.disks:
1482
          if _RecursiveCheckIfLVMBased(disk):
1483
            raise errors.OpPrereqError("Cannot disable lvm storage while"
1484
                                       " lvm-based instances exist")
1485

    
1486
    node_list = self.acquired_locks[locking.LEVEL_NODE]
1487

    
1488
    # if vg_name not None, checks given volume group on all nodes
1489
    if self.op.vg_name:
1490
      vglist = self.rpc.call_vg_list(node_list)
1491
      for node in node_list:
1492
        if vglist[node].failed:
1493
          # ignoring down node
1494
          self.LogWarning("Node %s unreachable/error, ignoring" % node)
1495
          continue
1496
        vgstatus = utils.CheckVolumeGroupSize(vglist[node].data,
1497
                                              self.op.vg_name,
1498
                                              constants.MIN_VG_SIZE)
1499
        if vgstatus:
1500
          raise errors.OpPrereqError("Error on node '%s': %s" %
1501
                                     (node, vgstatus))
1502

    
1503
    self.cluster = cluster = self.cfg.GetClusterInfo()
1504
    # validate beparams changes
1505
    if self.op.beparams:
1506
      utils.ForceDictType(self.op.beparams, constants.BES_PARAMETER_TYPES)
1507
      self.new_beparams = cluster.FillDict(
1508
        cluster.beparams[constants.BEGR_DEFAULT], self.op.beparams)
1509

    
1510
    # hypervisor list/parameters
1511
    self.new_hvparams = cluster.FillDict(cluster.hvparams, {})
1512
    if self.op.hvparams:
1513
      if not isinstance(self.op.hvparams, dict):
1514
        raise errors.OpPrereqError("Invalid 'hvparams' parameter on input")
1515
      for hv_name, hv_dict in self.op.hvparams.items():
1516
        if hv_name not in self.new_hvparams:
1517
          self.new_hvparams[hv_name] = hv_dict
1518
        else:
1519
          self.new_hvparams[hv_name].update(hv_dict)
1520

    
1521
    if self.op.enabled_hypervisors is not None:
1522
      self.hv_list = self.op.enabled_hypervisors
1523
    else:
1524
      self.hv_list = cluster.enabled_hypervisors
1525

    
1526
    if self.op.hvparams or self.op.enabled_hypervisors is not None:
1527
      # either the enabled list has changed, or the parameters have, validate
1528
      for hv_name, hv_params in self.new_hvparams.items():
1529
        if ((self.op.hvparams and hv_name in self.op.hvparams) or
1530
            (self.op.enabled_hypervisors and
1531
             hv_name in self.op.enabled_hypervisors)):
1532
          # either this is a new hypervisor, or its parameters have changed
1533
          hv_class = hypervisor.GetHypervisor(hv_name)
1534
          utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
1535
          hv_class.CheckParameterSyntax(hv_params)
1536
          _CheckHVParams(self, node_list, hv_name, hv_params)
1537

    
1538
  def Exec(self, feedback_fn):
1539
    """Change the parameters of the cluster.
1540

1541
    """
1542
    if self.op.vg_name is not None:
1543
      new_volume = self.op.vg_name
1544
      if not new_volume:
1545
        new_volume = None
1546
      if new_volume != self.cfg.GetVGName():
1547
        self.cfg.SetVGName(new_volume)
1548
      else:
1549
        feedback_fn("Cluster LVM configuration already in desired"
1550
                    " state, not changing")
1551
    if self.op.hvparams:
1552
      self.cluster.hvparams = self.new_hvparams
1553
    if self.op.enabled_hypervisors is not None:
1554
      self.cluster.enabled_hypervisors = self.op.enabled_hypervisors
1555
    if self.op.beparams:
1556
      self.cluster.beparams[constants.BEGR_DEFAULT] = self.new_beparams
1557
    if self.op.candidate_pool_size is not None:
1558
      self.cluster.candidate_pool_size = self.op.candidate_pool_size
1559
      # we need to update the pool size here, otherwise the save will fail
1560
      _AdjustCandidatePool(self)
1561

    
1562
    self.cfg.Update(self.cluster)
1563

    
1564

    
1565
class LURedistributeConfig(NoHooksLU):
1566
  """Force the redistribution of cluster configuration.
1567

1568
  This is a very simple LU.
1569

1570
  """
1571
  _OP_REQP = []
1572
  REQ_BGL = False
1573

    
1574
  def ExpandNames(self):
1575
    self.needed_locks = {
1576
      locking.LEVEL_NODE: locking.ALL_SET,
1577
    }
1578
    self.share_locks[locking.LEVEL_NODE] = 1
1579

    
1580
  def CheckPrereq(self):
1581
    """Check prerequisites.
1582

1583
    """
1584

    
1585
  def Exec(self, feedback_fn):
1586
    """Redistribute the configuration.
1587

1588
    """
1589
    self.cfg.Update(self.cfg.GetClusterInfo())
1590

    
1591

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

1595
  """
1596
  if not instance.disks:
1597
    return True
1598

    
1599
  if not oneshot:
1600
    lu.proc.LogInfo("Waiting for instance %s to sync disks." % instance.name)
1601

    
1602
  node = instance.primary_node
1603

    
1604
  for dev in instance.disks:
1605
    lu.cfg.SetDiskID(dev, node)
1606

    
1607
  retries = 0
1608
  degr_retries = 10 # in seconds, as we sleep 1 second each time
1609
  while True:
1610
    max_time = 0
1611
    done = True
1612
    cumul_degraded = False
1613
    rstats = lu.rpc.call_blockdev_getmirrorstatus(node, instance.disks)
1614
    if rstats.failed or not rstats.data:
1615
      lu.LogWarning("Can't get any data from node %s", node)
1616
      retries += 1
1617
      if retries >= 10:
1618
        raise errors.RemoteError("Can't contact node %s for mirror data,"
1619
                                 " aborting." % node)
1620
      time.sleep(6)
1621
      continue
1622
    rstats = rstats.data
1623
    retries = 0
1624
    for i, mstat in enumerate(rstats):
1625
      if mstat is None:
1626
        lu.LogWarning("Can't compute data for node %s/%s",
1627
                           node, instance.disks[i].iv_name)
1628
        continue
1629
      # we ignore the ldisk parameter
1630
      perc_done, est_time, is_degraded, _ = mstat
1631
      cumul_degraded = cumul_degraded or (is_degraded and perc_done is None)
1632
      if perc_done is not None:
1633
        done = False
1634
        if est_time is not None:
1635
          rem_time = "%d estimated seconds remaining" % est_time
1636
          max_time = est_time
1637
        else:
1638
          rem_time = "no time estimate"
1639
        lu.proc.LogInfo("- device %s: %5.2f%% done, %s" %
1640
                        (instance.disks[i].iv_name, perc_done, rem_time))
1641

    
1642
    # if we're done but degraded, let's do a few small retries, to
1643
    # make sure we see a stable and not transient situation; therefore
1644
    # we force restart of the loop
1645
    if (done or oneshot) and cumul_degraded and degr_retries > 0:
1646
      logging.info("Degraded disks found, %d retries left", degr_retries)
1647
      degr_retries -= 1
1648
      time.sleep(1)
1649
      continue
1650

    
1651
    if done or oneshot:
1652
      break
1653

    
1654
    time.sleep(min(60, max_time))
1655

    
1656
  if done:
1657
    lu.proc.LogInfo("Instance %s's disks are in sync." % instance.name)
1658
  return not cumul_degraded
1659

    
1660

    
1661
def _CheckDiskConsistency(lu, dev, node, on_primary, ldisk=False):
1662
  """Check that mirrors are not degraded.
1663

1664
  The ldisk parameter, if True, will change the test from the
1665
  is_degraded attribute (which represents overall non-ok status for
1666
  the device(s)) to the ldisk (representing the local storage status).
1667

1668
  """
1669
  lu.cfg.SetDiskID(dev, node)
1670
  if ldisk:
1671
    idx = 6
1672
  else:
1673
    idx = 5
1674

    
1675
  result = True
1676
  if on_primary or dev.AssembleOnSecondary():
1677
    rstats = lu.rpc.call_blockdev_find(node, dev)
1678
    msg = rstats.RemoteFailMsg()
1679
    if msg:
1680
      lu.LogWarning("Can't find disk on node %s: %s", node, msg)
1681
      result = False
1682
    elif not rstats.payload:
1683
      lu.LogWarning("Can't find disk on node %s", node)
1684
      result = False
1685
    else:
1686
      result = result and (not rstats.payload[idx])
1687
  if dev.children:
1688
    for child in dev.children:
1689
      result = result and _CheckDiskConsistency(lu, child, node, on_primary)
1690

    
1691
  return result
1692

    
1693

    
1694
class LUDiagnoseOS(NoHooksLU):
1695
  """Logical unit for OS diagnose/query.
1696

1697
  """
1698
  _OP_REQP = ["output_fields", "names"]
1699
  REQ_BGL = False
1700
  _FIELDS_STATIC = utils.FieldSet()
1701
  _FIELDS_DYNAMIC = utils.FieldSet("name", "valid", "node_status")
1702

    
1703
  def ExpandNames(self):
1704
    if self.op.names:
1705
      raise errors.OpPrereqError("Selective OS query not supported")
1706

    
1707
    _CheckOutputFields(static=self._FIELDS_STATIC,
1708
                       dynamic=self._FIELDS_DYNAMIC,
1709
                       selected=self.op.output_fields)
1710

    
1711
    # Lock all nodes, in shared mode
1712
    # Temporary removal of locks, should be reverted later
1713
    # TODO: reintroduce locks when they are lighter-weight
1714
    self.needed_locks = {}
1715
    #self.share_locks[locking.LEVEL_NODE] = 1
1716
    #self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
1717

    
1718
  def CheckPrereq(self):
1719
    """Check prerequisites.
1720

1721
    """
1722

    
1723
  @staticmethod
1724
  def _DiagnoseByOS(node_list, rlist):
1725
    """Remaps a per-node return list into an a per-os per-node dictionary
1726

1727
    @param node_list: a list with the names of all nodes
1728
    @param rlist: a map with node names as keys and OS objects as values
1729

1730
    @rtype: dict
1731
    @return: a dictionary with osnames as keys and as value another map, with
1732
        nodes as keys and list of OS objects as values, eg::
1733

1734
          {"debian-etch": {"node1": [<object>,...],
1735
                           "node2": [<object>,]}
1736
          }
1737

1738
    """
1739
    all_os = {}
1740
    # we build here the list of nodes that didn't fail the RPC (at RPC
1741
    # level), so that nodes with a non-responding node daemon don't
1742
    # make all OSes invalid
1743
    good_nodes = [node_name for node_name in rlist
1744
                  if not rlist[node_name].failed]
1745
    for node_name, nr in rlist.iteritems():
1746
      if nr.failed or not nr.data:
1747
        continue
1748
      for os_obj in nr.data:
1749
        if os_obj.name not in all_os:
1750
          # build a list of nodes for this os containing empty lists
1751
          # for each node in node_list
1752
          all_os[os_obj.name] = {}
1753
          for nname in good_nodes:
1754
            all_os[os_obj.name][nname] = []
1755
        all_os[os_obj.name][node_name].append(os_obj)
1756
    return all_os
1757

    
1758
  def Exec(self, feedback_fn):
1759
    """Compute the list of OSes.
1760

1761
    """
1762
    valid_nodes = [node for node in self.cfg.GetOnlineNodeList()]
1763
    node_data = self.rpc.call_os_diagnose(valid_nodes)
1764
    if node_data == False:
1765
      raise errors.OpExecError("Can't gather the list of OSes")
1766
    pol = self._DiagnoseByOS(valid_nodes, node_data)
1767
    output = []
1768
    for os_name, os_data in pol.iteritems():
1769
      row = []
1770
      for field in self.op.output_fields:
1771
        if field == "name":
1772
          val = os_name
1773
        elif field == "valid":
1774
          val = utils.all([osl and osl[0] for osl in os_data.values()])
1775
        elif field == "node_status":
1776
          val = {}
1777
          for node_name, nos_list in os_data.iteritems():
1778
            val[node_name] = [(v.status, v.path) for v in nos_list]
1779
        else:
1780
          raise errors.ParameterError(field)
1781
        row.append(val)
1782
      output.append(row)
1783

    
1784
    return output
1785

    
1786

    
1787
class LURemoveNode(LogicalUnit):
1788
  """Logical unit for removing a node.
1789

1790
  """
1791
  HPATH = "node-remove"
1792
  HTYPE = constants.HTYPE_NODE
1793
  _OP_REQP = ["node_name"]
1794

    
1795
  def BuildHooksEnv(self):
1796
    """Build hooks env.
1797

1798
    This doesn't run on the target node in the pre phase as a failed
1799
    node would then be impossible to remove.
1800

1801
    """
1802
    env = {
1803
      "OP_TARGET": self.op.node_name,
1804
      "NODE_NAME": self.op.node_name,
1805
      }
1806
    all_nodes = self.cfg.GetNodeList()
1807
    all_nodes.remove(self.op.node_name)
1808
    return env, all_nodes, all_nodes
1809

    
1810
  def CheckPrereq(self):
1811
    """Check prerequisites.
1812

1813
    This checks:
1814
     - the node exists in the configuration
1815
     - it does not have primary or secondary instances
1816
     - it's not the master
1817

1818
    Any errors are signalled by raising errors.OpPrereqError.
1819

1820
    """
1821
    node = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.node_name))
1822
    if node is None:
1823
      raise errors.OpPrereqError, ("Node '%s' is unknown." % self.op.node_name)
1824

    
1825
    instance_list = self.cfg.GetInstanceList()
1826

    
1827
    masternode = self.cfg.GetMasterNode()
1828
    if node.name == masternode:
1829
      raise errors.OpPrereqError("Node is the master node,"
1830
                                 " you need to failover first.")
1831

    
1832
    for instance_name in instance_list:
1833
      instance = self.cfg.GetInstanceInfo(instance_name)
1834
      if node.name in instance.all_nodes:
1835
        raise errors.OpPrereqError("Instance %s is still running on the node,"
1836
                                   " please remove first." % instance_name)
1837
    self.op.node_name = node.name
1838
    self.node = node
1839

    
1840
  def Exec(self, feedback_fn):
1841
    """Removes the node from the cluster.
1842

1843
    """
1844
    node = self.node
1845
    logging.info("Stopping the node daemon and removing configs from node %s",
1846
                 node.name)
1847

    
1848
    self.context.RemoveNode(node.name)
1849

    
1850
    self.rpc.call_node_leave_cluster(node.name)
1851

    
1852
    # Promote nodes to master candidate as needed
1853
    _AdjustCandidatePool(self)
1854

    
1855

    
1856
class LUQueryNodes(NoHooksLU):
1857
  """Logical unit for querying nodes.
1858

1859
  """
1860
  _OP_REQP = ["output_fields", "names", "use_locking"]
1861
  REQ_BGL = False
1862
  _FIELDS_DYNAMIC = utils.FieldSet(
1863
    "dtotal", "dfree",
1864
    "mtotal", "mnode", "mfree",
1865
    "bootid",
1866
    "ctotal", "cnodes", "csockets",
1867
    )
1868

    
1869
  _FIELDS_STATIC = utils.FieldSet(
1870
    "name", "pinst_cnt", "sinst_cnt",
1871
    "pinst_list", "sinst_list",
1872
    "pip", "sip", "tags",
1873
    "serial_no",
1874
    "master_candidate",
1875
    "master",
1876
    "offline",
1877
    "drained",
1878
    "role",
1879
    )
1880

    
1881
  def ExpandNames(self):
1882
    _CheckOutputFields(static=self._FIELDS_STATIC,
1883
                       dynamic=self._FIELDS_DYNAMIC,
1884
                       selected=self.op.output_fields)
1885

    
1886
    self.needed_locks = {}
1887
    self.share_locks[locking.LEVEL_NODE] = 1
1888

    
1889
    if self.op.names:
1890
      self.wanted = _GetWantedNodes(self, self.op.names)
1891
    else:
1892
      self.wanted = locking.ALL_SET
1893

    
1894
    self.do_node_query = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
1895
    self.do_locking = self.do_node_query and self.op.use_locking
1896
    if self.do_locking:
1897
      # if we don't request only static fields, we need to lock the nodes
1898
      self.needed_locks[locking.LEVEL_NODE] = self.wanted
1899

    
1900

    
1901
  def CheckPrereq(self):
1902
    """Check prerequisites.
1903

1904
    """
1905
    # The validation of the node list is done in the _GetWantedNodes,
1906
    # if non empty, and if empty, there's no validation to do
1907
    pass
1908

    
1909
  def Exec(self, feedback_fn):
1910
    """Computes the list of nodes and their attributes.
1911

1912
    """
1913
    all_info = self.cfg.GetAllNodesInfo()
1914
    if self.do_locking:
1915
      nodenames = self.acquired_locks[locking.LEVEL_NODE]
1916
    elif self.wanted != locking.ALL_SET:
1917
      nodenames = self.wanted
1918
      missing = set(nodenames).difference(all_info.keys())
1919
      if missing:
1920
        raise errors.OpExecError(
1921
          "Some nodes were removed before retrieving their data: %s" % missing)
1922
    else:
1923
      nodenames = all_info.keys()
1924

    
1925
    nodenames = utils.NiceSort(nodenames)
1926
    nodelist = [all_info[name] for name in nodenames]
1927

    
1928
    # begin data gathering
1929

    
1930
    if self.do_node_query:
1931
      live_data = {}
1932
      node_data = self.rpc.call_node_info(nodenames, self.cfg.GetVGName(),
1933
                                          self.cfg.GetHypervisorType())
1934
      for name in nodenames:
1935
        nodeinfo = node_data[name]
1936
        if not nodeinfo.failed and nodeinfo.data:
1937
          nodeinfo = nodeinfo.data
1938
          fn = utils.TryConvert
1939
          live_data[name] = {
1940
            "mtotal": fn(int, nodeinfo.get('memory_total', None)),
1941
            "mnode": fn(int, nodeinfo.get('memory_dom0', None)),
1942
            "mfree": fn(int, nodeinfo.get('memory_free', None)),
1943
            "dtotal": fn(int, nodeinfo.get('vg_size', None)),
1944
            "dfree": fn(int, nodeinfo.get('vg_free', None)),
1945
            "ctotal": fn(int, nodeinfo.get('cpu_total', None)),
1946
            "bootid": nodeinfo.get('bootid', None),
1947
            "cnodes": fn(int, nodeinfo.get('cpu_nodes', None)),
1948
            "csockets": fn(int, nodeinfo.get('cpu_sockets', None)),
1949
            }
1950
        else:
1951
          live_data[name] = {}
1952
    else:
1953
      live_data = dict.fromkeys(nodenames, {})
1954

    
1955
    node_to_primary = dict([(name, set()) for name in nodenames])
1956
    node_to_secondary = dict([(name, set()) for name in nodenames])
1957

    
1958
    inst_fields = frozenset(("pinst_cnt", "pinst_list",
1959
                             "sinst_cnt", "sinst_list"))
1960
    if inst_fields & frozenset(self.op.output_fields):
1961
      instancelist = self.cfg.GetInstanceList()
1962

    
1963
      for instance_name in instancelist:
1964
        inst = self.cfg.GetInstanceInfo(instance_name)
1965
        if inst.primary_node in node_to_primary:
1966
          node_to_primary[inst.primary_node].add(inst.name)
1967
        for secnode in inst.secondary_nodes:
1968
          if secnode in node_to_secondary:
1969
            node_to_secondary[secnode].add(inst.name)
1970

    
1971
    master_node = self.cfg.GetMasterNode()
1972

    
1973
    # end data gathering
1974

    
1975
    output = []
1976
    for node in nodelist:
1977
      node_output = []
1978
      for field in self.op.output_fields:
1979
        if field == "name":
1980
          val = node.name
1981
        elif field == "pinst_list":
1982
          val = list(node_to_primary[node.name])
1983
        elif field == "sinst_list":
1984
          val = list(node_to_secondary[node.name])
1985
        elif field == "pinst_cnt":
1986
          val = len(node_to_primary[node.name])
1987
        elif field == "sinst_cnt":
1988
          val = len(node_to_secondary[node.name])
1989
        elif field == "pip":
1990
          val = node.primary_ip
1991
        elif field == "sip":
1992
          val = node.secondary_ip
1993
        elif field == "tags":
1994
          val = list(node.GetTags())
1995
        elif field == "serial_no":
1996
          val = node.serial_no
1997
        elif field == "master_candidate":
1998
          val = node.master_candidate
1999
        elif field == "master":
2000
          val = node.name == master_node
2001
        elif field == "offline":
2002
          val = node.offline
2003
        elif field == "drained":
2004
          val = node.drained
2005
        elif self._FIELDS_DYNAMIC.Matches(field):
2006
          val = live_data[node.name].get(field, None)
2007
        elif field == "role":
2008
          if node.name == master_node:
2009
            val = "M"
2010
          elif node.master_candidate:
2011
            val = "C"
2012
          elif node.drained:
2013
            val = "D"
2014
          elif node.offline:
2015
            val = "O"
2016
          else:
2017
            val = "R"
2018
        else:
2019
          raise errors.ParameterError(field)
2020
        node_output.append(val)
2021
      output.append(node_output)
2022

    
2023
    return output
2024

    
2025

    
2026
class LUQueryNodeVolumes(NoHooksLU):
2027
  """Logical unit for getting volumes on node(s).
2028

2029
  """
2030
  _OP_REQP = ["nodes", "output_fields"]
2031
  REQ_BGL = False
2032
  _FIELDS_DYNAMIC = utils.FieldSet("phys", "vg", "name", "size", "instance")
2033
  _FIELDS_STATIC = utils.FieldSet("node")
2034

    
2035
  def ExpandNames(self):
2036
    _CheckOutputFields(static=self._FIELDS_STATIC,
2037
                       dynamic=self._FIELDS_DYNAMIC,
2038
                       selected=self.op.output_fields)
2039

    
2040
    self.needed_locks = {}
2041
    self.share_locks[locking.LEVEL_NODE] = 1
2042
    if not self.op.nodes:
2043
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
2044
    else:
2045
      self.needed_locks[locking.LEVEL_NODE] = \
2046
        _GetWantedNodes(self, self.op.nodes)
2047

    
2048
  def CheckPrereq(self):
2049
    """Check prerequisites.
2050

2051
    This checks that the fields required are valid output fields.
2052

2053
    """
2054
    self.nodes = self.acquired_locks[locking.LEVEL_NODE]
2055

    
2056
  def Exec(self, feedback_fn):
2057
    """Computes the list of nodes and their attributes.
2058

2059
    """
2060
    nodenames = self.nodes
2061
    volumes = self.rpc.call_node_volumes(nodenames)
2062

    
2063
    ilist = [self.cfg.GetInstanceInfo(iname) for iname
2064
             in self.cfg.GetInstanceList()]
2065

    
2066
    lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
2067

    
2068
    output = []
2069
    for node in nodenames:
2070
      if node not in volumes or volumes[node].failed or not volumes[node].data:
2071
        continue
2072

    
2073
      node_vols = volumes[node].data[:]
2074
      node_vols.sort(key=lambda vol: vol['dev'])
2075

    
2076
      for vol in node_vols:
2077
        node_output = []
2078
        for field in self.op.output_fields:
2079
          if field == "node":
2080
            val = node
2081
          elif field == "phys":
2082
            val = vol['dev']
2083
          elif field == "vg":
2084
            val = vol['vg']
2085
          elif field == "name":
2086
            val = vol['name']
2087
          elif field == "size":
2088
            val = int(float(vol['size']))
2089
          elif field == "instance":
2090
            for inst in ilist:
2091
              if node not in lv_by_node[inst]:
2092
                continue
2093
              if vol['name'] in lv_by_node[inst][node]:
2094
                val = inst.name
2095
                break
2096
            else:
2097
              val = '-'
2098
          else:
2099
            raise errors.ParameterError(field)
2100
          node_output.append(str(val))
2101

    
2102
        output.append(node_output)
2103

    
2104
    return output
2105

    
2106

    
2107
class LUAddNode(LogicalUnit):
2108
  """Logical unit for adding node to the cluster.
2109

2110
  """
2111
  HPATH = "node-add"
2112
  HTYPE = constants.HTYPE_NODE
2113
  _OP_REQP = ["node_name"]
2114

    
2115
  def BuildHooksEnv(self):
2116
    """Build hooks env.
2117

2118
    This will run on all nodes before, and on all nodes + the new node after.
2119

2120
    """
2121
    env = {
2122
      "OP_TARGET": self.op.node_name,
2123
      "NODE_NAME": self.op.node_name,
2124
      "NODE_PIP": self.op.primary_ip,
2125
      "NODE_SIP": self.op.secondary_ip,
2126
      }
2127
    nodes_0 = self.cfg.GetNodeList()
2128
    nodes_1 = nodes_0 + [self.op.node_name, ]
2129
    return env, nodes_0, nodes_1
2130

    
2131
  def CheckPrereq(self):
2132
    """Check prerequisites.
2133

2134
    This checks:
2135
     - the new node is not already in the config
2136
     - it is resolvable
2137
     - its parameters (single/dual homed) matches the cluster
2138

2139
    Any errors are signalled by raising errors.OpPrereqError.
2140

2141
    """
2142
    node_name = self.op.node_name
2143
    cfg = self.cfg
2144

    
2145
    dns_data = utils.HostInfo(node_name)
2146

    
2147
    node = dns_data.name
2148
    primary_ip = self.op.primary_ip = dns_data.ip
2149
    secondary_ip = getattr(self.op, "secondary_ip", None)
2150
    if secondary_ip is None:
2151
      secondary_ip = primary_ip
2152
    if not utils.IsValidIP(secondary_ip):
2153
      raise errors.OpPrereqError("Invalid secondary IP given")
2154
    self.op.secondary_ip = secondary_ip
2155

    
2156
    node_list = cfg.GetNodeList()
2157
    if not self.op.readd and node in node_list:
2158
      raise errors.OpPrereqError("Node %s is already in the configuration" %
2159
                                 node)
2160
    elif self.op.readd and node not in node_list:
2161
      raise errors.OpPrereqError("Node %s is not in the configuration" % node)
2162

    
2163
    for existing_node_name in node_list:
2164
      existing_node = cfg.GetNodeInfo(existing_node_name)
2165

    
2166
      if self.op.readd and node == existing_node_name:
2167
        if (existing_node.primary_ip != primary_ip or
2168
            existing_node.secondary_ip != secondary_ip):
2169
          raise errors.OpPrereqError("Readded node doesn't have the same IP"
2170
                                     " address configuration as before")
2171
        continue
2172

    
2173
      if (existing_node.primary_ip == primary_ip or
2174
          existing_node.secondary_ip == primary_ip or
2175
          existing_node.primary_ip == secondary_ip or
2176
          existing_node.secondary_ip == secondary_ip):
2177
        raise errors.OpPrereqError("New node ip address(es) conflict with"
2178
                                   " existing node %s" % existing_node.name)
2179

    
2180
    # check that the type of the node (single versus dual homed) is the
2181
    # same as for the master
2182
    myself = cfg.GetNodeInfo(self.cfg.GetMasterNode())
2183
    master_singlehomed = myself.secondary_ip == myself.primary_ip
2184
    newbie_singlehomed = secondary_ip == primary_ip
2185
    if master_singlehomed != newbie_singlehomed:
2186
      if master_singlehomed:
2187
        raise errors.OpPrereqError("The master has no private ip but the"
2188
                                   " new node has one")
2189
      else:
2190
        raise errors.OpPrereqError("The master has a private ip but the"
2191
                                   " new node doesn't have one")
2192

    
2193
    # checks reachablity
2194
    if not utils.TcpPing(primary_ip, constants.DEFAULT_NODED_PORT):
2195
      raise errors.OpPrereqError("Node not reachable by ping")
2196

    
2197
    if not newbie_singlehomed:
2198
      # check reachability from my secondary ip to newbie's secondary ip
2199
      if not utils.TcpPing(secondary_ip, constants.DEFAULT_NODED_PORT,
2200
                           source=myself.secondary_ip):
2201
        raise errors.OpPrereqError("Node secondary ip not reachable by TCP"
2202
                                   " based ping to noded port")
2203

    
2204
    cp_size = self.cfg.GetClusterInfo().candidate_pool_size
2205
    mc_now, _ = self.cfg.GetMasterCandidateStats()
2206
    master_candidate = mc_now < cp_size
2207

    
2208
    self.new_node = objects.Node(name=node,
2209
                                 primary_ip=primary_ip,
2210
                                 secondary_ip=secondary_ip,
2211
                                 master_candidate=master_candidate,
2212
                                 offline=False, drained=False)
2213

    
2214
  def Exec(self, feedback_fn):
2215
    """Adds the new node to the cluster.
2216

2217
    """
2218
    new_node = self.new_node
2219
    node = new_node.name
2220

    
2221
    # check connectivity
2222
    result = self.rpc.call_version([node])[node]
2223
    result.Raise()
2224
    if result.data:
2225
      if constants.PROTOCOL_VERSION == result.data:
2226
        logging.info("Communication to node %s fine, sw version %s match",
2227
                     node, result.data)
2228
      else:
2229
        raise errors.OpExecError("Version mismatch master version %s,"
2230
                                 " node version %s" %
2231
                                 (constants.PROTOCOL_VERSION, result.data))
2232
    else:
2233
      raise errors.OpExecError("Cannot get version from the new node")
2234

    
2235
    # setup ssh on node
2236
    logging.info("Copy ssh key to node %s", node)
2237
    priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
2238
    keyarray = []
2239
    keyfiles = [constants.SSH_HOST_DSA_PRIV, constants.SSH_HOST_DSA_PUB,
2240
                constants.SSH_HOST_RSA_PRIV, constants.SSH_HOST_RSA_PUB,
2241
                priv_key, pub_key]
2242

    
2243
    for i in keyfiles:
2244
      f = open(i, 'r')
2245
      try:
2246
        keyarray.append(f.read())
2247
      finally:
2248
        f.close()
2249

    
2250
    result = self.rpc.call_node_add(node, keyarray[0], keyarray[1],
2251
                                    keyarray[2],
2252
                                    keyarray[3], keyarray[4], keyarray[5])
2253

    
2254
    msg = result.RemoteFailMsg()
2255
    if msg:
2256
      raise errors.OpExecError("Cannot transfer ssh keys to the"
2257
                               " new node: %s" % msg)
2258

    
2259
    # Add node to our /etc/hosts, and add key to known_hosts
2260
    utils.AddHostToEtcHosts(new_node.name)
2261

    
2262
    if new_node.secondary_ip != new_node.primary_ip:
2263
      result = self.rpc.call_node_has_ip_address(new_node.name,
2264
                                                 new_node.secondary_ip)
2265
      if result.failed or not result.data:
2266
        raise errors.OpExecError("Node claims it doesn't have the secondary ip"
2267
                                 " you gave (%s). Please fix and re-run this"
2268
                                 " command." % new_node.secondary_ip)
2269

    
2270
    node_verify_list = [self.cfg.GetMasterNode()]
2271
    node_verify_param = {
2272
      'nodelist': [node],
2273
      # TODO: do a node-net-test as well?
2274
    }
2275

    
2276
    result = self.rpc.call_node_verify(node_verify_list, node_verify_param,
2277
                                       self.cfg.GetClusterName())
2278
    for verifier in node_verify_list:
2279
      if result[verifier].failed or not result[verifier].data:
2280
        raise errors.OpExecError("Cannot communicate with %s's node daemon"
2281
                                 " for remote verification" % verifier)
2282
      if result[verifier].data['nodelist']:
2283
        for failed in result[verifier].data['nodelist']:
2284
          feedback_fn("ssh/hostname verification failed %s -> %s" %
2285
                      (verifier, result[verifier].data['nodelist'][failed]))
2286
        raise errors.OpExecError("ssh/hostname verification failed.")
2287

    
2288
    # Distribute updated /etc/hosts and known_hosts to all nodes,
2289
    # including the node just added
2290
    myself = self.cfg.GetNodeInfo(self.cfg.GetMasterNode())
2291
    dist_nodes = self.cfg.GetNodeList()
2292
    if not self.op.readd:
2293
      dist_nodes.append(node)
2294
    if myself.name in dist_nodes:
2295
      dist_nodes.remove(myself.name)
2296

    
2297
    logging.debug("Copying hosts and known_hosts to all nodes")
2298
    for fname in (constants.ETC_HOSTS, constants.SSH_KNOWN_HOSTS_FILE):
2299
      result = self.rpc.call_upload_file(dist_nodes, fname)
2300
      for to_node, to_result in result.iteritems():
2301
        if to_result.failed or not to_result.data:
2302
          logging.error("Copy of file %s to node %s failed", fname, to_node)
2303

    
2304
    to_copy = []
2305
    enabled_hypervisors = self.cfg.GetClusterInfo().enabled_hypervisors
2306
    if constants.HTS_COPY_VNC_PASSWORD.intersection(enabled_hypervisors):
2307
      to_copy.append(constants.VNC_PASSWORD_FILE)
2308

    
2309
    for fname in to_copy:
2310
      result = self.rpc.call_upload_file([node], fname)
2311
      if result[node].failed or not result[node]:
2312
        logging.error("Could not copy file %s to node %s", fname, node)
2313

    
2314
    if self.op.readd:
2315
      self.context.ReaddNode(new_node)
2316
    else:
2317
      self.context.AddNode(new_node)
2318

    
2319

    
2320
class LUSetNodeParams(LogicalUnit):
2321
  """Modifies the parameters of a node.
2322

2323
  """
2324
  HPATH = "node-modify"
2325
  HTYPE = constants.HTYPE_NODE
2326
  _OP_REQP = ["node_name"]
2327
  REQ_BGL = False
2328

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

    
2344
  def ExpandNames(self):
2345
    self.needed_locks = {locking.LEVEL_NODE: self.op.node_name}
2346

    
2347
  def BuildHooksEnv(self):
2348
    """Build hooks env.
2349

2350
    This runs on the master node.
2351

2352
    """
2353
    env = {
2354
      "OP_TARGET": self.op.node_name,
2355
      "MASTER_CANDIDATE": str(self.op.master_candidate),
2356
      "OFFLINE": str(self.op.offline),
2357
      "DRAINED": str(self.op.drained),
2358
      }
2359
    nl = [self.cfg.GetMasterNode(),
2360
          self.op.node_name]
2361
    return env, nl, nl
2362

    
2363
  def CheckPrereq(self):
2364
    """Check prerequisites.
2365

2366
    This only checks the instance list against the existing names.
2367

2368
    """
2369
    node = self.node = self.cfg.GetNodeInfo(self.op.node_name)
2370

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

    
2387
    if (self.op.master_candidate == True and
2388
        ((node.offline and not self.op.offline == False) or
2389
         (node.drained and not self.op.drained == False))):
2390
      raise errors.OpPrereqError("Node '%s' is offline or drained, can't set"
2391
                                 " to master_candidate" % node.name)
2392

    
2393
    return
2394

    
2395
  def Exec(self, feedback_fn):
2396
    """Modifies a node.
2397

2398
    """
2399
    node = self.node
2400

    
2401
    result = []
2402
    changed_mc = False
2403

    
2404
    if self.op.offline is not None:
2405
      node.offline = self.op.offline
2406
      result.append(("offline", str(self.op.offline)))
2407
      if self.op.offline == True:
2408
        if node.master_candidate:
2409
          node.master_candidate = False
2410
          changed_mc = True
2411
          result.append(("master_candidate", "auto-demotion due to offline"))
2412
        if node.drained:
2413
          node.drained = False
2414
          result.append(("drained", "clear drained status due to offline"))
2415

    
2416
    if self.op.master_candidate is not None:
2417
      node.master_candidate = self.op.master_candidate
2418
      changed_mc = True
2419
      result.append(("master_candidate", str(self.op.master_candidate)))
2420
      if self.op.master_candidate == False:
2421
        rrc = self.rpc.call_node_demote_from_mc(node.name)
2422
        msg = rrc.RemoteFailMsg()
2423
        if msg:
2424
          self.LogWarning("Node failed to demote itself: %s" % msg)
2425

    
2426
    if self.op.drained is not None:
2427
      node.drained = self.op.drained
2428
      result.append(("drained", str(self.op.drained)))
2429
      if self.op.drained == True:
2430
        if node.master_candidate:
2431
          node.master_candidate = False
2432
          changed_mc = True
2433
          result.append(("master_candidate", "auto-demotion due to drain"))
2434
        if node.offline:
2435
          node.offline = False
2436
          result.append(("offline", "clear offline status due to drain"))
2437

    
2438
    # this will trigger configuration file update, if needed
2439
    self.cfg.Update(node)
2440
    # this will trigger job queue propagation or cleanup
2441
    if changed_mc:
2442
      self.context.ReaddNode(node)
2443

    
2444
    return result
2445

    
2446

    
2447
class LUQueryClusterInfo(NoHooksLU):
2448
  """Query cluster configuration.
2449

2450
  """
2451
  _OP_REQP = []
2452
  REQ_BGL = False
2453

    
2454
  def ExpandNames(self):
2455
    self.needed_locks = {}
2456

    
2457
  def CheckPrereq(self):
2458
    """No prerequsites needed for this LU.
2459

2460
    """
2461
    pass
2462

    
2463
  def Exec(self, feedback_fn):
2464
    """Return cluster config.
2465

2466
    """
2467
    cluster = self.cfg.GetClusterInfo()
2468
    result = {
2469
      "software_version": constants.RELEASE_VERSION,
2470
      "protocol_version": constants.PROTOCOL_VERSION,
2471
      "config_version": constants.CONFIG_VERSION,
2472
      "os_api_version": constants.OS_API_VERSION,
2473
      "export_version": constants.EXPORT_VERSION,
2474
      "architecture": (platform.architecture()[0], platform.machine()),
2475
      "name": cluster.cluster_name,
2476
      "master": cluster.master_node,
2477
      "default_hypervisor": cluster.default_hypervisor,
2478
      "enabled_hypervisors": cluster.enabled_hypervisors,
2479
      "hvparams": dict([(hypervisor, cluster.hvparams[hypervisor])
2480
                        for hypervisor in cluster.enabled_hypervisors]),
2481
      "beparams": cluster.beparams,
2482
      "candidate_pool_size": cluster.candidate_pool_size,
2483
      "default_bridge": cluster.default_bridge,
2484
      "master_netdev": cluster.master_netdev,
2485
      "volume_group_name": cluster.volume_group_name,
2486
      "file_storage_dir": cluster.file_storage_dir,
2487
      }
2488

    
2489
    return result
2490

    
2491

    
2492
class LUQueryConfigValues(NoHooksLU):
2493
  """Return configuration values.
2494

2495
  """
2496
  _OP_REQP = []
2497
  REQ_BGL = False
2498
  _FIELDS_DYNAMIC = utils.FieldSet()
2499
  _FIELDS_STATIC = utils.FieldSet("cluster_name", "master_node", "drain_flag")
2500

    
2501
  def ExpandNames(self):
2502
    self.needed_locks = {}
2503

    
2504
    _CheckOutputFields(static=self._FIELDS_STATIC,
2505
                       dynamic=self._FIELDS_DYNAMIC,
2506
                       selected=self.op.output_fields)
2507

    
2508
  def CheckPrereq(self):
2509
    """No prerequisites.
2510

2511
    """
2512
    pass
2513

    
2514
  def Exec(self, feedback_fn):
2515
    """Dump a representation of the cluster config to the standard output.
2516

2517
    """
2518
    values = []
2519
    for field in self.op.output_fields:
2520
      if field == "cluster_name":
2521
        entry = self.cfg.GetClusterName()
2522
      elif field == "master_node":
2523
        entry = self.cfg.GetMasterNode()
2524
      elif field == "drain_flag":
2525
        entry = os.path.exists(constants.JOB_QUEUE_DRAIN_FILE)
2526
      else:
2527
        raise errors.ParameterError(field)
2528
      values.append(entry)
2529
    return values
2530

    
2531

    
2532
class LUActivateInstanceDisks(NoHooksLU):
2533
  """Bring up an instance's disks.
2534

2535
  """
2536
  _OP_REQP = ["instance_name"]
2537
  REQ_BGL = False
2538

    
2539
  def ExpandNames(self):
2540
    self._ExpandAndLockInstance()
2541
    self.needed_locks[locking.LEVEL_NODE] = []
2542
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2543

    
2544
  def DeclareLocks(self, level):
2545
    if level == locking.LEVEL_NODE:
2546
      self._LockInstancesNodes()
2547

    
2548
  def CheckPrereq(self):
2549
    """Check prerequisites.
2550

2551
    This checks that the instance is in the cluster.
2552

2553
    """
2554
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2555
    assert self.instance is not None, \
2556
      "Cannot retrieve locked instance %s" % self.op.instance_name
2557
    _CheckNodeOnline(self, self.instance.primary_node)
2558

    
2559
  def Exec(self, feedback_fn):
2560
    """Activate the disks.
2561

2562
    """
2563
    disks_ok, disks_info = _AssembleInstanceDisks(self, self.instance)
2564
    if not disks_ok:
2565
      raise errors.OpExecError("Cannot activate block devices")
2566

    
2567
    return disks_info
2568

    
2569

    
2570
def _AssembleInstanceDisks(lu, instance, ignore_secondaries=False):
2571
  """Prepare the block devices for an instance.
2572

2573
  This sets up the block devices on all nodes.
2574

2575
  @type lu: L{LogicalUnit}
2576
  @param lu: the logical unit on whose behalf we execute
2577
  @type instance: L{objects.Instance}
2578
  @param instance: the instance for whose disks we assemble
2579
  @type ignore_secondaries: boolean
2580
  @param ignore_secondaries: if true, errors on secondary nodes
2581
      won't result in an error return from the function
2582
  @return: False if the operation failed, otherwise a list of
2583
      (host, instance_visible_name, node_visible_name)
2584
      with the mapping from node devices to instance devices
2585

2586
  """
2587
  device_info = []
2588
  disks_ok = True
2589
  iname = instance.name
2590
  # With the two passes mechanism we try to reduce the window of
2591
  # opportunity for the race condition of switching DRBD to primary
2592
  # before handshaking occured, but we do not eliminate it
2593

    
2594
  # The proper fix would be to wait (with some limits) until the
2595
  # connection has been made and drbd transitions from WFConnection
2596
  # into any other network-connected state (Connected, SyncTarget,
2597
  # SyncSource, etc.)
2598

    
2599
  # 1st pass, assemble on all nodes in secondary mode
2600
  for inst_disk in instance.disks:
2601
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
2602
      lu.cfg.SetDiskID(node_disk, node)
2603
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, False)
2604
      msg = result.RemoteFailMsg()
2605
      if msg:
2606
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
2607
                           " (is_primary=False, pass=1): %s",
2608
                           inst_disk.iv_name, node, msg)
2609
        if not ignore_secondaries:
2610
          disks_ok = False
2611

    
2612
  # FIXME: race condition on drbd migration to primary
2613

    
2614
  # 2nd pass, do only the primary node
2615
  for inst_disk in instance.disks:
2616
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
2617
      if node != instance.primary_node:
2618
        continue
2619
      lu.cfg.SetDiskID(node_disk, node)
2620
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, True)
2621
      msg = result.RemoteFailMsg()
2622
      if msg:
2623
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
2624
                           " (is_primary=True, pass=2): %s",
2625
                           inst_disk.iv_name, node, msg)
2626
        disks_ok = False
2627
    device_info.append((instance.primary_node, inst_disk.iv_name,
2628
                        result.payload))
2629

    
2630
  # leave the disks configured for the primary node
2631
  # this is a workaround that would be fixed better by
2632
  # improving the logical/physical id handling
2633
  for disk in instance.disks:
2634
    lu.cfg.SetDiskID(disk, instance.primary_node)
2635

    
2636
  return disks_ok, device_info
2637

    
2638

    
2639
def _StartInstanceDisks(lu, instance, force):
2640
  """Start the disks of an instance.
2641

2642
  """
2643
  disks_ok, dummy = _AssembleInstanceDisks(lu, instance,
2644
                                           ignore_secondaries=force)
2645
  if not disks_ok:
2646
    _ShutdownInstanceDisks(lu, instance)
2647
    if force is not None and not force:
2648
      lu.proc.LogWarning("", hint="If the message above refers to a"
2649
                         " secondary node,"
2650
                         " you can retry the operation using '--force'.")
2651
    raise errors.OpExecError("Disk consistency error")
2652

    
2653

    
2654
class LUDeactivateInstanceDisks(NoHooksLU):
2655
  """Shutdown an instance's disks.
2656

2657
  """
2658
  _OP_REQP = ["instance_name"]
2659
  REQ_BGL = False
2660

    
2661
  def ExpandNames(self):
2662
    self._ExpandAndLockInstance()
2663
    self.needed_locks[locking.LEVEL_NODE] = []
2664
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2665

    
2666
  def DeclareLocks(self, level):
2667
    if level == locking.LEVEL_NODE:
2668
      self._LockInstancesNodes()
2669

    
2670
  def CheckPrereq(self):
2671
    """Check prerequisites.
2672

2673
    This checks that the instance is in the cluster.
2674

2675
    """
2676
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2677
    assert self.instance is not None, \
2678
      "Cannot retrieve locked instance %s" % self.op.instance_name
2679

    
2680
  def Exec(self, feedback_fn):
2681
    """Deactivate the disks
2682

2683
    """
2684
    instance = self.instance
2685
    _SafeShutdownInstanceDisks(self, instance)
2686

    
2687

    
2688
def _SafeShutdownInstanceDisks(lu, instance):
2689
  """Shutdown block devices of an instance.
2690

2691
  This function checks if an instance is running, before calling
2692
  _ShutdownInstanceDisks.
2693

2694
  """
2695
  ins_l = lu.rpc.call_instance_list([instance.primary_node],
2696
                                      [instance.hypervisor])
2697
  ins_l = ins_l[instance.primary_node]
2698
  if ins_l.failed or not isinstance(ins_l.data, list):
2699
    raise errors.OpExecError("Can't contact node '%s'" %
2700
                             instance.primary_node)
2701

    
2702
  if instance.name in ins_l.data:
2703
    raise errors.OpExecError("Instance is running, can't shutdown"
2704
                             " block devices.")
2705

    
2706
  _ShutdownInstanceDisks(lu, instance)
2707

    
2708

    
2709
def _ShutdownInstanceDisks(lu, instance, ignore_primary=False):
2710
  """Shutdown block devices of an instance.
2711

2712
  This does the shutdown on all nodes of the instance.
2713

2714
  If the ignore_primary is false, errors on the primary node are
2715
  ignored.
2716

2717
  """
2718
  all_result = True
2719
  for disk in instance.disks:
2720
    for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
2721
      lu.cfg.SetDiskID(top_disk, node)
2722
      result = lu.rpc.call_blockdev_shutdown(node, top_disk)
2723
      msg = result.RemoteFailMsg()
2724
      if msg:
2725
        lu.LogWarning("Could not shutdown block device %s on node %s: %s",
2726
                      disk.iv_name, node, msg)
2727
        if not ignore_primary or node != instance.primary_node:
2728
          all_result = False
2729
  return all_result
2730

    
2731

    
2732
def _CheckNodeFreeMemory(lu, node, reason, requested, hypervisor_name):
2733
  """Checks if a node has enough free memory.
2734

2735
  This function check if a given node has the needed amount of free
2736
  memory. In case the node has less memory or we cannot get the
2737
  information from the node, this function raise an OpPrereqError
2738
  exception.
2739

2740
  @type lu: C{LogicalUnit}
2741
  @param lu: a logical unit from which we get configuration data
2742
  @type node: C{str}
2743
  @param node: the node to check
2744
  @type reason: C{str}
2745
  @param reason: string to use in the error message
2746
  @type requested: C{int}
2747
  @param requested: the amount of memory in MiB to check for
2748
  @type hypervisor_name: C{str}
2749
  @param hypervisor_name: the hypervisor to ask for memory stats
2750
  @raise errors.OpPrereqError: if the node doesn't have enough memory, or
2751
      we cannot check the node
2752

2753
  """
2754
  nodeinfo = lu.rpc.call_node_info([node], lu.cfg.GetVGName(), hypervisor_name)
2755
  nodeinfo[node].Raise()
2756
  free_mem = nodeinfo[node].data.get('memory_free')
2757
  if not isinstance(free_mem, int):
2758
    raise errors.OpPrereqError("Can't compute free memory on node %s, result"
2759
                             " was '%s'" % (node, free_mem))
2760
  if requested > free_mem:
2761
    raise errors.OpPrereqError("Not enough memory on node %s for %s:"
2762
                             " needed %s MiB, available %s MiB" %
2763
                             (node, reason, requested, free_mem))
2764

    
2765

    
2766
class LUStartupInstance(LogicalUnit):
2767
  """Starts an instance.
2768

2769
  """
2770
  HPATH = "instance-start"
2771
  HTYPE = constants.HTYPE_INSTANCE
2772
  _OP_REQP = ["instance_name", "force"]
2773
  REQ_BGL = False
2774

    
2775
  def ExpandNames(self):
2776
    self._ExpandAndLockInstance()
2777

    
2778
  def BuildHooksEnv(self):
2779
    """Build hooks env.
2780

2781
    This runs on master, primary and secondary nodes of the instance.
2782

2783
    """
2784
    env = {
2785
      "FORCE": self.op.force,
2786
      }
2787
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
2788
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2789
    return env, nl, nl
2790

    
2791
  def CheckPrereq(self):
2792
    """Check prerequisites.
2793

2794
    This checks that the instance is in the cluster.
2795

2796
    """
2797
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2798
    assert self.instance is not None, \
2799
      "Cannot retrieve locked instance %s" % self.op.instance_name
2800

    
2801
    # extra beparams
2802
    self.beparams = getattr(self.op, "beparams", {})
2803
    if self.beparams:
2804
      if not isinstance(self.beparams, dict):
2805
        raise errors.OpPrereqError("Invalid beparams passed: %s, expected"
2806
                                   " dict" % (type(self.beparams), ))
2807
      # fill the beparams dict
2808
      utils.ForceDictType(self.beparams, constants.BES_PARAMETER_TYPES)
2809
      self.op.beparams = self.beparams
2810

    
2811
    # extra hvparams
2812
    self.hvparams = getattr(self.op, "hvparams", {})
2813
    if self.hvparams:
2814
      if not isinstance(self.hvparams, dict):
2815
        raise errors.OpPrereqError("Invalid hvparams passed: %s, expected"
2816
                                   " dict" % (type(self.hvparams), ))
2817

    
2818
      # check hypervisor parameter syntax (locally)
2819
      cluster = self.cfg.GetClusterInfo()
2820
      utils.ForceDictType(self.hvparams, constants.HVS_PARAMETER_TYPES)
2821
      filled_hvp = cluster.FillDict(cluster.hvparams[instance.hypervisor],
2822
                                    instance.hvparams)
2823
      filled_hvp.update(self.hvparams)
2824
      hv_type = hypervisor.GetHypervisor(instance.hypervisor)
2825
      hv_type.CheckParameterSyntax(filled_hvp)
2826
      _CheckHVParams(self, instance.all_nodes, instance.hypervisor, filled_hvp)
2827
      self.op.hvparams = self.hvparams
2828

    
2829
    _CheckNodeOnline(self, instance.primary_node)
2830

    
2831
    bep = self.cfg.GetClusterInfo().FillBE(instance)
2832
    # check bridges existance
2833
    _CheckInstanceBridgesExist(self, instance)
2834

    
2835
    remote_info = self.rpc.call_instance_info(instance.primary_node,
2836
                                              instance.name,
2837
                                              instance.hypervisor)
2838
    remote_info.Raise()
2839
    if not remote_info.data:
2840
      _CheckNodeFreeMemory(self, instance.primary_node,
2841
                           "starting instance %s" % instance.name,
2842
                           bep[constants.BE_MEMORY], instance.hypervisor)
2843

    
2844
  def Exec(self, feedback_fn):
2845
    """Start the instance.
2846

2847
    """
2848
    instance = self.instance
2849
    force = self.op.force
2850

    
2851
    self.cfg.MarkInstanceUp(instance.name)
2852

    
2853
    node_current = instance.primary_node
2854

    
2855
    _StartInstanceDisks(self, instance, force)
2856

    
2857
    result = self.rpc.call_instance_start(node_current, instance,
2858
                                          self.hvparams, self.beparams)
2859
    msg = result.RemoteFailMsg()
2860
    if msg:
2861
      _ShutdownInstanceDisks(self, instance)
2862
      raise errors.OpExecError("Could not start instance: %s" % msg)
2863

    
2864

    
2865
class LURebootInstance(LogicalUnit):
2866
  """Reboot an instance.
2867

2868
  """
2869
  HPATH = "instance-reboot"
2870
  HTYPE = constants.HTYPE_INSTANCE
2871
  _OP_REQP = ["instance_name", "ignore_secondaries", "reboot_type"]
2872
  REQ_BGL = False
2873

    
2874
  def ExpandNames(self):
2875
    if self.op.reboot_type not in [constants.INSTANCE_REBOOT_SOFT,
2876
                                   constants.INSTANCE_REBOOT_HARD,
2877
                                   constants.INSTANCE_REBOOT_FULL]:
2878
      raise errors.ParameterError("reboot type not in [%s, %s, %s]" %
2879
                                  (constants.INSTANCE_REBOOT_SOFT,
2880
                                   constants.INSTANCE_REBOOT_HARD,
2881
                                   constants.INSTANCE_REBOOT_FULL))
2882
    self._ExpandAndLockInstance()
2883

    
2884
  def BuildHooksEnv(self):
2885
    """Build hooks env.
2886

2887
    This runs on master, primary and secondary nodes of the instance.
2888

2889
    """
2890
    env = {
2891
      "IGNORE_SECONDARIES": self.op.ignore_secondaries,
2892
      "REBOOT_TYPE": self.op.reboot_type,
2893
      }
2894
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
2895
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2896
    return env, nl, nl
2897

    
2898
  def CheckPrereq(self):
2899
    """Check prerequisites.
2900

2901
    This checks that the instance is in the cluster.
2902

2903
    """
2904
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2905
    assert self.instance is not None, \
2906
      "Cannot retrieve locked instance %s" % self.op.instance_name
2907

    
2908
    _CheckNodeOnline(self, instance.primary_node)
2909

    
2910
    # check bridges existance
2911
    _CheckInstanceBridgesExist(self, instance)
2912

    
2913
  def Exec(self, feedback_fn):
2914
    """Reboot the instance.
2915

2916
    """
2917
    instance = self.instance
2918
    ignore_secondaries = self.op.ignore_secondaries
2919
    reboot_type = self.op.reboot_type
2920

    
2921
    node_current = instance.primary_node
2922

    
2923
    if reboot_type in [constants.INSTANCE_REBOOT_SOFT,
2924
                       constants.INSTANCE_REBOOT_HARD]:
2925
      for disk in instance.disks:
2926
        self.cfg.SetDiskID(disk, node_current)
2927
      result = self.rpc.call_instance_reboot(node_current, instance,
2928
                                             reboot_type)
2929
      msg = result.RemoteFailMsg()
2930
      if msg:
2931
        raise errors.OpExecError("Could not reboot instance: %s" % msg)
2932
    else:
2933
      result = self.rpc.call_instance_shutdown(node_current, instance)
2934
      msg = result.RemoteFailMsg()
2935
      if msg:
2936
        raise errors.OpExecError("Could not shutdown instance for"
2937
                                 " full reboot: %s" % msg)
2938
      _ShutdownInstanceDisks(self, instance)
2939
      _StartInstanceDisks(self, instance, ignore_secondaries)
2940
      result = self.rpc.call_instance_start(node_current, instance, None, None)
2941
      msg = result.RemoteFailMsg()
2942
      if msg:
2943
        _ShutdownInstanceDisks(self, instance)
2944
        raise errors.OpExecError("Could not start instance for"
2945
                                 " full reboot: %s" % msg)
2946

    
2947
    self.cfg.MarkInstanceUp(instance.name)
2948

    
2949

    
2950
class LUShutdownInstance(LogicalUnit):
2951
  """Shutdown an instance.
2952

2953
  """
2954
  HPATH = "instance-stop"
2955
  HTYPE = constants.HTYPE_INSTANCE
2956
  _OP_REQP = ["instance_name"]
2957
  REQ_BGL = False
2958

    
2959
  def ExpandNames(self):
2960
    self._ExpandAndLockInstance()
2961

    
2962
  def BuildHooksEnv(self):
2963
    """Build hooks env.
2964

2965
    This runs on master, primary and secondary nodes of the instance.
2966

2967
    """
2968
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2969
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2970
    return env, nl, nl
2971

    
2972
  def CheckPrereq(self):
2973
    """Check prerequisites.
2974

2975
    This checks that the instance is in the cluster.
2976

2977
    """
2978
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2979
    assert self.instance is not None, \
2980
      "Cannot retrieve locked instance %s" % self.op.instance_name
2981
    _CheckNodeOnline(self, self.instance.primary_node)
2982

    
2983
  def Exec(self, feedback_fn):
2984
    """Shutdown the instance.
2985

2986
    """
2987
    instance = self.instance
2988
    node_current = instance.primary_node
2989
    self.cfg.MarkInstanceDown(instance.name)
2990
    result = self.rpc.call_instance_shutdown(node_current, instance)
2991
    msg = result.RemoteFailMsg()
2992
    if msg:
2993
      self.proc.LogWarning("Could not shutdown instance: %s" % msg)
2994

    
2995
    _ShutdownInstanceDisks(self, instance)
2996

    
2997

    
2998
class LUReinstallInstance(LogicalUnit):
2999
  """Reinstall an instance.
3000

3001
  """
3002
  HPATH = "instance-reinstall"
3003
  HTYPE = constants.HTYPE_INSTANCE
3004
  _OP_REQP = ["instance_name"]
3005
  REQ_BGL = False
3006

    
3007
  def ExpandNames(self):
3008
    self._ExpandAndLockInstance()
3009

    
3010
  def BuildHooksEnv(self):
3011
    """Build hooks env.
3012

3013
    This runs on master, primary and secondary nodes of the instance.
3014

3015
    """
3016
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3017
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3018
    return env, nl, nl
3019

    
3020
  def CheckPrereq(self):
3021
    """Check prerequisites.
3022

3023
    This checks that the instance is in the cluster and is not running.
3024

3025
    """
3026
    instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3027
    assert instance is not None, \
3028
      "Cannot retrieve locked instance %s" % self.op.instance_name
3029
    _CheckNodeOnline(self, instance.primary_node)
3030

    
3031
    if instance.disk_template == constants.DT_DISKLESS:
3032
      raise errors.OpPrereqError("Instance '%s' has no disks" %
3033
                                 self.op.instance_name)
3034
    if instance.admin_up:
3035
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
3036
                                 self.op.instance_name)
3037
    remote_info = self.rpc.call_instance_info(instance.primary_node,
3038
                                              instance.name,
3039
                                              instance.hypervisor)
3040
    remote_info.Raise()
3041
    if remote_info.data:
3042
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
3043
                                 (self.op.instance_name,
3044
                                  instance.primary_node))
3045

    
3046
    self.op.os_type = getattr(self.op, "os_type", None)
3047
    if self.op.os_type is not None:
3048
      # OS verification
3049
      pnode = self.cfg.GetNodeInfo(
3050
        self.cfg.ExpandNodeName(instance.primary_node))
3051
      if pnode is None:
3052
        raise errors.OpPrereqError("Primary node '%s' is unknown" %
3053
                                   self.op.pnode)
3054
      result = self.rpc.call_os_get(pnode.name, self.op.os_type)
3055
      result.Raise()
3056
      if not isinstance(result.data, objects.OS):
3057
        raise errors.OpPrereqError("OS '%s' not in supported OS list for"
3058
                                   " primary node"  % self.op.os_type)
3059

    
3060
    self.instance = instance
3061

    
3062
  def Exec(self, feedback_fn):
3063
    """Reinstall the instance.
3064

3065
    """
3066
    inst = self.instance
3067

    
3068
    if self.op.os_type is not None:
3069
      feedback_fn("Changing OS to '%s'..." % self.op.os_type)
3070
      inst.os = self.op.os_type
3071
      self.cfg.Update(inst)
3072

    
3073
    _StartInstanceDisks(self, inst, None)
3074
    try:
3075
      feedback_fn("Running the instance OS create scripts...")
3076
      result = self.rpc.call_instance_os_add(inst.primary_node, inst)
3077
      msg = result.RemoteFailMsg()
3078
      if msg:
3079
        raise errors.OpExecError("Could not install OS for instance %s"
3080
                                 " on node %s: %s" %
3081
                                 (inst.name, inst.primary_node, msg))
3082
    finally:
3083
      _ShutdownInstanceDisks(self, inst)
3084

    
3085

    
3086
class LURenameInstance(LogicalUnit):
3087
  """Rename an instance.
3088

3089
  """
3090
  HPATH = "instance-rename"
3091
  HTYPE = constants.HTYPE_INSTANCE
3092
  _OP_REQP = ["instance_name", "new_name"]
3093

    
3094
  def BuildHooksEnv(self):
3095
    """Build hooks env.
3096

3097
    This runs on master, primary and secondary nodes of the instance.
3098

3099
    """
3100
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3101
    env["INSTANCE_NEW_NAME"] = self.op.new_name
3102
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3103
    return env, nl, nl
3104

    
3105
  def CheckPrereq(self):
3106
    """Check prerequisites.
3107

3108
    This checks that the instance is in the cluster and is not running.
3109

3110
    """
3111
    instance = self.cfg.GetInstanceInfo(
3112
      self.cfg.ExpandInstanceName(self.op.instance_name))
3113
    if instance is None:
3114
      raise errors.OpPrereqError("Instance '%s' not known" %
3115
                                 self.op.instance_name)
3116
    _CheckNodeOnline(self, instance.primary_node)
3117

    
3118
    if instance.admin_up:
3119
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
3120
                                 self.op.instance_name)
3121
    remote_info = self.rpc.call_instance_info(instance.primary_node,
3122
                                              instance.name,
3123
                                              instance.hypervisor)
3124
    remote_info.Raise()
3125
    if remote_info.data:
3126
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
3127
                                 (self.op.instance_name,
3128
                                  instance.primary_node))
3129
    self.instance = instance
3130

    
3131
    # new name verification
3132
    name_info = utils.HostInfo(self.op.new_name)
3133

    
3134
    self.op.new_name = new_name = name_info.name
3135
    instance_list = self.cfg.GetInstanceList()
3136
    if new_name in instance_list:
3137
      raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
3138
                                 new_name)
3139

    
3140
    if not getattr(self.op, "ignore_ip", False):
3141
      if utils.TcpPing(name_info.ip, constants.DEFAULT_NODED_PORT):
3142
        raise errors.OpPrereqError("IP %s of instance %s already in use" %
3143
                                   (name_info.ip, new_name))
3144

    
3145

    
3146
  def Exec(self, feedback_fn):
3147
    """Reinstall the instance.
3148

3149
    """
3150
    inst = self.instance
3151
    old_name = inst.name
3152

    
3153
    if inst.disk_template == constants.DT_FILE:
3154
      old_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
3155

    
3156
    self.cfg.RenameInstance(inst.name, self.op.new_name)
3157
    # Change the instance lock. This is definitely safe while we hold the BGL
3158
    self.context.glm.remove(locking.LEVEL_INSTANCE, old_name)
3159
    self.context.glm.add(locking.LEVEL_INSTANCE, self.op.new_name)
3160

    
3161
    # re-read the instance from the configuration after rename
3162
    inst = self.cfg.GetInstanceInfo(self.op.new_name)
3163

    
3164
    if inst.disk_template == constants.DT_FILE:
3165
      new_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
3166
      result = self.rpc.call_file_storage_dir_rename(inst.primary_node,
3167
                                                     old_file_storage_dir,
3168
                                                     new_file_storage_dir)
3169
      result.Raise()
3170
      if not result.data:
3171
        raise errors.OpExecError("Could not connect to node '%s' to rename"
3172
                                 " directory '%s' to '%s' (but the instance"
3173
                                 " has been renamed in Ganeti)" % (
3174
                                 inst.primary_node, old_file_storage_dir,
3175
                                 new_file_storage_dir))
3176

    
3177
      if not result.data[0]:
3178
        raise errors.OpExecError("Could not rename directory '%s' to '%s'"
3179
                                 " (but the instance has been renamed in"
3180
                                 " Ganeti)" % (old_file_storage_dir,
3181
                                               new_file_storage_dir))
3182

    
3183
    _StartInstanceDisks(self, inst, None)
3184
    try:
3185
      result = self.rpc.call_instance_run_rename(inst.primary_node, inst,
3186
                                                 old_name)
3187
      msg = result.RemoteFailMsg()
3188
      if msg:
3189
        msg = ("Could not run OS rename script for instance %s on node %s"
3190
               " (but the instance has been renamed in Ganeti): %s" %
3191
               (inst.name, inst.primary_node, msg))
3192
        self.proc.LogWarning(msg)
3193
    finally:
3194
      _ShutdownInstanceDisks(self, inst)
3195

    
3196

    
3197
class LURemoveInstance(LogicalUnit):
3198
  """Remove an instance.
3199

3200
  """
3201
  HPATH = "instance-remove"
3202
  HTYPE = constants.HTYPE_INSTANCE
3203
  _OP_REQP = ["instance_name", "ignore_failures"]
3204
  REQ_BGL = False
3205

    
3206
  def ExpandNames(self):
3207
    self._ExpandAndLockInstance()
3208
    self.needed_locks[locking.LEVEL_NODE] = []
3209
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3210

    
3211
  def DeclareLocks(self, level):
3212
    if level == locking.LEVEL_NODE:
3213
      self._LockInstancesNodes()
3214

    
3215
  def BuildHooksEnv(self):
3216
    """Build hooks env.
3217

3218
    This runs on master, primary and secondary nodes of the instance.
3219

3220
    """
3221
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3222
    nl = [self.cfg.GetMasterNode()]
3223
    return env, nl, nl
3224

    
3225
  def CheckPrereq(self):
3226
    """Check prerequisites.
3227

3228
    This checks that the instance is in the cluster.
3229

3230
    """
3231
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3232
    assert self.instance is not None, \
3233
      "Cannot retrieve locked instance %s" % self.op.instance_name
3234

    
3235
  def Exec(self, feedback_fn):
3236
    """Remove the instance.
3237

3238
    """
3239
    instance = self.instance
3240
    logging.info("Shutting down instance %s on node %s",
3241
                 instance.name, instance.primary_node)
3242

    
3243
    result = self.rpc.call_instance_shutdown(instance.primary_node, instance)
3244
    msg = result.RemoteFailMsg()
3245
    if msg:
3246
      if self.op.ignore_failures:
3247
        feedback_fn("Warning: can't shutdown instance: %s" % msg)
3248
      else:
3249
        raise errors.OpExecError("Could not shutdown instance %s on"
3250
                                 " node %s: %s" %
3251
                                 (instance.name, instance.primary_node, msg))
3252

    
3253
    logging.info("Removing block devices for instance %s", instance.name)
3254

    
3255
    if not _RemoveDisks(self, instance):
3256
      if self.op.ignore_failures:
3257
        feedback_fn("Warning: can't remove instance's disks")
3258
      else:
3259
        raise errors.OpExecError("Can't remove instance's disks")
3260

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

    
3263
    self.cfg.RemoveInstance(instance.name)
3264
    self.remove_locks[locking.LEVEL_INSTANCE] = instance.name
3265

    
3266

    
3267
class LUQueryInstances(NoHooksLU):
3268
  """Logical unit for querying instances.
3269

3270
  """
3271
  _OP_REQP = ["output_fields", "names", "use_locking"]
3272
  REQ_BGL = False
3273
  _FIELDS_STATIC = utils.FieldSet(*["name", "os", "pnode", "snodes",
3274
                                    "admin_state",
3275
                                    "disk_template", "ip", "mac", "bridge",
3276
                                    "sda_size", "sdb_size", "vcpus", "tags",
3277
                                    "network_port", "beparams",
3278
                                    r"(disk)\.(size)/([0-9]+)",
3279
                                    r"(disk)\.(sizes)", "disk_usage",
3280
                                    r"(nic)\.(mac|ip|bridge)/([0-9]+)",
3281
                                    r"(nic)\.(macs|ips|bridges)",
3282
                                    r"(disk|nic)\.(count)",
3283
                                    "serial_no", "hypervisor", "hvparams",] +
3284
                                  ["hv/%s" % name
3285
                                   for name in constants.HVS_PARAMETERS] +
3286
                                  ["be/%s" % name
3287
                                   for name in constants.BES_PARAMETERS])
3288
  _FIELDS_DYNAMIC = utils.FieldSet("oper_state", "oper_ram", "status")
3289

    
3290

    
3291
  def ExpandNames(self):
3292
    _CheckOutputFields(static=self._FIELDS_STATIC,
3293
                       dynamic=self._FIELDS_DYNAMIC,
3294
                       selected=self.op.output_fields)
3295

    
3296
    self.needed_locks = {}
3297
    self.share_locks[locking.LEVEL_INSTANCE] = 1
3298
    self.share_locks[locking.LEVEL_NODE] = 1
3299

    
3300
    if self.op.names:
3301
      self.wanted = _GetWantedInstances(self, self.op.names)
3302
    else:
3303
      self.wanted = locking.ALL_SET
3304

    
3305
    self.do_node_query = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
3306
    self.do_locking = self.do_node_query and self.op.use_locking
3307
    if self.do_locking:
3308
      self.needed_locks[locking.LEVEL_INSTANCE] = self.wanted
3309
      self.needed_locks[locking.LEVEL_NODE] = []
3310
      self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3311

    
3312
  def DeclareLocks(self, level):
3313
    if level == locking.LEVEL_NODE and self.do_locking:
3314
      self._LockInstancesNodes()
3315

    
3316
  def CheckPrereq(self):
3317
    """Check prerequisites.
3318

3319
    """
3320
    pass
3321

    
3322
  def Exec(self, feedback_fn):
3323
    """Computes the list of nodes and their attributes.
3324

3325
    """
3326
    all_info = self.cfg.GetAllInstancesInfo()
3327
    if self.wanted == locking.ALL_SET:
3328
      # caller didn't specify instance names, so ordering is not important
3329
      if self.do_locking:
3330
        instance_names = self.acquired_locks[locking.LEVEL_INSTANCE]
3331
      else:
3332
        instance_names = all_info.keys()
3333
      instance_names = utils.NiceSort(instance_names)
3334
    else:
3335
      # caller did specify names, so we must keep the ordering
3336
      if self.do_locking:
3337
        tgt_set = self.acquired_locks[locking.LEVEL_INSTANCE]
3338
      else:
3339
        tgt_set = all_info.keys()
3340
      missing = set(self.wanted).difference(tgt_set)
3341
      if missing:
3342
        raise errors.OpExecError("Some instances were removed before"
3343
                                 " retrieving their data: %s" % missing)
3344
      instance_names = self.wanted
3345

    
3346
    instance_list = [all_info[iname] for iname in instance_names]
3347

    
3348
    # begin data gathering
3349

    
3350
    nodes = frozenset([inst.primary_node for inst in instance_list])
3351
    hv_list = list(set([inst.hypervisor for inst in instance_list]))
3352

    
3353
    bad_nodes = []
3354
    off_nodes = []
3355
    if self.do_node_query:
3356
      live_data = {}
3357
      node_data = self.rpc.call_all_instances_info(nodes, hv_list)
3358
      for name in nodes:
3359
        result = node_data[name]
3360
        if result.offline:
3361
          # offline nodes will be in both lists
3362
          off_nodes.append(name)
3363
        if result.failed:
3364
          bad_nodes.append(name)
3365
        else:
3366
          if result.data:
3367
            live_data.update(result.data)
3368
            # else no instance is alive
3369
    else:
3370
      live_data = dict([(name, {}) for name in instance_names])
3371

    
3372
    # end data gathering
3373

    
3374
    HVPREFIX = "hv/"
3375
    BEPREFIX = "be/"
3376
    output = []
3377
    for instance in instance_list:
3378
      iout = []
3379
      i_hv = self.cfg.GetClusterInfo().FillHV(instance)
3380
      i_be = self.cfg.GetClusterInfo().FillBE(instance)
3381
      for field in self.op.output_fields:
3382
        st_match = self._FIELDS_STATIC.Matches(field)
3383
        if field == "name":
3384
          val = instance.name
3385
        elif field == "os":
3386
          val = instance.os
3387
        elif field == "pnode":
3388
          val = instance.primary_node
3389
        elif field == "snodes":
3390
          val = list(instance.secondary_nodes)
3391
        elif field == "admin_state":
3392
          val = instance.admin_up
3393
        elif field == "oper_state":
3394
          if instance.primary_node in bad_nodes:
3395
            val = None
3396
          else:
3397
            val = bool(live_data.get(instance.name))
3398
        elif field == "status":
3399
          if instance.primary_node in off_nodes:
3400
            val = "ERROR_nodeoffline"
3401
          elif instance.primary_node in bad_nodes:
3402
            val = "ERROR_nodedown"
3403
          else:
3404
            running = bool(live_data.get(instance.name))
3405
            if running:
3406
              if instance.admin_up:
3407
                val = "running"
3408
              else:
3409
                val = "ERROR_up"
3410
            else:
3411
              if instance.admin_up:
3412
                val = "ERROR_down"
3413
              else:
3414
                val = "ADMIN_down"
3415
        elif field == "oper_ram":
3416
          if instance.primary_node in bad_nodes:
3417
            val = None
3418
          elif instance.name in live_data:
3419
            val = live_data[instance.name].get("memory", "?")
3420
          else:
3421
            val = "-"
3422
        elif field == "vcpus":
3423
          val = i_be[constants.BE_VCPUS]
3424
        elif field == "disk_template":
3425
          val = instance.disk_template
3426
        elif field == "ip":
3427
          if instance.nics:
3428
            val = instance.nics[0].ip
3429
          else:
3430
            val = None
3431
        elif field == "bridge":
3432
          if instance.nics:
3433
            val = instance.nics[0].bridge
3434
          else:
3435
            val = None
3436
        elif field == "mac":
3437
          if instance.nics:
3438
            val = instance.nics[0].mac
3439
          else:
3440
            val = None
3441
        elif field == "sda_size" or field == "sdb_size":
3442
          idx = ord(field[2]) - ord('a')
3443
          try:
3444
            val = instance.FindDisk(idx).size
3445
          except errors.OpPrereqError:
3446
            val = None
3447
        elif field == "disk_usage": # total disk usage per node
3448
          disk_sizes = [{'size': disk.size} for disk in instance.disks]
3449
          val = _ComputeDiskSize(instance.disk_template, disk_sizes)
3450
        elif field == "tags":
3451
          val = list(instance.GetTags())
3452
        elif field == "serial_no":
3453
          val = instance.serial_no
3454
        elif field == "network_port":
3455
          val = instance.network_port
3456
        elif field == "hypervisor":
3457
          val = instance.hypervisor
3458
        elif field == "hvparams":
3459
          val = i_hv
3460
        elif (field.startswith(HVPREFIX) and
3461
              field[len(HVPREFIX):] in constants.HVS_PARAMETERS):
3462
          val = i_hv.get(field[len(HVPREFIX):], None)
3463
        elif field == "beparams":
3464
          val = i_be
3465
        elif (field.startswith(BEPREFIX) and
3466
              field[len(BEPREFIX):] in constants.BES_PARAMETERS):
3467
          val = i_be.get(field[len(BEPREFIX):], None)
3468
        elif st_match and st_match.groups():
3469
          # matches a variable list
3470
          st_groups = st_match.groups()
3471
          if st_groups and st_groups[0] == "disk":
3472
            if st_groups[1] == "count":
3473
              val = len(instance.disks)
3474
            elif st_groups[1] == "sizes":
3475
              val = [disk.size for disk in instance.disks]
3476
            elif st_groups[1] == "size":
3477
              try:
3478
                val = instance.FindDisk(st_groups[2]).size
3479
              except errors.OpPrereqError:
3480
                val = None
3481
            else:
3482
              assert False, "Unhandled disk parameter"
3483
          elif st_groups[0] == "nic":
3484
            if st_groups[1] == "count":
3485
              val = len(instance.nics)
3486
            elif st_groups[1] == "macs":
3487
              val = [nic.mac for nic in instance.nics]
3488
            elif st_groups[1] == "ips":
3489
              val = [nic.ip for nic in instance.nics]
3490
            elif st_groups[1] == "bridges":
3491
              val = [nic.bridge for nic in instance.nics]
3492
            else:
3493
              # index-based item
3494
              nic_idx = int(st_groups[2])
3495
              if nic_idx >= len(instance.nics):
3496
                val = None
3497
              else:
3498
                if st_groups[1] == "mac":
3499
                  val = instance.nics[nic_idx].mac
3500
                elif st_groups[1] == "ip":
3501
                  val = instance.nics[nic_idx].ip
3502
                elif st_groups[1] == "bridge":
3503
                  val = instance.nics[nic_idx].bridge
3504
                else:
3505
                  assert False, "Unhandled NIC parameter"
3506
          else:
3507
            assert False, ("Declared but unhandled variable parameter '%s'" %
3508
                           field)
3509
        else:
3510
          assert False, "Declared but unhandled parameter '%s'" % field
3511
        iout.append(val)
3512
      output.append(iout)
3513

    
3514
    return output
3515

    
3516

    
3517
class LUFailoverInstance(LogicalUnit):
3518
  """Failover an instance.
3519

3520
  """
3521
  HPATH = "instance-failover"
3522
  HTYPE = constants.HTYPE_INSTANCE
3523
  _OP_REQP = ["instance_name", "ignore_consistency"]
3524
  REQ_BGL = False
3525

    
3526
  def ExpandNames(self):
3527
    self._ExpandAndLockInstance()
3528
    self.needed_locks[locking.LEVEL_NODE] = []
3529
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3530

    
3531
  def DeclareLocks(self, level):
3532
    if level == locking.LEVEL_NODE:
3533
      self._LockInstancesNodes()
3534

    
3535
  def BuildHooksEnv(self):
3536
    """Build hooks env.
3537

3538
    This runs on master, primary and secondary nodes of the instance.
3539

3540
    """
3541
    env = {
3542
      "IGNORE_CONSISTENCY": self.op.ignore_consistency,
3543
      }
3544
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3545
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3546
    return env, nl, nl
3547

    
3548
  def CheckPrereq(self):
3549
    """Check prerequisites.
3550

3551
    This checks that the instance is in the cluster.
3552

3553
    """
3554
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3555
    assert self.instance is not None, \
3556
      "Cannot retrieve locked instance %s" % self.op.instance_name
3557

    
3558
    bep = self.cfg.GetClusterInfo().FillBE(instance)
3559
    if instance.disk_template not in constants.DTS_NET_MIRROR:
3560
      raise errors.OpPrereqError("Instance's disk layout is not"
3561
                                 " network mirrored, cannot failover.")
3562

    
3563
    secondary_nodes = instance.secondary_nodes
3564
    if not secondary_nodes:
3565
      raise errors.ProgrammerError("no secondary node but using "
3566
                                   "a mirrored disk template")
3567

    
3568
    target_node = secondary_nodes[0]
3569
    _CheckNodeOnline(self, target_node)
3570
    _CheckNodeNotDrained(self, target_node)
3571

    
3572
    if instance.admin_up:
3573
      # check memory requirements on the secondary node
3574
      _CheckNodeFreeMemory(self, target_node, "failing over instance %s" %
3575
                           instance.name, bep[constants.BE_MEMORY],
3576
                           instance.hypervisor)
3577
    else:
3578
      self.LogInfo("Not checking memory on the secondary node as"
3579
                   " instance will not be started")
3580

    
3581
    # check bridge existance
3582
    brlist = [nic.bridge for nic in instance.nics]
3583
    result = self.rpc.call_bridges_exist(target_node, brlist)
3584
    result.Raise()
3585
    if not result.data:
3586
      raise errors.OpPrereqError("One or more target bridges %s does not"
3587
                                 " exist on destination node '%s'" %
3588
                                 (brlist, target_node))
3589

    
3590
  def Exec(self, feedback_fn):
3591
    """Failover an instance.
3592

3593
    The failover is done by shutting it down on its present node and
3594
    starting it on the secondary.
3595

3596
    """
3597
    instance = self.instance
3598

    
3599
    source_node = instance.primary_node
3600
    target_node = instance.secondary_nodes[0]
3601

    
3602
    feedback_fn("* checking disk consistency between source and target")
3603
    for dev in instance.disks:
3604
      # for drbd, these are drbd over lvm
3605
      if not _CheckDiskConsistency(self, dev, target_node, False):
3606
        if instance.admin_up and not self.op.ignore_consistency:
3607
          raise errors.OpExecError("Disk %s is degraded on target node,"
3608
                                   " aborting failover." % dev.iv_name)
3609

    
3610
    feedback_fn("* shutting down instance on source node")
3611
    logging.info("Shutting down instance %s on node %s",
3612
                 instance.name, source_node)
3613

    
3614
    result = self.rpc.call_instance_shutdown(source_node, instance)
3615
    msg = result.RemoteFailMsg()
3616
    if msg:
3617
      if self.op.ignore_consistency:
3618
        self.proc.LogWarning("Could not shutdown instance %s on node %s."
3619
                             " Proceeding anyway. Please make sure node"
3620
                             " %s is down. Error details: %s",
3621
                             instance.name, source_node, source_node, msg)
3622
      else:
3623
        raise errors.OpExecError("Could not shutdown instance %s on"
3624
                                 " node %s: %s" %
3625
                                 (instance.name, source_node, msg))
3626

    
3627
    feedback_fn("* deactivating the instance's disks on source node")
3628
    if not _ShutdownInstanceDisks(self, instance, ignore_primary=True):
3629
      raise errors.OpExecError("Can't shut down the instance's disks.")
3630

    
3631
    instance.primary_node = target_node
3632
    # distribute new instance config to the other nodes
3633
    self.cfg.Update(instance)
3634

    
3635
    # Only start the instance if it's marked as up
3636
    if instance.admin_up:
3637
      feedback_fn("* activating the instance's disks on target node")
3638
      logging.info("Starting instance %s on node %s",
3639
                   instance.name, target_node)
3640

    
3641
      disks_ok, dummy = _AssembleInstanceDisks(self, instance,
3642
                                               ignore_secondaries=True)
3643
      if not disks_ok:
3644
        _ShutdownInstanceDisks(self, instance)
3645
        raise errors.OpExecError("Can't activate the instance's disks")
3646

    
3647
      feedback_fn("* starting the instance on the target node")
3648
      result = self.rpc.call_instance_start(target_node, instance, None, None)
3649
      msg = result.RemoteFailMsg()
3650
      if msg:
3651
        _ShutdownInstanceDisks(self, instance)
3652
        raise errors.OpExecError("Could not start instance %s on node %s: %s" %
3653
                                 (instance.name, target_node, msg))
3654

    
3655

    
3656
class LUMigrateInstance(LogicalUnit):
3657
  """Migrate an instance.
3658

3659
  This is migration without shutting down, compared to the failover,
3660
  which is done with shutdown.
3661

3662
  """
3663
  HPATH = "instance-migrate"
3664
  HTYPE = constants.HTYPE_INSTANCE
3665
  _OP_REQP = ["instance_name", "live", "cleanup"]
3666

    
3667
  REQ_BGL = False
3668

    
3669
  def ExpandNames(self):
3670
    self._ExpandAndLockInstance()
3671
    self.needed_locks[locking.LEVEL_NODE] = []
3672
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3673

    
3674
  def DeclareLocks(self, level):
3675
    if level == locking.LEVEL_NODE:
3676
      self._LockInstancesNodes()
3677

    
3678
  def BuildHooksEnv(self):
3679
    """Build hooks env.
3680

3681
    This runs on master, primary and secondary nodes of the instance.
3682

3683
    """
3684
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3685
    env["MIGRATE_LIVE"] = self.op.live
3686
    env["MIGRATE_CLEANUP"] = self.op.cleanup
3687
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3688
    return env, nl, nl
3689

    
3690
  def CheckPrereq(self):
3691
    """Check prerequisites.
3692

3693
    This checks that the instance is in the cluster.
3694

3695
    """
3696
    instance = self.cfg.GetInstanceInfo(
3697
      self.cfg.ExpandInstanceName(self.op.instance_name))
3698
    if instance is None:
3699
      raise errors.OpPrereqError("Instance '%s' not known" %
3700
                                 self.op.instance_name)
3701

    
3702
    if instance.disk_template != constants.DT_DRBD8:
3703
      raise errors.OpPrereqError("Instance's disk layout is not"
3704
                                 " drbd8, cannot migrate.")
3705

    
3706
    secondary_nodes = instance.secondary_nodes
3707
    if not secondary_nodes:
3708
      raise errors.ConfigurationError("No secondary node but using"
3709
                                      " drbd8 disk template")
3710

    
3711
    i_be = self.cfg.GetClusterInfo().FillBE(instance)
3712

    
3713
    target_node = secondary_nodes[0]
3714
    # check memory requirements on the secondary node
3715
    _CheckNodeFreeMemory(self, target_node, "migrating instance %s" %
3716
                         instance.name, i_be[constants.BE_MEMORY],
3717
                         instance.hypervisor)
3718

    
3719
    # check bridge existance
3720
    brlist = [nic.bridge for nic in instance.nics]
3721
    result = self.rpc.call_bridges_exist(target_node, brlist)
3722
    if result.failed or not result.data:
3723
      raise errors.OpPrereqError("One or more target bridges %s does not"
3724
                                 " exist on destination node '%s'" %
3725
                                 (brlist, target_node))
3726

    
3727
    if not self.op.cleanup:
3728
      _CheckNodeNotDrained(self, target_node)
3729
      result = self.rpc.call_instance_migratable(instance.primary_node,
3730
                                                 instance)
3731
      msg = result.RemoteFailMsg()
3732
      if msg:
3733
        raise errors.OpPrereqError("Can't migrate: %s - please use failover" %
3734
                                   msg)
3735

    
3736
    self.instance = instance
3737

    
3738
  def _WaitUntilSync(self):
3739
    """Poll with custom rpc for disk sync.
3740

3741
    This uses our own step-based rpc call.
3742

3743
    """
3744
    self.feedback_fn("* wait until resync is done")
3745
    all_done = False
3746
    while not all_done:
3747
      all_done = True
3748
      result = self.rpc.call_drbd_wait_sync(self.all_nodes,
3749
                                            self.nodes_ip,
3750
                                            self.instance.disks)
3751
      min_percent = 100
3752
      for node, nres in result.items():
3753
        msg = nres.RemoteFailMsg()
3754
        if msg:
3755
          raise errors.OpExecError("Cannot resync disks on node %s: %s" %
3756
                                   (node, msg))
3757
        node_done, node_percent = nres.payload
3758
        all_done = all_done and node_done
3759
        if node_percent is not None:
3760
          min_percent = min(min_percent, node_percent)
3761
      if not all_done:
3762
        if min_percent < 100:
3763
          self.feedback_fn("   - progress: %.1f%%" % min_percent)
3764
        time.sleep(2)
3765

    
3766
  def _EnsureSecondary(self, node):
3767
    """Demote a node to secondary.
3768

3769
    """
3770
    self.feedback_fn("* switching node %s to secondary mode" % node)
3771

    
3772
    for dev in self.instance.disks:
3773
      self.cfg.SetDiskID(dev, node)
3774

    
3775
    result = self.rpc.call_blockdev_close(node, self.instance.name,
3776
                                          self.instance.disks)
3777
    msg = result.RemoteFailMsg()
3778
    if msg:
3779
      raise errors.OpExecError("Cannot change disk to secondary on node %s,"
3780
                               " error %s" % (node, msg))
3781

    
3782
  def _GoStandalone(self):
3783
    """Disconnect from the network.
3784

3785
    """
3786
    self.feedback_fn("* changing into standalone mode")
3787
    result = self.rpc.call_drbd_disconnect_net(self.all_nodes, self.nodes_ip,
3788
                                               self.instance.disks)
3789
    for node, nres in result.items():
3790
      msg = nres.RemoteFailMsg()
3791
      if msg:
3792
        raise errors.OpExecError("Cannot disconnect disks node %s,"
3793
                                 " error %s" % (node, msg))
3794

    
3795
  def _GoReconnect(self, multimaster):
3796
    """Reconnect to the network.
3797

3798
    """
3799
    if multimaster:
3800
      msg = "dual-master"
3801
    else:
3802
      msg = "single-master"
3803
    self.feedback_fn("* changing disks into %s mode" % msg)
3804
    result = self.rpc.call_drbd_attach_net(self.all_nodes, self.nodes_ip,
3805
                                           self.instance.disks,
3806
                                           self.instance.name, multimaster)
3807
    for node, nres in result.items():
3808
      msg = nres.RemoteFailMsg()
3809
      if msg:
3810
        raise errors.OpExecError("Cannot change disks config on node %s,"
3811
                                 " error: %s" % (node, msg))
3812

    
3813
  def _ExecCleanup(self):
3814
    """Try to cleanup after a failed migration.
3815

3816
    The cleanup is done by:
3817
      - check that the instance is running only on one node
3818
        (and update the config if needed)
3819
      - change disks on its secondary node to secondary
3820
      - wait until disks are fully synchronized
3821
      - disconnect from the network
3822
      - change disks into single-master mode
3823
      - wait again until disks are fully synchronized
3824

3825
    """
3826
    instance = self.instance
3827
    target_node = self.target_node
3828
    source_node = self.source_node
3829

    
3830
    # check running on only one node
3831
    self.feedback_fn("* checking where the instance actually runs"
3832
                     " (if this hangs, the hypervisor might be in"
3833
                     " a bad state)")
3834
    ins_l = self.rpc.call_instance_list(self.all_nodes, [instance.hypervisor])
3835
    for node, result in ins_l.items():
3836
      result.Raise()
3837
      if not isinstance(result.data, list):
3838
        raise errors.OpExecError("Can't contact node '%s'" % node)
3839

    
3840
    runningon_source = instance.name in ins_l[source_node].data
3841
    runningon_target = instance.name in ins_l[target_node].data
3842

    
3843
    if runningon_source and runningon_target:
3844
      raise errors.OpExecError("Instance seems to be running on two nodes,"
3845
                               " or the hypervisor is confused. You will have"
3846
                               " to ensure manually that it runs only on one"
3847
                               " and restart this operation.")
3848

    
3849
    if not (runningon_source or runningon_target):
3850
      raise errors.OpExecError("Instance does not seem to be running at all."
3851
                               " In this case, it's safer to repair by"
3852
                               " running 'gnt-instance stop' to ensure disk"
3853
                               " shutdown, and then restarting it.")
3854

    
3855
    if runningon_target:
3856
      # the migration has actually succeeded, we need to update the config
3857
      self.feedback_fn("* instance running on secondary node (%s),"
3858
                       " updating config" % target_node)
3859
      instance.primary_node = target_node
3860
      self.cfg.Update(instance)
3861
      demoted_node = source_node
3862
    else:
3863
      self.feedback_fn("* instance confirmed to be running on its"
3864
                       " primary node (%s)" % source_node)
3865
      demoted_node = target_node
3866

    
3867
    self._EnsureSecondary(demoted_node)
3868
    try:
3869
      self._WaitUntilSync()
3870
    except errors.OpExecError:
3871
      # we ignore here errors, since if the device is standalone, it
3872
      # won't be able to sync
3873
      pass
3874
    self._GoStandalone()
3875
    self._GoReconnect(False)
3876
    self._WaitUntilSync()
3877

    
3878
    self.feedback_fn("* done")
3879

    
3880
  def _RevertDiskStatus(self):
3881
    """Try to revert the disk status after a failed migration.
3882

3883
    """
3884
    target_node = self.target_node
3885
    try:
3886
      self._EnsureSecondary(target_node)
3887
      self._GoStandalone()
3888
      self._GoReconnect(False)
3889
      self._WaitUntilSync()
3890
    except errors.OpExecError, err:
3891
      self.LogWarning("Migration failed and I can't reconnect the"
3892
                      " drives: error '%s'\n"
3893
                      "Please look and recover the instance status" %
3894
                      str(err))
3895

    
3896
  def _AbortMigration(self):
3897
    """Call the hypervisor code to abort a started migration.
3898

3899
    """
3900
    instance = self.instance
3901
    target_node = self.target_node
3902
    migration_info = self.migration_info
3903

    
3904
    abort_result = self.rpc.call_finalize_migration(target_node,
3905
                                                    instance,
3906
                                                    migration_info,
3907
                                                    False)
3908
    abort_msg = abort_result.RemoteFailMsg()
3909
    if abort_msg:
3910
      logging.error("Aborting migration failed on target node %s: %s" %
3911
                    (target_node, abort_msg))
3912
      # Don't raise an exception here, as we stil have to try to revert the
3913
      # disk status, even if this step failed.
3914

    
3915
  def _ExecMigration(self):
3916
    """Migrate an instance.
3917

3918
    The migrate is done by:
3919
      - change the disks into dual-master mode
3920
      - wait until disks are fully synchronized again
3921
      - migrate the instance
3922
      - change disks on the new secondary node (the old primary) to secondary
3923
      - wait until disks are fully synchronized
3924
      - change disks into single-master mode
3925

3926
    """
3927
    instance = self.instance
3928
    target_node = self.target_node
3929
    source_node = self.source_node
3930

    
3931
    self.feedback_fn("* checking disk consistency between source and target")
3932
    for dev in instance.disks:
3933
      if not _CheckDiskConsistency(self, dev, target_node, False):
3934
        raise errors.OpExecError("Disk %s is degraded or not fully"
3935
                                 " synchronized on target node,"
3936
                                 " aborting migrate." % dev.iv_name)
3937

    
3938
    # First get the migration information from the remote node
3939
    result = self.rpc.call_migration_info(source_node, instance)
3940
    msg = result.RemoteFailMsg()
3941
    if msg:
3942
      log_err = ("Failed fetching source migration information from %s: %s" %
3943
                 (source_node, msg))
3944
      logging.error(log_err)
3945
      raise errors.OpExecError(log_err)
3946

    
3947
    self.migration_info = migration_info = result.payload
3948

    
3949
    # Then switch the disks to master/master mode
3950
    self._EnsureSecondary(target_node)
3951
    self._GoStandalone()
3952
    self._GoReconnect(True)
3953
    self._WaitUntilSync()
3954

    
3955
    self.feedback_fn("* preparing %s to accept the instance" % target_node)
3956
    result = self.rpc.call_accept_instance(target_node,
3957
                                           instance,
3958
                                           migration_info,
3959
                                           self.nodes_ip[target_node])
3960

    
3961
    msg = result.RemoteFailMsg()
3962
    if msg:
3963
      logging.error("Instance pre-migration failed, trying to revert"
3964
                    " disk status: %s", msg)
3965
      self._AbortMigration()
3966
      self._RevertDiskStatus()
3967
      raise errors.OpExecError("Could not pre-migrate instance %s: %s" %
3968
                               (instance.name, msg))
3969

    
3970
    self.feedback_fn("* migrating instance to %s" % target_node)
3971
    time.sleep(10)
3972
    result = self.rpc.call_instance_migrate(source_node, instance,
3973
                                            self.nodes_ip[target_node],
3974
                                            self.op.live)
3975
    msg = result.RemoteFailMsg()
3976
    if msg:
3977
      logging.error("Instance migration failed, trying to revert"
3978
                    " disk status: %s", msg)
3979
      self._AbortMigration()
3980
      self._RevertDiskStatus()
3981
      raise errors.OpExecError("Could not migrate instance %s: %s" %
3982
                               (instance.name, msg))
3983