Statistics
| Branch: | Tag: | Revision:

root / lib / cmdlib.py @ 39a02558

History | View | Annotate | Download (245 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: non master-candidate has old/wrong file"
755
                        " '%s'" % 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

    
1560
    self.cfg.Update(self.cluster)
1561

    
1562
    # we want to update nodes after the cluster so that if any errors
1563
    # happen, we have recorded and saved the cluster info
1564
    if self.op.candidate_pool_size is not None:
1565
      _AdjustCandidatePool(self)
1566

    
1567

    
1568
class LURedistributeConfig(NoHooksLU):
1569
  """Force the redistribution of cluster configuration.
1570

1571
  This is a very simple LU.
1572

1573
  """
1574
  _OP_REQP = []
1575
  REQ_BGL = False
1576

    
1577
  def ExpandNames(self):
1578
    self.needed_locks = {
1579
      locking.LEVEL_NODE: locking.ALL_SET,
1580
    }
1581
    self.share_locks[locking.LEVEL_NODE] = 1
1582

    
1583
  def CheckPrereq(self):
1584
    """Check prerequisites.
1585

1586
    """
1587

    
1588
  def Exec(self, feedback_fn):
1589
    """Redistribute the configuration.
1590

1591
    """
1592
    self.cfg.Update(self.cfg.GetClusterInfo())
1593

    
1594

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

1598
  """
1599
  if not instance.disks:
1600
    return True
1601

    
1602
  if not oneshot:
1603
    lu.proc.LogInfo("Waiting for instance %s to sync disks." % instance.name)
1604

    
1605
  node = instance.primary_node
1606

    
1607
  for dev in instance.disks:
1608
    lu.cfg.SetDiskID(dev, node)
1609

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

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

    
1654
    if done or oneshot:
1655
      break
1656

    
1657
    time.sleep(min(60, max_time))
1658

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

    
1663

    
1664
def _CheckDiskConsistency(lu, dev, node, on_primary, ldisk=False):
1665
  """Check that mirrors are not degraded.
1666

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

1671
  """
1672
  lu.cfg.SetDiskID(dev, node)
1673
  if ldisk:
1674
    idx = 6
1675
  else:
1676
    idx = 5
1677

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

    
1694
  return result
1695

    
1696

    
1697
class LUDiagnoseOS(NoHooksLU):
1698
  """Logical unit for OS diagnose/query.
1699

1700
  """
1701
  _OP_REQP = ["output_fields", "names"]
1702
  REQ_BGL = False
1703
  _FIELDS_STATIC = utils.FieldSet()
1704
  _FIELDS_DYNAMIC = utils.FieldSet("name", "valid", "node_status")
1705

    
1706
  def ExpandNames(self):
1707
    if self.op.names:
1708
      raise errors.OpPrereqError("Selective OS query not supported")
1709

    
1710
    _CheckOutputFields(static=self._FIELDS_STATIC,
1711
                       dynamic=self._FIELDS_DYNAMIC,
1712
                       selected=self.op.output_fields)
1713

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

    
1721
  def CheckPrereq(self):
1722
    """Check prerequisites.
1723

1724
    """
1725

    
1726
  @staticmethod
1727
  def _DiagnoseByOS(node_list, rlist):
1728
    """Remaps a per-node return list into an a per-os per-node dictionary
1729

1730
    @param node_list: a list with the names of all nodes
1731
    @param rlist: a map with node names as keys and OS objects as values
1732

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

1737
          {"debian-etch": {"node1": [<object>,...],
1738
                           "node2": [<object>,]}
1739
          }
1740

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

    
1761
  def Exec(self, feedback_fn):
1762
    """Compute the list of OSes.
1763

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

    
1787
    return output
1788

    
1789

    
1790
class LURemoveNode(LogicalUnit):
1791
  """Logical unit for removing a node.
1792

1793
  """
1794
  HPATH = "node-remove"
1795
  HTYPE = constants.HTYPE_NODE
1796
  _OP_REQP = ["node_name"]
1797

    
1798
  def BuildHooksEnv(self):
1799
    """Build hooks env.
1800

1801
    This doesn't run on the target node in the pre phase as a failed
1802
    node would then be impossible to remove.
1803

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

    
1813
  def CheckPrereq(self):
1814
    """Check prerequisites.
1815

1816
    This checks:
1817
     - the node exists in the configuration
1818
     - it does not have primary or secondary instances
1819
     - it's not the master
1820

1821
    Any errors are signalled by raising errors.OpPrereqError.
1822

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

    
1828
    instance_list = self.cfg.GetInstanceList()
1829

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

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

    
1843
  def Exec(self, feedback_fn):
1844
    """Removes the node from the cluster.
1845

1846
    """
1847
    node = self.node
1848
    logging.info("Stopping the node daemon and removing configs from node %s",
1849
                 node.name)
1850

    
1851
    self.context.RemoveNode(node.name)
1852

    
1853
    self.rpc.call_node_leave_cluster(node.name)
1854

    
1855
    # Promote nodes to master candidate as needed
1856
    _AdjustCandidatePool(self)
1857

    
1858

    
1859
class LUQueryNodes(NoHooksLU):
1860
  """Logical unit for querying nodes.
1861

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

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

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

    
1888
    self.needed_locks = {}
1889
    self.share_locks[locking.LEVEL_NODE] = 1
1890

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

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

    
1902

    
1903
  def CheckPrereq(self):
1904
    """Check prerequisites.
1905

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

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

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

    
1927
    nodenames = utils.NiceSort(nodenames)
1928
    nodelist = [all_info[name] for name in nodenames]
1929

    
1930
    # begin data gathering
1931

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

    
1957
    node_to_primary = dict([(name, set()) for name in nodenames])
1958
    node_to_secondary = dict([(name, set()) for name in nodenames])
1959

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

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

    
1973
    master_node = self.cfg.GetMasterNode()
1974

    
1975
    # end data gathering
1976

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

    
2014
    return output
2015

    
2016

    
2017
class LUQueryNodeVolumes(NoHooksLU):
2018
  """Logical unit for getting volumes on node(s).
2019

2020
  """
2021
  _OP_REQP = ["nodes", "output_fields"]
2022
  REQ_BGL = False
2023
  _FIELDS_DYNAMIC = utils.FieldSet("phys", "vg", "name", "size", "instance")
2024
  _FIELDS_STATIC = utils.FieldSet("node")
2025

    
2026
  def ExpandNames(self):
2027
    _CheckOutputFields(static=self._FIELDS_STATIC,
2028
                       dynamic=self._FIELDS_DYNAMIC,
2029
                       selected=self.op.output_fields)
2030

    
2031
    self.needed_locks = {}
2032
    self.share_locks[locking.LEVEL_NODE] = 1
2033
    if not self.op.nodes:
2034
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
2035
    else:
2036
      self.needed_locks[locking.LEVEL_NODE] = \
2037
        _GetWantedNodes(self, self.op.nodes)
2038

    
2039
  def CheckPrereq(self):
2040
    """Check prerequisites.
2041

2042
    This checks that the fields required are valid output fields.
2043

2044
    """
2045
    self.nodes = self.acquired_locks[locking.LEVEL_NODE]
2046

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

2050
    """
2051
    nodenames = self.nodes
2052
    volumes = self.rpc.call_node_volumes(nodenames)
2053

    
2054
    ilist = [self.cfg.GetInstanceInfo(iname) for iname
2055
             in self.cfg.GetInstanceList()]
2056

    
2057
    lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
2058

    
2059
    output = []
2060
    for node in nodenames:
2061
      if node not in volumes or volumes[node].failed or not volumes[node].data:
2062
        continue
2063

    
2064
      node_vols = volumes[node].data[:]
2065
      node_vols.sort(key=lambda vol: vol['dev'])
2066

    
2067
      for vol in node_vols:
2068
        node_output = []
2069
        for field in self.op.output_fields:
2070
          if field == "node":
2071
            val = node
2072
          elif field == "phys":
2073
            val = vol['dev']
2074
          elif field == "vg":
2075
            val = vol['vg']
2076
          elif field == "name":
2077
            val = vol['name']
2078
          elif field == "size":
2079
            val = int(float(vol['size']))
2080
          elif field == "instance":
2081
            for inst in ilist:
2082
              if node not in lv_by_node[inst]:
2083
                continue
2084
              if vol['name'] in lv_by_node[inst][node]:
2085
                val = inst.name
2086
                break
2087
            else:
2088
              val = '-'
2089
          else:
2090
            raise errors.ParameterError(field)
2091
          node_output.append(str(val))
2092

    
2093
        output.append(node_output)
2094

    
2095
    return output
2096

    
2097

    
2098
class LUAddNode(LogicalUnit):
2099
  """Logical unit for adding node to the cluster.
2100

2101
  """
2102
  HPATH = "node-add"
2103
  HTYPE = constants.HTYPE_NODE
2104
  _OP_REQP = ["node_name"]
2105

    
2106
  def BuildHooksEnv(self):
2107
    """Build hooks env.
2108

2109
    This will run on all nodes before, and on all nodes + the new node after.
2110

2111
    """
2112
    env = {
2113
      "OP_TARGET": self.op.node_name,
2114
      "NODE_NAME": self.op.node_name,
2115
      "NODE_PIP": self.op.primary_ip,
2116
      "NODE_SIP": self.op.secondary_ip,
2117
      }
2118
    nodes_0 = self.cfg.GetNodeList()
2119
    nodes_1 = nodes_0 + [self.op.node_name, ]
2120
    return env, nodes_0, nodes_1
2121

    
2122
  def CheckPrereq(self):
2123
    """Check prerequisites.
2124

2125
    This checks:
2126
     - the new node is not already in the config
2127
     - it is resolvable
2128
     - its parameters (single/dual homed) matches the cluster
2129

2130
    Any errors are signalled by raising errors.OpPrereqError.
2131

2132
    """
2133
    node_name = self.op.node_name
2134
    cfg = self.cfg
2135

    
2136
    dns_data = utils.HostInfo(node_name)
2137

    
2138
    node = dns_data.name
2139
    primary_ip = self.op.primary_ip = dns_data.ip
2140
    secondary_ip = getattr(self.op, "secondary_ip", None)
2141
    if secondary_ip is None:
2142
      secondary_ip = primary_ip
2143
    if not utils.IsValidIP(secondary_ip):
2144
      raise errors.OpPrereqError("Invalid secondary IP given")
2145
    self.op.secondary_ip = secondary_ip
2146

    
2147
    node_list = cfg.GetNodeList()
2148
    if not self.op.readd and node in node_list:
2149
      raise errors.OpPrereqError("Node %s is already in the configuration" %
2150
                                 node)
2151
    elif self.op.readd and node not in node_list:
2152
      raise errors.OpPrereqError("Node %s is not in the configuration" % node)
2153

    
2154
    for existing_node_name in node_list:
2155
      existing_node = cfg.GetNodeInfo(existing_node_name)
2156

    
2157
      if self.op.readd and node == existing_node_name:
2158
        if (existing_node.primary_ip != primary_ip or
2159
            existing_node.secondary_ip != secondary_ip):
2160
          raise errors.OpPrereqError("Readded node doesn't have the same IP"
2161
                                     " address configuration as before")
2162
        continue
2163

    
2164
      if (existing_node.primary_ip == primary_ip or
2165
          existing_node.secondary_ip == primary_ip or
2166
          existing_node.primary_ip == secondary_ip or
2167
          existing_node.secondary_ip == secondary_ip):
2168
        raise errors.OpPrereqError("New node ip address(es) conflict with"
2169
                                   " existing node %s" % existing_node.name)
2170

    
2171
    # check that the type of the node (single versus dual homed) is the
2172
    # same as for the master
2173
    myself = cfg.GetNodeInfo(self.cfg.GetMasterNode())
2174
    master_singlehomed = myself.secondary_ip == myself.primary_ip
2175
    newbie_singlehomed = secondary_ip == primary_ip
2176
    if master_singlehomed != newbie_singlehomed:
2177
      if master_singlehomed:
2178
        raise errors.OpPrereqError("The master has no private ip but the"
2179
                                   " new node has one")
2180
      else:
2181
        raise errors.OpPrereqError("The master has a private ip but the"
2182
                                   " new node doesn't have one")
2183

    
2184
    # checks reachablity
2185
    if not utils.TcpPing(primary_ip, constants.DEFAULT_NODED_PORT):
2186
      raise errors.OpPrereqError("Node not reachable by ping")
2187

    
2188
    if not newbie_singlehomed:
2189
      # check reachability from my secondary ip to newbie's secondary ip
2190
      if not utils.TcpPing(secondary_ip, constants.DEFAULT_NODED_PORT,
2191
                           source=myself.secondary_ip):
2192
        raise errors.OpPrereqError("Node secondary ip not reachable by TCP"
2193
                                   " based ping to noded port")
2194

    
2195
    cp_size = self.cfg.GetClusterInfo().candidate_pool_size
2196
    mc_now, _ = self.cfg.GetMasterCandidateStats()
2197
    master_candidate = mc_now < cp_size
2198

    
2199
    self.new_node = objects.Node(name=node,
2200
                                 primary_ip=primary_ip,
2201
                                 secondary_ip=secondary_ip,
2202
                                 master_candidate=master_candidate,
2203
                                 offline=False, drained=False)
2204

    
2205
  def Exec(self, feedback_fn):
2206
    """Adds the new node to the cluster.
2207

2208
    """
2209
    new_node = self.new_node
2210
    node = new_node.name
2211

    
2212
    # check connectivity
2213
    result = self.rpc.call_version([node])[node]
2214
    result.Raise()
2215
    if result.data:
2216
      if constants.PROTOCOL_VERSION == result.data:
2217
        logging.info("Communication to node %s fine, sw version %s match",
2218
                     node, result.data)
2219
      else:
2220
        raise errors.OpExecError("Version mismatch master version %s,"
2221
                                 " node version %s" %
2222
                                 (constants.PROTOCOL_VERSION, result.data))
2223
    else:
2224
      raise errors.OpExecError("Cannot get version from the new node")
2225

    
2226
    # setup ssh on node
2227
    logging.info("Copy ssh key to node %s", node)
2228
    priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
2229
    keyarray = []
2230
    keyfiles = [constants.SSH_HOST_DSA_PRIV, constants.SSH_HOST_DSA_PUB,
2231
                constants.SSH_HOST_RSA_PRIV, constants.SSH_HOST_RSA_PUB,
2232
                priv_key, pub_key]
2233

    
2234
    for i in keyfiles:
2235
      f = open(i, 'r')
2236
      try:
2237
        keyarray.append(f.read())
2238
      finally:
2239
        f.close()
2240

    
2241
    result = self.rpc.call_node_add(node, keyarray[0], keyarray[1],
2242
                                    keyarray[2],
2243
                                    keyarray[3], keyarray[4], keyarray[5])
2244

    
2245
    msg = result.RemoteFailMsg()
2246
    if msg:
2247
      raise errors.OpExecError("Cannot transfer ssh keys to the"
2248
                               " new node: %s" % msg)
2249

    
2250
    # Add node to our /etc/hosts, and add key to known_hosts
2251
    utils.AddHostToEtcHosts(new_node.name)
2252

    
2253
    if new_node.secondary_ip != new_node.primary_ip:
2254
      result = self.rpc.call_node_has_ip_address(new_node.name,
2255
                                                 new_node.secondary_ip)
2256
      if result.failed or not result.data:
2257
        raise errors.OpExecError("Node claims it doesn't have the secondary ip"
2258
                                 " you gave (%s). Please fix and re-run this"
2259
                                 " command." % new_node.secondary_ip)
2260

    
2261
    node_verify_list = [self.cfg.GetMasterNode()]
2262
    node_verify_param = {
2263
      'nodelist': [node],
2264
      # TODO: do a node-net-test as well?
2265
    }
2266

    
2267
    result = self.rpc.call_node_verify(node_verify_list, node_verify_param,
2268
                                       self.cfg.GetClusterName())
2269
    for verifier in node_verify_list:
2270
      if result[verifier].failed or not result[verifier].data:
2271
        raise errors.OpExecError("Cannot communicate with %s's node daemon"
2272
                                 " for remote verification" % verifier)
2273
      if result[verifier].data['nodelist']:
2274
        for failed in result[verifier].data['nodelist']:
2275
          feedback_fn("ssh/hostname verification failed %s -> %s" %
2276
                      (verifier, result[verifier].data['nodelist'][failed]))
2277
        raise errors.OpExecError("ssh/hostname verification failed.")
2278

    
2279
    # Distribute updated /etc/hosts and known_hosts to all nodes,
2280
    # including the node just added
2281
    myself = self.cfg.GetNodeInfo(self.cfg.GetMasterNode())
2282
    dist_nodes = self.cfg.GetNodeList()
2283
    if not self.op.readd:
2284
      dist_nodes.append(node)
2285
    if myself.name in dist_nodes:
2286
      dist_nodes.remove(myself.name)
2287

    
2288
    logging.debug("Copying hosts and known_hosts to all nodes")
2289
    for fname in (constants.ETC_HOSTS, constants.SSH_KNOWN_HOSTS_FILE):
2290
      result = self.rpc.call_upload_file(dist_nodes, fname)
2291
      for to_node, to_result in result.iteritems():
2292
        if to_result.failed or not to_result.data:
2293
          logging.error("Copy of file %s to node %s failed", fname, to_node)
2294

    
2295
    to_copy = []
2296
    enabled_hypervisors = self.cfg.GetClusterInfo().enabled_hypervisors
2297
    if constants.HTS_COPY_VNC_PASSWORD.intersection(enabled_hypervisors):
2298
      to_copy.append(constants.VNC_PASSWORD_FILE)
2299

    
2300
    for fname in to_copy:
2301
      result = self.rpc.call_upload_file([node], fname)
2302
      if result[node].failed or not result[node]:
2303
        logging.error("Could not copy file %s to node %s", fname, node)
2304

    
2305
    if self.op.readd:
2306
      self.context.ReaddNode(new_node)
2307
    else:
2308
      self.context.AddNode(new_node)
2309

    
2310

    
2311
class LUSetNodeParams(LogicalUnit):
2312
  """Modifies the parameters of a node.
2313

2314
  """
2315
  HPATH = "node-modify"
2316
  HTYPE = constants.HTYPE_NODE
2317
  _OP_REQP = ["node_name"]
2318
  REQ_BGL = False
2319

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

    
2335
  def ExpandNames(self):
2336
    self.needed_locks = {locking.LEVEL_NODE: self.op.node_name}
2337

    
2338
  def BuildHooksEnv(self):
2339
    """Build hooks env.
2340

2341
    This runs on the master node.
2342

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

    
2354
  def CheckPrereq(self):
2355
    """Check prerequisites.
2356

2357
    This only checks the instance list against the existing names.
2358

2359
    """
2360
    node = self.node = self.cfg.GetNodeInfo(self.op.node_name)
2361

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

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

    
2384
    return
2385

    
2386
  def Exec(self, feedback_fn):
2387
    """Modifies a node.
2388

2389
    """
2390
    node = self.node
2391

    
2392
    result = []
2393
    changed_mc = False
2394

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

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

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

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

    
2435
    return result
2436

    
2437

    
2438
class LUQueryClusterInfo(NoHooksLU):
2439
  """Query cluster configuration.
2440

2441
  """
2442
  _OP_REQP = []
2443
  REQ_BGL = False
2444

    
2445
  def ExpandNames(self):
2446
    self.needed_locks = {}
2447

    
2448
  def CheckPrereq(self):
2449
    """No prerequsites needed for this LU.
2450

2451
    """
2452
    pass
2453

    
2454
  def Exec(self, feedback_fn):
2455
    """Return cluster config.
2456

2457
    """
2458
    cluster = self.cfg.GetClusterInfo()
2459
    result = {
2460
      "software_version": constants.RELEASE_VERSION,
2461
      "protocol_version": constants.PROTOCOL_VERSION,
2462
      "config_version": constants.CONFIG_VERSION,
2463
      "os_api_version": constants.OS_API_VERSION,
2464
      "export_version": constants.EXPORT_VERSION,
2465
      "architecture": (platform.architecture()[0], platform.machine()),
2466
      "name": cluster.cluster_name,
2467
      "master": cluster.master_node,
2468
      "default_hypervisor": cluster.default_hypervisor,
2469
      "enabled_hypervisors": cluster.enabled_hypervisors,
2470
      "hvparams": dict([(hypervisor, cluster.hvparams[hypervisor])
2471
                        for hypervisor in cluster.enabled_hypervisors]),
2472
      "beparams": cluster.beparams,
2473
      "candidate_pool_size": cluster.candidate_pool_size,
2474
      "default_bridge": cluster.default_bridge,
2475
      "master_netdev": cluster.master_netdev,
2476
      "volume_group_name": cluster.volume_group_name,
2477
      "file_storage_dir": cluster.file_storage_dir,
2478
      }
2479

    
2480
    return result
2481

    
2482

    
2483
class LUQueryConfigValues(NoHooksLU):
2484
  """Return configuration values.
2485

2486
  """
2487
  _OP_REQP = []
2488
  REQ_BGL = False
2489
  _FIELDS_DYNAMIC = utils.FieldSet()
2490
  _FIELDS_STATIC = utils.FieldSet("cluster_name", "master_node", "drain_flag")
2491

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

    
2495
    _CheckOutputFields(static=self._FIELDS_STATIC,
2496
                       dynamic=self._FIELDS_DYNAMIC,
2497
                       selected=self.op.output_fields)
2498

    
2499
  def CheckPrereq(self):
2500
    """No prerequisites.
2501

2502
    """
2503
    pass
2504

    
2505
  def Exec(self, feedback_fn):
2506
    """Dump a representation of the cluster config to the standard output.
2507

2508
    """
2509
    values = []
2510
    for field in self.op.output_fields:
2511
      if field == "cluster_name":
2512
        entry = self.cfg.GetClusterName()
2513
      elif field == "master_node":
2514
        entry = self.cfg.GetMasterNode()
2515
      elif field == "drain_flag":
2516
        entry = os.path.exists(constants.JOB_QUEUE_DRAIN_FILE)
2517
      else:
2518
        raise errors.ParameterError(field)
2519
      values.append(entry)
2520
    return values
2521

    
2522

    
2523
class LUActivateInstanceDisks(NoHooksLU):
2524
  """Bring up an instance's disks.
2525

2526
  """
2527
  _OP_REQP = ["instance_name"]
2528
  REQ_BGL = False
2529

    
2530
  def ExpandNames(self):
2531
    self._ExpandAndLockInstance()
2532
    self.needed_locks[locking.LEVEL_NODE] = []
2533
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2534

    
2535
  def DeclareLocks(self, level):
2536
    if level == locking.LEVEL_NODE:
2537
      self._LockInstancesNodes()
2538

    
2539
  def CheckPrereq(self):
2540
    """Check prerequisites.
2541

2542
    This checks that the instance is in the cluster.
2543

2544
    """
2545
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2546
    assert self.instance is not None, \
2547
      "Cannot retrieve locked instance %s" % self.op.instance_name
2548
    _CheckNodeOnline(self, self.instance.primary_node)
2549

    
2550
  def Exec(self, feedback_fn):
2551
    """Activate the disks.
2552

2553
    """
2554
    disks_ok, disks_info = _AssembleInstanceDisks(self, self.instance)
2555
    if not disks_ok:
2556
      raise errors.OpExecError("Cannot activate block devices")
2557

    
2558
    return disks_info
2559

    
2560

    
2561
def _AssembleInstanceDisks(lu, instance, ignore_secondaries=False):
2562
  """Prepare the block devices for an instance.
2563

2564
  This sets up the block devices on all nodes.
2565

2566
  @type lu: L{LogicalUnit}
2567
  @param lu: the logical unit on whose behalf we execute
2568
  @type instance: L{objects.Instance}
2569
  @param instance: the instance for whose disks we assemble
2570
  @type ignore_secondaries: boolean
2571
  @param ignore_secondaries: if true, errors on secondary nodes
2572
      won't result in an error return from the function
2573
  @return: False if the operation failed, otherwise a list of
2574
      (host, instance_visible_name, node_visible_name)
2575
      with the mapping from node devices to instance devices
2576

2577
  """
2578
  device_info = []
2579
  disks_ok = True
2580
  iname = instance.name
2581
  # With the two passes mechanism we try to reduce the window of
2582
  # opportunity for the race condition of switching DRBD to primary
2583
  # before handshaking occured, but we do not eliminate it
2584

    
2585
  # The proper fix would be to wait (with some limits) until the
2586
  # connection has been made and drbd transitions from WFConnection
2587
  # into any other network-connected state (Connected, SyncTarget,
2588
  # SyncSource, etc.)
2589

    
2590
  # 1st pass, assemble on all nodes in secondary mode
2591
  for inst_disk in instance.disks:
2592
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
2593
      lu.cfg.SetDiskID(node_disk, node)
2594
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, False)
2595
      msg = result.RemoteFailMsg()
2596
      if msg:
2597
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
2598
                           " (is_primary=False, pass=1): %s",
2599
                           inst_disk.iv_name, node, msg)
2600
        if not ignore_secondaries:
2601
          disks_ok = False
2602

    
2603
  # FIXME: race condition on drbd migration to primary
2604

    
2605
  # 2nd pass, do only the primary node
2606
  for inst_disk in instance.disks:
2607
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
2608
      if node != instance.primary_node:
2609
        continue
2610
      lu.cfg.SetDiskID(node_disk, node)
2611
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, True)
2612
      msg = result.RemoteFailMsg()
2613
      if msg:
2614
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
2615
                           " (is_primary=True, pass=2): %s",
2616
                           inst_disk.iv_name, node, msg)
2617
        disks_ok = False
2618
    device_info.append((instance.primary_node, inst_disk.iv_name,
2619
                        result.payload))
2620

    
2621
  # leave the disks configured for the primary node
2622
  # this is a workaround that would be fixed better by
2623
  # improving the logical/physical id handling
2624
  for disk in instance.disks:
2625
    lu.cfg.SetDiskID(disk, instance.primary_node)
2626

    
2627
  return disks_ok, device_info
2628

    
2629

    
2630
def _StartInstanceDisks(lu, instance, force):
2631
  """Start the disks of an instance.
2632

2633
  """
2634
  disks_ok, dummy = _AssembleInstanceDisks(lu, instance,
2635
                                           ignore_secondaries=force)
2636
  if not disks_ok:
2637
    _ShutdownInstanceDisks(lu, instance)
2638
    if force is not None and not force:
2639
      lu.proc.LogWarning("", hint="If the message above refers to a"
2640
                         " secondary node,"
2641
                         " you can retry the operation using '--force'.")
2642
    raise errors.OpExecError("Disk consistency error")
2643

    
2644

    
2645
class LUDeactivateInstanceDisks(NoHooksLU):
2646
  """Shutdown an instance's disks.
2647

2648
  """
2649
  _OP_REQP = ["instance_name"]
2650
  REQ_BGL = False
2651

    
2652
  def ExpandNames(self):
2653
    self._ExpandAndLockInstance()
2654
    self.needed_locks[locking.LEVEL_NODE] = []
2655
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2656

    
2657
  def DeclareLocks(self, level):
2658
    if level == locking.LEVEL_NODE:
2659
      self._LockInstancesNodes()
2660

    
2661
  def CheckPrereq(self):
2662
    """Check prerequisites.
2663

2664
    This checks that the instance is in the cluster.
2665

2666
    """
2667
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2668
    assert self.instance is not None, \
2669
      "Cannot retrieve locked instance %s" % self.op.instance_name
2670

    
2671
  def Exec(self, feedback_fn):
2672
    """Deactivate the disks
2673

2674
    """
2675
    instance = self.instance
2676
    _SafeShutdownInstanceDisks(self, instance)
2677

    
2678

    
2679
def _SafeShutdownInstanceDisks(lu, instance):
2680
  """Shutdown block devices of an instance.
2681

2682
  This function checks if an instance is running, before calling
2683
  _ShutdownInstanceDisks.
2684

2685
  """
2686
  ins_l = lu.rpc.call_instance_list([instance.primary_node],
2687
                                      [instance.hypervisor])
2688
  ins_l = ins_l[instance.primary_node]
2689
  if ins_l.failed or not isinstance(ins_l.data, list):
2690
    raise errors.OpExecError("Can't contact node '%s'" %
2691
                             instance.primary_node)
2692

    
2693
  if instance.name in ins_l.data:
2694
    raise errors.OpExecError("Instance is running, can't shutdown"
2695
                             " block devices.")
2696

    
2697
  _ShutdownInstanceDisks(lu, instance)
2698

    
2699

    
2700
def _ShutdownInstanceDisks(lu, instance, ignore_primary=False):
2701
  """Shutdown block devices of an instance.
2702

2703
  This does the shutdown on all nodes of the instance.
2704

2705
  If the ignore_primary is false, errors on the primary node are
2706
  ignored.
2707

2708
  """
2709
  all_result = True
2710
  for disk in instance.disks:
2711
    for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
2712
      lu.cfg.SetDiskID(top_disk, node)
2713
      result = lu.rpc.call_blockdev_shutdown(node, top_disk)
2714
      msg = result.RemoteFailMsg()
2715
      if msg:
2716
        lu.LogWarning("Could not shutdown block device %s on node %s: %s",
2717
                      disk.iv_name, node, msg)
2718
        if not ignore_primary or node != instance.primary_node:
2719
          all_result = False
2720
  return all_result
2721

    
2722

    
2723
def _CheckNodeFreeMemory(lu, node, reason, requested, hypervisor_name):
2724
  """Checks if a node has enough free memory.
2725

2726
  This function check if a given node has the needed amount of free
2727
  memory. In case the node has less memory or we cannot get the
2728
  information from the node, this function raise an OpPrereqError
2729
  exception.
2730

2731
  @type lu: C{LogicalUnit}
2732
  @param lu: a logical unit from which we get configuration data
2733
  @type node: C{str}
2734
  @param node: the node to check
2735
  @type reason: C{str}
2736
  @param reason: string to use in the error message
2737
  @type requested: C{int}
2738
  @param requested: the amount of memory in MiB to check for
2739
  @type hypervisor_name: C{str}
2740
  @param hypervisor_name: the hypervisor to ask for memory stats
2741
  @raise errors.OpPrereqError: if the node doesn't have enough memory, or
2742
      we cannot check the node
2743

2744
  """
2745
  nodeinfo = lu.rpc.call_node_info([node], lu.cfg.GetVGName(), hypervisor_name)
2746
  nodeinfo[node].Raise()
2747
  free_mem = nodeinfo[node].data.get('memory_free')
2748
  if not isinstance(free_mem, int):
2749
    raise errors.OpPrereqError("Can't compute free memory on node %s, result"
2750
                             " was '%s'" % (node, free_mem))
2751
  if requested > free_mem:
2752
    raise errors.OpPrereqError("Not enough memory on node %s for %s:"
2753
                             " needed %s MiB, available %s MiB" %
2754
                             (node, reason, requested, free_mem))
2755

    
2756

    
2757
class LUStartupInstance(LogicalUnit):
2758
  """Starts an instance.
2759

2760
  """
2761
  HPATH = "instance-start"
2762
  HTYPE = constants.HTYPE_INSTANCE
2763
  _OP_REQP = ["instance_name", "force"]
2764
  REQ_BGL = False
2765

    
2766
  def ExpandNames(self):
2767
    self._ExpandAndLockInstance()
2768

    
2769
  def BuildHooksEnv(self):
2770
    """Build hooks env.
2771

2772
    This runs on master, primary and secondary nodes of the instance.
2773

2774
    """
2775
    env = {
2776
      "FORCE": self.op.force,
2777
      }
2778
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
2779
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2780
    return env, nl, nl
2781

    
2782
  def CheckPrereq(self):
2783
    """Check prerequisites.
2784

2785
    This checks that the instance is in the cluster.
2786

2787
    """
2788
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2789
    assert self.instance is not None, \
2790
      "Cannot retrieve locked instance %s" % self.op.instance_name
2791

    
2792
    # extra beparams
2793
    self.beparams = getattr(self.op, "beparams", {})
2794
    if self.beparams:
2795
      if not isinstance(self.beparams, dict):
2796
        raise errors.OpPrereqError("Invalid beparams passed: %s, expected"
2797
                                   " dict" % (type(self.beparams), ))
2798
      # fill the beparams dict
2799
      utils.ForceDictType(self.beparams, constants.BES_PARAMETER_TYPES)
2800
      self.op.beparams = self.beparams
2801

    
2802
    # extra hvparams
2803
    self.hvparams = getattr(self.op, "hvparams", {})
2804
    if self.hvparams:
2805
      if not isinstance(self.hvparams, dict):
2806
        raise errors.OpPrereqError("Invalid hvparams passed: %s, expected"
2807
                                   " dict" % (type(self.hvparams), ))
2808

    
2809
      # check hypervisor parameter syntax (locally)
2810
      cluster = self.cfg.GetClusterInfo()
2811
      utils.ForceDictType(self.hvparams, constants.HVS_PARAMETER_TYPES)
2812
      filled_hvp = cluster.FillDict(cluster.hvparams[instance.hypervisor],
2813
                                    instance.hvparams)
2814
      filled_hvp.update(self.hvparams)
2815
      hv_type = hypervisor.GetHypervisor(instance.hypervisor)
2816
      hv_type.CheckParameterSyntax(filled_hvp)
2817
      _CheckHVParams(self, instance.all_nodes, instance.hypervisor, filled_hvp)
2818
      self.op.hvparams = self.hvparams
2819

    
2820
    _CheckNodeOnline(self, instance.primary_node)
2821

    
2822
    bep = self.cfg.GetClusterInfo().FillBE(instance)
2823
    # check bridges existance
2824
    _CheckInstanceBridgesExist(self, instance)
2825

    
2826
    remote_info = self.rpc.call_instance_info(instance.primary_node,
2827
                                              instance.name,
2828
                                              instance.hypervisor)
2829
    remote_info.Raise()
2830
    if not remote_info.data:
2831
      _CheckNodeFreeMemory(self, instance.primary_node,
2832
                           "starting instance %s" % instance.name,
2833
                           bep[constants.BE_MEMORY], instance.hypervisor)
2834

    
2835
  def Exec(self, feedback_fn):
2836
    """Start the instance.
2837

2838
    """
2839
    instance = self.instance
2840
    force = self.op.force
2841

    
2842
    self.cfg.MarkInstanceUp(instance.name)
2843

    
2844
    node_current = instance.primary_node
2845

    
2846
    _StartInstanceDisks(self, instance, force)
2847

    
2848
    result = self.rpc.call_instance_start(node_current, instance,
2849
                                          self.hvparams, self.beparams)
2850
    msg = result.RemoteFailMsg()
2851
    if msg:
2852
      _ShutdownInstanceDisks(self, instance)
2853
      raise errors.OpExecError("Could not start instance: %s" % msg)
2854

    
2855

    
2856
class LURebootInstance(LogicalUnit):
2857
  """Reboot an instance.
2858

2859
  """
2860
  HPATH = "instance-reboot"
2861
  HTYPE = constants.HTYPE_INSTANCE
2862
  _OP_REQP = ["instance_name", "ignore_secondaries", "reboot_type"]
2863
  REQ_BGL = False
2864

    
2865
  def ExpandNames(self):
2866
    if self.op.reboot_type not in [constants.INSTANCE_REBOOT_SOFT,
2867
                                   constants.INSTANCE_REBOOT_HARD,
2868
                                   constants.INSTANCE_REBOOT_FULL]:
2869
      raise errors.ParameterError("reboot type not in [%s, %s, %s]" %
2870
                                  (constants.INSTANCE_REBOOT_SOFT,
2871
                                   constants.INSTANCE_REBOOT_HARD,
2872
                                   constants.INSTANCE_REBOOT_FULL))
2873
    self._ExpandAndLockInstance()
2874

    
2875
  def BuildHooksEnv(self):
2876
    """Build hooks env.
2877

2878
    This runs on master, primary and secondary nodes of the instance.
2879

2880
    """
2881
    env = {
2882
      "IGNORE_SECONDARIES": self.op.ignore_secondaries,
2883
      "REBOOT_TYPE": self.op.reboot_type,
2884
      }
2885
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
2886
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2887
    return env, nl, nl
2888

    
2889
  def CheckPrereq(self):
2890
    """Check prerequisites.
2891

2892
    This checks that the instance is in the cluster.
2893

2894
    """
2895
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2896
    assert self.instance is not None, \
2897
      "Cannot retrieve locked instance %s" % self.op.instance_name
2898

    
2899
    _CheckNodeOnline(self, instance.primary_node)
2900

    
2901
    # check bridges existance
2902
    _CheckInstanceBridgesExist(self, instance)
2903

    
2904
  def Exec(self, feedback_fn):
2905
    """Reboot the instance.
2906

2907
    """
2908
    instance = self.instance
2909
    ignore_secondaries = self.op.ignore_secondaries
2910
    reboot_type = self.op.reboot_type
2911

    
2912
    node_current = instance.primary_node
2913

    
2914
    if reboot_type in [constants.INSTANCE_REBOOT_SOFT,
2915
                       constants.INSTANCE_REBOOT_HARD]:
2916
      for disk in instance.disks:
2917
        self.cfg.SetDiskID(disk, node_current)
2918
      result = self.rpc.call_instance_reboot(node_current, instance,
2919
                                             reboot_type)
2920
      msg = result.RemoteFailMsg()
2921
      if msg:
2922
        raise errors.OpExecError("Could not reboot instance: %s" % msg)
2923
    else:
2924
      result = self.rpc.call_instance_shutdown(node_current, instance)
2925
      msg = result.RemoteFailMsg()
2926
      if msg:
2927
        raise errors.OpExecError("Could not shutdown instance for"
2928
                                 " full reboot: %s" % msg)
2929
      _ShutdownInstanceDisks(self, instance)
2930
      _StartInstanceDisks(self, instance, ignore_secondaries)
2931
      result = self.rpc.call_instance_start(node_current, instance, None, None)
2932
      msg = result.RemoteFailMsg()
2933
      if msg:
2934
        _ShutdownInstanceDisks(self, instance)
2935
        raise errors.OpExecError("Could not start instance for"
2936
                                 " full reboot: %s" % msg)
2937

    
2938
    self.cfg.MarkInstanceUp(instance.name)
2939

    
2940

    
2941
class LUShutdownInstance(LogicalUnit):
2942
  """Shutdown an instance.
2943

2944
  """
2945
  HPATH = "instance-stop"
2946
  HTYPE = constants.HTYPE_INSTANCE
2947
  _OP_REQP = ["instance_name"]
2948
  REQ_BGL = False
2949

    
2950
  def ExpandNames(self):
2951
    self._ExpandAndLockInstance()
2952

    
2953
  def BuildHooksEnv(self):
2954
    """Build hooks env.
2955

2956
    This runs on master, primary and secondary nodes of the instance.
2957

2958
    """
2959
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2960
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2961
    return env, nl, nl
2962

    
2963
  def CheckPrereq(self):
2964
    """Check prerequisites.
2965

2966
    This checks that the instance is in the cluster.
2967

2968
    """
2969
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2970
    assert self.instance is not None, \
2971
      "Cannot retrieve locked instance %s" % self.op.instance_name
2972
    _CheckNodeOnline(self, self.instance.primary_node)
2973

    
2974
  def Exec(self, feedback_fn):
2975
    """Shutdown the instance.
2976

2977
    """
2978
    instance = self.instance
2979
    node_current = instance.primary_node
2980
    self.cfg.MarkInstanceDown(instance.name)
2981
    result = self.rpc.call_instance_shutdown(node_current, instance)
2982
    msg = result.RemoteFailMsg()
2983
    if msg:
2984
      self.proc.LogWarning("Could not shutdown instance: %s" % msg)
2985

    
2986
    _ShutdownInstanceDisks(self, instance)
2987

    
2988

    
2989
class LUReinstallInstance(LogicalUnit):
2990
  """Reinstall an instance.
2991

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

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

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

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

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

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

3014
    This checks that the instance is in the cluster and is not running.
3015

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

    
3022
    if instance.disk_template == constants.DT_DISKLESS:
3023
      raise errors.OpPrereqError("Instance '%s' has no disks" %
3024
                                 self.op.instance_name)
3025
    if instance.admin_up:
3026
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
3027
                                 self.op.instance_name)
3028
    remote_info = self.rpc.call_instance_info(instance.primary_node,
3029
                                              instance.name,
3030
                                              instance.hypervisor)
3031
    remote_info.Raise()
3032
    if remote_info.data:
3033
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
3034
                                 (self.op.instance_name,
3035
                                  instance.primary_node))
3036

    
3037
    self.op.os_type = getattr(self.op, "os_type", None)
3038
    if self.op.os_type is not None:
3039
      # OS verification
3040
      pnode = self.cfg.GetNodeInfo(
3041
        self.cfg.ExpandNodeName(instance.primary_node))
3042
      if pnode is None:
3043
        raise errors.OpPrereqError("Primary node '%s' is unknown" %
3044
                                   self.op.pnode)
3045
      result = self.rpc.call_os_get(pnode.name, self.op.os_type)
3046
      result.Raise()
3047
      if not isinstance(result.data, objects.OS):
3048
        raise errors.OpPrereqError("OS '%s' not in supported OS list for"
3049
                                   " primary node"  % self.op.os_type)
3050

    
3051
    self.instance = instance
3052

    
3053
  def Exec(self, feedback_fn):
3054
    """Reinstall the instance.
3055

3056
    """
3057
    inst = self.instance
3058

    
3059
    if self.op.os_type is not None:
3060
      feedback_fn("Changing OS to '%s'..." % self.op.os_type)
3061
      inst.os = self.op.os_type
3062
      self.cfg.Update(inst)
3063

    
3064
    _StartInstanceDisks(self, inst, None)
3065
    try:
3066
      feedback_fn("Running the instance OS create scripts...")
3067
      result = self.rpc.call_instance_os_add(inst.primary_node, inst)
3068
      msg = result.RemoteFailMsg()
3069
      if msg:
3070
        raise errors.OpExecError("Could not install OS for instance %s"
3071
                                 " on node %s: %s" %
3072
                                 (inst.name, inst.primary_node, msg))
3073
    finally:
3074
      _ShutdownInstanceDisks(self, inst)
3075

    
3076

    
3077
class LURenameInstance(LogicalUnit):
3078
  """Rename an instance.
3079

3080
  """
3081
  HPATH = "instance-rename"
3082
  HTYPE = constants.HTYPE_INSTANCE
3083
  _OP_REQP = ["instance_name", "new_name"]
3084

    
3085
  def BuildHooksEnv(self):
3086
    """Build hooks env.
3087

3088
    This runs on master, primary and secondary nodes of the instance.
3089

3090
    """
3091
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3092
    env["INSTANCE_NEW_NAME"] = self.op.new_name
3093
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3094
    return env, nl, nl
3095

    
3096
  def CheckPrereq(self):
3097
    """Check prerequisites.
3098

3099
    This checks that the instance is in the cluster and is not running.
3100

3101
    """
3102
    instance = self.cfg.GetInstanceInfo(
3103
      self.cfg.ExpandInstanceName(self.op.instance_name))
3104
    if instance is None:
3105
      raise errors.OpPrereqError("Instance '%s' not known" %
3106
                                 self.op.instance_name)
3107
    _CheckNodeOnline(self, instance.primary_node)
3108

    
3109
    if instance.admin_up:
3110
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
3111
                                 self.op.instance_name)
3112
    remote_info = self.rpc.call_instance_info(instance.primary_node,
3113
                                              instance.name,
3114
                                              instance.hypervisor)
3115
    remote_info.Raise()
3116
    if remote_info.data:
3117
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
3118
                                 (self.op.instance_name,
3119
                                  instance.primary_node))
3120
    self.instance = instance
3121

    
3122
    # new name verification
3123
    name_info = utils.HostInfo(self.op.new_name)
3124

    
3125
    self.op.new_name = new_name = name_info.name
3126
    instance_list = self.cfg.GetInstanceList()
3127
    if new_name in instance_list:
3128
      raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
3129
                                 new_name)
3130

    
3131
    if not getattr(self.op, "ignore_ip", False):
3132
      if utils.TcpPing(name_info.ip, constants.DEFAULT_NODED_PORT):
3133
        raise errors.OpPrereqError("IP %s of instance %s already in use" %
3134
                                   (name_info.ip, new_name))
3135

    
3136

    
3137
  def Exec(self, feedback_fn):
3138
    """Reinstall the instance.
3139

3140
    """
3141
    inst = self.instance
3142
    old_name = inst.name
3143

    
3144
    if inst.disk_template == constants.DT_FILE:
3145
      old_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
3146

    
3147
    self.cfg.RenameInstance(inst.name, self.op.new_name)
3148
    # Change the instance lock. This is definitely safe while we hold the BGL
3149
    self.context.glm.remove(locking.LEVEL_INSTANCE, old_name)
3150
    self.context.glm.add(locking.LEVEL_INSTANCE, self.op.new_name)
3151

    
3152
    # re-read the instance from the configuration after rename
3153
    inst = self.cfg.GetInstanceInfo(self.op.new_name)
3154

    
3155
    if inst.disk_template == constants.DT_FILE:
3156
      new_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
3157
      result = self.rpc.call_file_storage_dir_rename(inst.primary_node,
3158
                                                     old_file_storage_dir,
3159
                                                     new_file_storage_dir)
3160
      result.Raise()
3161
      if not result.data:
3162
        raise errors.OpExecError("Could not connect to node '%s' to rename"
3163
                                 " directory '%s' to '%s' (but the instance"
3164
                                 " has been renamed in Ganeti)" % (
3165
                                 inst.primary_node, old_file_storage_dir,
3166
                                 new_file_storage_dir))
3167

    
3168
      if not result.data[0]:
3169
        raise errors.OpExecError("Could not rename directory '%s' to '%s'"
3170
                                 " (but the instance has been renamed in"
3171
                                 " Ganeti)" % (old_file_storage_dir,
3172
                                               new_file_storage_dir))
3173

    
3174
    _StartInstanceDisks(self, inst, None)
3175
    try:
3176
      result = self.rpc.call_instance_run_rename(inst.primary_node, inst,
3177
                                                 old_name)
3178
      msg = result.RemoteFailMsg()
3179
      if msg:
3180
        msg = ("Could not run OS rename script for instance %s on node %s"
3181
               " (but the instance has been renamed in Ganeti): %s" %
3182
               (inst.name, inst.primary_node, msg))
3183
        self.proc.LogWarning(msg)
3184
    finally:
3185
      _ShutdownInstanceDisks(self, inst)
3186

    
3187

    
3188
class LURemoveInstance(LogicalUnit):
3189
  """Remove an instance.
3190

3191
  """
3192
  HPATH = "instance-remove"
3193
  HTYPE = constants.HTYPE_INSTANCE
3194
  _OP_REQP = ["instance_name", "ignore_failures"]
3195
  REQ_BGL = False
3196

    
3197
  def ExpandNames(self):
3198
    self._ExpandAndLockInstance()
3199
    self.needed_locks[locking.LEVEL_NODE] = []
3200
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3201

    
3202
  def DeclareLocks(self, level):
3203
    if level == locking.LEVEL_NODE:
3204
      self._LockInstancesNodes()
3205

    
3206
  def BuildHooksEnv(self):
3207
    """Build hooks env.
3208

3209
    This runs on master, primary and secondary nodes of the instance.
3210

3211
    """
3212
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3213
    nl = [self.cfg.GetMasterNode()]
3214
    return env, nl, nl
3215

    
3216
  def CheckPrereq(self):
3217
    """Check prerequisites.
3218

3219
    This checks that the instance is in the cluster.
3220

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

    
3226
  def Exec(self, feedback_fn):
3227
    """Remove the instance.
3228

3229
    """
3230
    instance = self.instance
3231
    logging.info("Shutting down instance %s on node %s",
3232
                 instance.name, instance.primary_node)
3233

    
3234
    result = self.rpc.call_instance_shutdown(instance.primary_node, instance)
3235
    msg = result.RemoteFailMsg()
3236
    if msg:
3237
      if self.op.ignore_failures:
3238
        feedback_fn("Warning: can't shutdown instance: %s" % msg)
3239
      else:
3240
        raise errors.OpExecError("Could not shutdown instance %s on"
3241
                                 " node %s: %s" %
3242
                                 (instance.name, instance.primary_node, msg))
3243

    
3244
    logging.info("Removing block devices for instance %s", instance.name)
3245

    
3246
    if not _RemoveDisks(self, instance):
3247
      if self.op.ignore_failures:
3248
        feedback_fn("Warning: can't remove instance's disks")
3249
      else:
3250
        raise errors.OpExecError("Can't remove instance's disks")
3251

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

    
3254
    self.cfg.RemoveInstance(instance.name)
3255
    self.remove_locks[locking.LEVEL_INSTANCE] = instance.name
3256

    
3257

    
3258
class LUQueryInstances(NoHooksLU):
3259
  """Logical unit for querying instances.
3260

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

    
3281

    
3282
  def ExpandNames(self):
3283
    _CheckOutputFields(static=self._FIELDS_STATIC,
3284
                       dynamic=self._FIELDS_DYNAMIC,
3285
                       selected=self.op.output_fields)
3286

    
3287
    self.needed_locks = {}
3288
    self.share_locks[locking.LEVEL_INSTANCE] = 1
3289
    self.share_locks[locking.LEVEL_NODE] = 1
3290

    
3291
    if self.op.names:
3292
      self.wanted = _GetWantedInstances(self, self.op.names)
3293
    else:
3294
      self.wanted = locking.ALL_SET
3295

    
3296
    self.do_node_query = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
3297
    self.do_locking = self.do_node_query and self.op.use_locking
3298
    if self.do_locking:
3299
      self.needed_locks[locking.LEVEL_INSTANCE] = self.wanted
3300
      self.needed_locks[locking.LEVEL_NODE] = []
3301
      self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3302

    
3303
  def DeclareLocks(self, level):
3304
    if level == locking.LEVEL_NODE and self.do_locking:
3305
      self._LockInstancesNodes()
3306

    
3307
  def CheckPrereq(self):
3308
    """Check prerequisites.
3309

3310
    """
3311
    pass
3312

    
3313
  def Exec(self, feedback_fn):
3314
    """Computes the list of nodes and their attributes.
3315

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

    
3337
    instance_list = [all_info[iname] for iname in instance_names]
3338

    
3339
    # begin data gathering
3340

    
3341
    nodes = frozenset([inst.primary_node for inst in instance_list])
3342
    hv_list = list(set([inst.hypervisor for inst in instance_list]))
3343

    
3344
    bad_nodes = []
3345
    off_nodes = []
3346
    if self.do_node_query:
3347
      live_data = {}
3348
      node_data = self.rpc.call_all_instances_info(nodes, hv_list)
3349
      for name in nodes:
3350
        result = node_data[name]
3351
        if result.offline:
3352
          # offline nodes will be in both lists
3353
          off_nodes.append(name)
3354
        if result.failed:
3355
          bad_nodes.append(name)
3356
        else:
3357
          if result.data:
3358
            live_data.update(result.data)
3359
            # else no instance is alive
3360
    else:
3361
      live_data = dict([(name, {}) for name in instance_names])
3362

    
3363
    # end data gathering
3364

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

    
3502
    return output
3503

    
3504

    
3505
class LUFailoverInstance(LogicalUnit):
3506
  """Failover an instance.
3507

3508
  """
3509
  HPATH = "instance-failover"
3510
  HTYPE = constants.HTYPE_INSTANCE
3511
  _OP_REQP = ["instance_name", "ignore_consistency"]
3512
  REQ_BGL = False
3513

    
3514
  def ExpandNames(self):
3515
    self._ExpandAndLockInstance()
3516
    self.needed_locks[locking.LEVEL_NODE] = []
3517
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3518

    
3519
  def DeclareLocks(self, level):
3520
    if level == locking.LEVEL_NODE:
3521
      self._LockInstancesNodes()
3522

    
3523
  def BuildHooksEnv(self):
3524
    """Build hooks env.
3525

3526
    This runs on master, primary and secondary nodes of the instance.
3527

3528
    """
3529
    env = {
3530
      "IGNORE_CONSISTENCY": self.op.ignore_consistency,
3531
      }
3532
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3533
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3534
    return env, nl, nl
3535

    
3536
  def CheckPrereq(self):
3537
    """Check prerequisites.
3538

3539
    This checks that the instance is in the cluster.
3540

3541
    """
3542
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3543
    assert self.instance is not None, \
3544
      "Cannot retrieve locked instance %s" % self.op.instance_name
3545

    
3546
    bep = self.cfg.GetClusterInfo().FillBE(instance)
3547
    if instance.disk_template not in constants.DTS_NET_MIRROR:
3548
      raise errors.OpPrereqError("Instance's disk layout is not"
3549
                                 " network mirrored, cannot failover.")
3550

    
3551
    secondary_nodes = instance.secondary_nodes
3552
    if not secondary_nodes:
3553
      raise errors.ProgrammerError("no secondary node but using "
3554
                                   "a mirrored disk template")
3555

    
3556
    target_node = secondary_nodes[0]
3557
    _CheckNodeOnline(self, target_node)
3558
    _CheckNodeNotDrained(self, target_node)
3559

    
3560
    if instance.admin_up:
3561
      # check memory requirements on the secondary node
3562
      _CheckNodeFreeMemory(self, target_node, "failing over instance %s" %
3563
                           instance.name, bep[constants.BE_MEMORY],
3564
                           instance.hypervisor)
3565
    else:
3566
      self.LogInfo("Not checking memory on the secondary node as"
3567
                   " instance will not be started")
3568

    
3569
    # check bridge existance
3570
    brlist = [nic.bridge for nic in instance.nics]
3571
    result = self.rpc.call_bridges_exist(target_node, brlist)
3572
    result.Raise()
3573
    if not result.data:
3574
      raise errors.OpPrereqError("One or more target bridges %s does not"
3575
                                 " exist on destination node '%s'" %
3576
                                 (brlist, target_node))
3577

    
3578
  def Exec(self, feedback_fn):
3579
    """Failover an instance.
3580

3581
    The failover is done by shutting it down on its present node and
3582
    starting it on the secondary.
3583

3584
    """
3585
    instance = self.instance
3586

    
3587
    source_node = instance.primary_node
3588
    target_node = instance.secondary_nodes[0]
3589

    
3590
    feedback_fn("* checking disk consistency between source and target")
3591
    for dev in instance.disks:
3592
      # for drbd, these are drbd over lvm
3593
      if not _CheckDiskConsistency(self, dev, target_node, False):
3594
        if instance.admin_up and not self.op.ignore_consistency:
3595
          raise errors.OpExecError("Disk %s is degraded on target node,"
3596
                                   " aborting failover." % dev.iv_name)
3597

    
3598
    feedback_fn("* shutting down instance on source node")
3599
    logging.info("Shutting down instance %s on node %s",
3600
                 instance.name, source_node)
3601

    
3602
    result = self.rpc.call_instance_shutdown(source_node, instance)
3603
    msg = result.RemoteFailMsg()
3604
    if msg:
3605
      if self.op.ignore_consistency:
3606
        self.proc.LogWarning("Could not shutdown instance %s on node %s."
3607
                             " Proceeding anyway. Please make sure node"
3608
                             " %s is down. Error details: %s",
3609
                             instance.name, source_node, source_node, msg)
3610
      else:
3611
        raise errors.OpExecError("Could not shutdown instance %s on"
3612
                                 " node %s: %s" %
3613
                                 (instance.name, source_node, msg))
3614

    
3615
    feedback_fn("* deactivating the instance's disks on source node")
3616
    if not _ShutdownInstanceDisks(self, instance, ignore_primary=True):
3617
      raise errors.OpExecError("Can't shut down the instance's disks.")
3618

    
3619
    instance.primary_node = target_node
3620
    # distribute new instance config to the other nodes
3621
    self.cfg.Update(instance)
3622

    
3623
    # Only start the instance if it's marked as up
3624
    if instance.admin_up:
3625
      feedback_fn("* activating the instance's disks on target node")
3626
      logging.info("Starting instance %s on node %s",
3627
                   instance.name, target_node)
3628

    
3629
      disks_ok, dummy = _AssembleInstanceDisks(self, instance,
3630
                                               ignore_secondaries=True)
3631
      if not disks_ok:
3632
        _ShutdownInstanceDisks(self, instance)
3633
        raise errors.OpExecError("Can't activate the instance's disks")
3634

    
3635
      feedback_fn("* starting the instance on the target node")
3636
      result = self.rpc.call_instance_start(target_node, instance, None, None)
3637
      msg = result.RemoteFailMsg()
3638
      if msg:
3639
        _ShutdownInstanceDisks(self, instance)
3640
        raise errors.OpExecError("Could not start instance %s on node %s: %s" %
3641
                                 (instance.name, target_node, msg))
3642

    
3643

    
3644
class LUMigrateInstance(LogicalUnit):
3645
  """Migrate an instance.
3646

3647
  This is migration without shutting down, compared to the failover,
3648
  which is done with shutdown.
3649

3650
  """
3651
  HPATH = "instance-migrate"
3652
  HTYPE = constants.HTYPE_INSTANCE
3653
  _OP_REQP = ["instance_name", "live", "cleanup"]
3654

    
3655
  REQ_BGL = False
3656

    
3657
  def ExpandNames(self):
3658
    self._ExpandAndLockInstance()
3659
    self.needed_locks[locking.LEVEL_NODE] = []
3660
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3661

    
3662
  def DeclareLocks(self, level):
3663
    if level == locking.LEVEL_NODE:
3664
      self._LockInstancesNodes()
3665

    
3666
  def BuildHooksEnv(self):
3667
    """Build hooks env.
3668

3669
    This runs on master, primary and secondary nodes of the instance.
3670

3671
    """
3672
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3673
    env["MIGRATE_LIVE"] = self.op.live
3674
    env["MIGRATE_CLEANUP"] = self.op.cleanup
3675
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3676
    return env, nl, nl
3677

    
3678
  def CheckPrereq(self):
3679
    """Check prerequisites.
3680

3681
    This checks that the instance is in the cluster.
3682

3683
    """
3684
    instance = self.cfg.GetInstanceInfo(
3685
      self.cfg.ExpandInstanceName(self.op.instance_name))
3686
    if instance is None:
3687
      raise errors.OpPrereqError("Instance '%s' not known" %
3688
                                 self.op.instance_name)
3689

    
3690
    if instance.disk_template != constants.DT_DRBD8:
3691
      raise errors.OpPrereqError("Instance's disk layout is not"
3692
                                 " drbd8, cannot migrate.")
3693

    
3694
    secondary_nodes = instance.secondary_nodes
3695
    if not secondary_nodes:
3696
      raise errors.ConfigurationError("No secondary node but using"
3697
                                      " drbd8 disk template")
3698

    
3699
    i_be = self.cfg.GetClusterInfo().FillBE(instance)
3700

    
3701
    target_node = secondary_nodes[0]
3702
    # check memory requirements on the secondary node
3703
    _CheckNodeFreeMemory(self, target_node, "migrating instance %s" %
3704
                         instance.name, i_be[constants.BE_MEMORY],
3705
                         instance.hypervisor)
3706

    
3707
    # check bridge existance
3708
    brlist = [nic.bridge for nic in instance.nics]
3709
    result = self.rpc.call_bridges_exist(target_node, brlist)
3710
    if result.failed or not result.data:
3711
      raise errors.OpPrereqError("One or more target bridges %s does not"
3712
                                 " exist on destination node '%s'" %
3713
                                 (brlist, target_node))
3714

    
3715
    if not self.op.cleanup:
3716
      _CheckNodeNotDrained(self, target_node)
3717
      result = self.rpc.call_instance_migratable(instance.primary_node,
3718
                                                 instance)
3719
      msg = result.RemoteFailMsg()
3720
      if msg:
3721
        raise errors.OpPrereqError("Can't migrate: %s - please use failover" %
3722
                                   msg)
3723

    
3724
    self.instance = instance
3725

    
3726
  def _WaitUntilSync(self):
3727
    """Poll with custom rpc for disk sync.
3728

3729
    This uses our own step-based rpc call.
3730

3731
    """
3732
    self.feedback_fn("* wait until resync is done")
3733
    all_done = False
3734
    while not all_done:
3735
      all_done = True
3736
      result = self.rpc.call_drbd_wait_sync(self.all_nodes,
3737
                                            self.nodes_ip,
3738
                                            self.instance.disks)
3739
      min_percent = 100
3740
      for node, nres in result.items():
3741
        msg = nres.RemoteFailMsg()
3742
        if msg:
3743
          raise errors.OpExecError("Cannot resync disks on node %s: %s" %
3744
                                   (node, msg))
3745
        node_done, node_percent = nres.payload
3746
        all_done = all_done and node_done
3747
        if node_percent is not None:
3748
          min_percent = min(min_percent, node_percent)
3749
      if not all_done:
3750
        if min_percent < 100:
3751
          self.feedback_fn("   - progress: %.1f%%" % min_percent)
3752
        time.sleep(2)
3753

    
3754
  def _EnsureSecondary(self, node):
3755
    """Demote a node to secondary.
3756

3757
    """
3758
    self.feedback_fn("* switching node %s to secondary mode" % node)
3759

    
3760
    for dev in self.instance.disks:
3761
      self.cfg.SetDiskID(dev, node)
3762

    
3763
    result = self.rpc.call_blockdev_close(node, self.instance.name,
3764
                                          self.instance.disks)
3765
    msg = result.RemoteFailMsg()
3766
    if msg:
3767
      raise errors.OpExecError("Cannot change disk to secondary on node %s,"
3768
                               " error %s" % (node, msg))
3769

    
3770
  def _GoStandalone(self):
3771
    """Disconnect from the network.
3772

3773
    """
3774
    self.feedback_fn("* changing into standalone mode")
3775
    result = self.rpc.call_drbd_disconnect_net(self.all_nodes, self.nodes_ip,
3776
                                               self.instance.disks)
3777
    for node, nres in result.items():
3778
      msg = nres.RemoteFailMsg()
3779
      if msg:
3780
        raise errors.OpExecError("Cannot disconnect disks node %s,"
3781
                                 " error %s" % (node, msg))
3782

    
3783
  def _GoReconnect(self, multimaster):
3784
    """Reconnect to the network.
3785

3786
    """
3787
    if multimaster:
3788
      msg = "dual-master"
3789
    else:
3790
      msg = "single-master"
3791
    self.feedback_fn("* changing disks into %s mode" % msg)
3792
    result = self.rpc.call_drbd_attach_net(self.all_nodes, self.nodes_ip,
3793
                                           self.instance.disks,
3794
                                           self.instance.name, multimaster)
3795
    for node, nres in result.items():
3796
      msg = nres.RemoteFailMsg()
3797
      if msg:
3798
        raise errors.OpExecError("Cannot change disks config on node %s,"
3799
                                 " error: %s" % (node, msg))
3800

    
3801
  def _ExecCleanup(self):
3802
    """Try to cleanup after a failed migration.
3803

3804
    The cleanup is done by:
3805
      - check that the instance is running only on one node
3806
        (and update the config if needed)
3807
      - change disks on its secondary node to secondary
3808
      - wait until disks are fully synchronized
3809
      - disconnect from the network
3810
      - change disks into single-master mode
3811
      - wait again until disks are fully synchronized
3812

3813
    """
3814
    instance = self.instance
3815
    target_node = self.target_node
3816
    source_node = self.source_node
3817

    
3818
    # check running on only one node
3819
    self.feedback_fn("* checking where the instance actually runs"
3820
                     " (if this hangs, the hypervisor might be in"
3821
                     " a bad state)")
3822
    ins_l = self.rpc.call_instance_list(self.all_nodes, [instance.hypervisor])
3823
    for node, result in ins_l.items():
3824
      result.Raise()
3825
      if not isinstance(result.data, list):
3826
        raise errors.OpExecError("Can't contact node '%s'" % node)
3827

    
3828
    runningon_source = instance.name in ins_l[source_node].data
3829
    runningon_target = instance.name in ins_l[target_node].data
3830

    
3831
    if runningon_source and runningon_target:
3832
      raise errors.OpExecError("Instance seems to be running on two nodes,"
3833
                               " or the hypervisor is confused. You will have"
3834
                               " to ensure manually that it runs only on one"
3835
                               " and restart this operation.")
3836

    
3837
    if not (runningon_source or runningon_target):
3838
      raise errors.OpExecError("Instance does not seem to be running at all."
3839
                               " In this case, it's safer to repair by"
3840
                               " running 'gnt-instance stop' to ensure disk"
3841
                               " shutdown, and then restarting it.")
3842

    
3843
    if runningon_target:
3844
      # the migration has actually succeeded, we need to update the config
3845
      self.feedback_fn("* instance running on secondary node (%s),"
3846
                       " updating config" % target_node)
3847
      instance.primary_node = target_node
3848
      self.cfg.Update(instance)
3849
      demoted_node = source_node
3850
    else:
3851
      self.feedback_fn("* instance confirmed to be running on its"
3852
                       " primary node (%s)" % source_node)
3853
      demoted_node = target_node
3854

    
3855
    self._EnsureSecondary(demoted_node)
3856
    try:
3857
      self._WaitUntilSync()
3858
    except errors.OpExecError:
3859
      # we ignore here errors, since if the device is standalone, it
3860
      # won't be able to sync
3861
      pass
3862
    self._GoStandalone()
3863
    self._GoReconnect(False)
3864
    self._WaitUntilSync()
3865

    
3866
    self.feedback_fn("* done")
3867

    
3868
  def _RevertDiskStatus(self):
3869
    """Try to revert the disk status after a failed migration.
3870

3871
    """
3872
    target_node = self.target_node
3873
    try:
3874
      self._EnsureSecondary(target_node)
3875
      self._GoStandalone()
3876
      self._GoReconnect(False)
3877
      self._WaitUntilSync()
3878
    except errors.OpExecError, err:
3879
      self.LogWarning("Migration failed and I can't reconnect the"
3880
                      " drives: error '%s'\n"
3881
                      "Please look and recover the instance status" %
3882
                      str(err))
3883

    
3884
  def _AbortMigration(self):
3885
    """Call the hypervisor code to abort a started migration.
3886

3887
    """
3888
    instance = self.instance
3889
    target_node = self.target_node
3890
    migration_info = self.migration_info
3891

    
3892
    abort_result = self.rpc.call_finalize_migration(target_node,
3893
                                                    instance,
3894
                                                    migration_info,
3895
                                                    False)
3896
    abort_msg = abort_result.RemoteFailMsg()
3897
    if abort_msg:
3898
      logging.error("Aborting migration failed on target node %s: %s" %
3899
                    (target_node, abort_msg))
3900
      # Don't raise an exception here, as we stil have to try to revert the
3901
      # disk status, even if this step failed.
3902

    
3903
  def _ExecMigration(self):
3904
    """Migrate an instance.
3905

3906
    The migrate is done by:
3907
      - change the disks into dual-master mode
3908
      - wait until disks are fully synchronized again
3909
      - migrate the instance
3910
      - change disks on the new secondary node (the old primary) to secondary
3911
      - wait until disks are fully synchronized
3912
      - change disks into single-master mode
3913

3914
    """
3915
    instance = self.instance
3916
    target_node = self.target_node
3917
    source_node = self.source_node
3918

    
3919
    self.feedback_fn("* checking disk consistency between source and target")
3920
    for dev in instance.disks:
3921
      if not _CheckDiskConsistency(self, dev, target_node, False):
3922
        raise errors.OpExecError("Disk %s is degraded or not fully"
3923
                                 " synchronized on target node,"
3924
                                 " aborting migrate." % dev.iv_name)
3925

    
3926
    # First get the migration information from the remote node
3927
    result = self.rpc.call_migration_info(source_node, instance)
3928
    msg = result.RemoteFailMsg()
3929
    if msg:
3930
      log_err = ("Failed fetching source migration information from %s: %s" %
3931
                 (source_node, msg))
3932
      logging.error(log_err)
3933
      raise errors.OpExecError(log_err)
3934

    
3935
    self.migration_info = migration_info = result.payload
3936

    
3937
    # Then switch the disks to master/master mode
3938
    self._EnsureSecondary(target_node)
3939
    self._GoStandalone()
3940
    self._GoReconnect(True)
3941
    self._WaitUntilSync()
3942

    
3943
    self.feedback_fn("* preparing %s to accept the instance" % target_node)
3944
    result = self.rpc.call_accept_instance(target_node,
3945
                                           instance,
3946
                                           migration_info,
3947
                                           self.nodes_ip[target_node])
3948

    
3949
    msg = result.RemoteFailMsg()
3950
    if msg:
3951
      logging.error("Instance pre-migration failed, trying to revert"
3952
                    " disk status: %s", msg)
3953
      self._AbortMigration()
3954
      self._RevertDiskStatus()
3955
      raise errors.OpExecError("Could not pre-migrate instance %s: %s" %
3956
                               (instance.name, msg))
3957

    
3958
    self.feedback_fn("* migrating instance to %s" % target_node)
3959
    time.sleep(10)
3960
    result = self.rpc.call_instance_migrate(source_node, instance,
3961
                                            self.nodes_ip[target_node],
3962
                                            self.op.live)
3963
    msg = result.RemoteFailMsg()
3964
    if msg:
3965
      logging.error("Instance migration failed, trying to revert"
3966
                    " disk status: %s", msg)
3967
      self._AbortMigration()
3968
      self._RevertDiskStatus()
3969
      raise errors.OpExecError("Could not migrate instance %s: %s" %
3970
                               (instance.name, msg))
3971
    time.sleep(10)
3972

    
3973
    instance.primary_node = target_node
3974
    # distribute new instance config to the other nodes
3975
    self.cfg.Update(instance)
3976

    
3977
    result = self.rpc.call_finalize_migration(target_node,
3978
                                              instance,
3979
                                              migration_info,
3980
                                              True)
3981
    msg = result.RemoteFailMsg()
3982
    if msg:
3983
      logging.error("Instance migration succeeded, but finalization failed:"
3984
                    " %s" % msg)
3985
      raise errors.OpExecError("Could not finalize instance migration: %s" %