Statistics
| Branch: | Tag: | Revision:

root / lib / cmdlib.py @ b2b8bcce

History | View | Annotate | Download (252.6 kB)

1
#
2
#
3

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

    
21

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

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

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

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

    
47

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

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

60
  Note that all commands require root permissions.
61

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

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

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

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

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

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

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

    
108
  ssh = property(fget=__GetSSH)
109

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

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

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

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

125
    """
126
    pass
127

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

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

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

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

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

149
    Examples::
150

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

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

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

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

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

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

188
    """
189

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

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

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

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

204
    """
205
    raise NotImplementedError
206

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

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

214
    """
215
    raise NotImplementedError
216

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

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

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

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

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

236
    """
237
    raise NotImplementedError
238

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

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

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

257
    """
258
    return lu_result
259

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

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

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

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

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

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

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

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

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

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

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

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

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

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

    
325
    del self.recalculate_locks[locking.LEVEL_NODE]
326

    
327

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

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

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

    
338

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

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

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

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

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

    
365
  return utils.NiceSort(wanted)
366

    
367

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

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

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

    
384
  if instances:
385
    wanted = []
386

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

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

    
397

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

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

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

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

    
416

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

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

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

    
430

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

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

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

    
442

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

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

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

    
454

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

459
  This builds the hook environment from individual variables.
460

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

485
  """
486
  if status:
487
    str_status = "up"
488
  else:
489
    str_status = "down"
490
  env = {
491
    "OP_TARGET": name,
492
    "INSTANCE_NAME": name,
493
    "INSTANCE_PRIMARY": primary_node,
494
    "INSTANCE_SECONDARIES": " ".join(secondary_nodes),
495
    "INSTANCE_OS_TYPE": os_type,
496
    "INSTANCE_STATUS": str_status,
497
    "INSTANCE_MEMORY": memory,
498
    "INSTANCE_VCPUS": vcpus,
499
    "INSTANCE_DISK_TEMPLATE": disk_template,
500
  }
501

    
502
  if nics:
503
    nic_count = len(nics)
504
    for idx, (ip, mac, mode, link) in enumerate(nics):
505
      if ip is None:
506
        ip = ""
507
      env["INSTANCE_NIC%d_IP" % idx] = ip
508
      env["INSTANCE_NIC%d_MAC" % idx] = mac
509
      env["INSTANCE_NIC%d_MODE" % idx] = mode
510
      env["INSTANCE_NIC%d_LINK" % idx] = link
511
      if mode == constants.NIC_MODE_BRIDGED:
512
        env["INSTANCE_NIC%d_BRIDGE" % idx] = link
513
  else:
514
    nic_count = 0
515

    
516
  env["INSTANCE_NIC_COUNT"] = nic_count
517

    
518
  if disks:
519
    disk_count = len(disks)
520
    for idx, (size, mode) in enumerate(disks):
521
      env["INSTANCE_DISK%d_SIZE" % idx] = size
522
      env["INSTANCE_DISK%d_MODE" % idx] = mode
523
  else:
524
    disk_count = 0
525

    
526
  env["INSTANCE_DISK_COUNT"] = disk_count
527

    
528
  return env
529

    
530
def _PreBuildNICHooksList(lu, nics):
531
  """Build a list of nic information tuples.
532

533
  This list is suitable to be passed to _BuildInstanceHookEnv.
534

535
  @type lu:  L{LogicalUnit}
536
  @param lu: the logical unit on whose behalf we execute
537
  @type nics: list of L{objects.NIC}
538
  @param nics: list of nics to convert to hooks tuples
539

540
  """
541
  hooks_nics = []
542
  c_nicparams = lu.cfg.GetClusterInfo().nicparams[constants.PP_DEFAULT]
543
  for nic in nics:
544
    ip = nic.ip
545
    mac = nic.mac
546
    filled_params = objects.FillDict(c_nicparams, nic.nicparams)
547
    mode = filled_params[constants.NIC_MODE]
548
    link = filled_params[constants.NIC_LINK]
549
    hooks_nics.append((ip, mac, mode, link))
550
  return hooks_nics
551

    
552
def _BuildInstanceHookEnvByObject(lu, instance, override=None):
553
  """Builds instance related env variables for hooks from an object.
554

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

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

    
584

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

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

    
600

    
601
def _CheckNicsBridgesExist(lu, target_nics, target_node,
602
                               profile=constants.PP_DEFAULT):
603
  """Check that the brigdes needed by a list of nics exist.
604

605
  """
606
  c_nicparams = lu.cfg.GetClusterInfo().nicparams[profile]
607
  paramslist = [objects.FillDict(c_nicparams, nic.nicparams)
608
                for nic in target_nics]
609
  brlist = [params[constants.NIC_LINK] for params in paramslist
610
            if params[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED]
611
  if brlist:
612
    result = lu.rpc.call_bridges_exist(target_node, brlist)
613
    msg = result.RemoteFailMsg()
614
    if msg:
615
      raise errors.OpPrereqError("Error checking bridges on destination node"
616
                                 " '%s': %s" % (target_node, msg))
617

    
618

    
619
def _CheckInstanceBridgesExist(lu, instance, node=None):
620
  """Check that the brigdes needed by an instance exist.
621

622
  """
623
  if node is None:
624
    node=instance.primary_node
625
  _CheckNicsBridgesExist(lu, instance.nics, node)
626

    
627

    
628
class LUDestroyCluster(NoHooksLU):
629
  """Logical unit for destroying the cluster.
630

631
  """
632
  _OP_REQP = []
633

    
634
  def CheckPrereq(self):
635
    """Check prerequisites.
636

637
    This checks whether the cluster is empty.
638

639
    Any errors are signalled by raising errors.OpPrereqError.
640

641
    """
642
    master = self.cfg.GetMasterNode()
643

    
644
    nodelist = self.cfg.GetNodeList()
645
    if len(nodelist) != 1 or nodelist[0] != master:
646
      raise errors.OpPrereqError("There are still %d node(s) in"
647
                                 " this cluster." % (len(nodelist) - 1))
648
    instancelist = self.cfg.GetInstanceList()
649
    if instancelist:
650
      raise errors.OpPrereqError("There are still %d instance(s) in"
651
                                 " this cluster." % len(instancelist))
652

    
653
  def Exec(self, feedback_fn):
654
    """Destroys the cluster.
655

656
    """
657
    master = self.cfg.GetMasterNode()
658
    result = self.rpc.call_node_stop_master(master, False)
659
    msg = result.RemoteFailMsg()
660
    if msg:
661
      raise errors.OpExecError("Could not disable the master role: %s" % msg)
662
    priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
663
    utils.CreateBackup(priv_key)
664
    utils.CreateBackup(pub_key)
665
    return master
666

    
667

    
668
class LUVerifyCluster(LogicalUnit):
669
  """Verifies the cluster status.
670

671
  """
672
  HPATH = "cluster-verify"
673
  HTYPE = constants.HTYPE_CLUSTER
674
  _OP_REQP = ["skip_checks"]
675
  REQ_BGL = False
676

    
677
  def ExpandNames(self):
678
    self.needed_locks = {
679
      locking.LEVEL_NODE: locking.ALL_SET,
680
      locking.LEVEL_INSTANCE: locking.ALL_SET,
681
    }
682
    self.share_locks = dict(((i, 1) for i in locking.LEVELS))
683

    
684
  def _VerifyNode(self, nodeinfo, file_list, local_cksum,
685
                  node_result, feedback_fn, master_files,
686
                  drbd_map, vg_name):
687
    """Run multiple tests against a node.
688

689
    Test list:
690

691
      - compares ganeti version
692
      - checks vg existance and size > 20G
693
      - checks config file checksum
694
      - checks ssh to other nodes
695

696
    @type nodeinfo: L{objects.Node}
697
    @param nodeinfo: the node to check
698
    @param file_list: required list of files
699
    @param local_cksum: dictionary of local files and their checksums
700
    @param node_result: the results from the node
701
    @param feedback_fn: function used to accumulate results
702
    @param master_files: list of files that only masters should have
703
    @param drbd_map: the useddrbd minors for this node, in
704
        form of minor: (instance, must_exist) which correspond to instances
705
        and their running status
706
    @param vg_name: Ganeti Volume Group (result of self.cfg.GetVGName())
707

708
    """
709
    node = nodeinfo.name
710

    
711
    # main result, node_result should be a non-empty dict
712
    if not node_result or not isinstance(node_result, dict):
713
      feedback_fn("  - ERROR: unable to verify node %s." % (node,))
714
      return True
715

    
716
    # compares ganeti version
717
    local_version = constants.PROTOCOL_VERSION
718
    remote_version = node_result.get('version', None)
719
    if not (remote_version and isinstance(remote_version, (list, tuple)) and
720
            len(remote_version) == 2):
721
      feedback_fn("  - ERROR: connection to %s failed" % (node))
722
      return True
723

    
724
    if local_version != remote_version[0]:
725
      feedback_fn("  - ERROR: incompatible protocol versions: master %s,"
726
                  " node %s %s" % (local_version, node, remote_version[0]))
727
      return True
728

    
729
    # node seems compatible, we can actually try to look into its results
730

    
731
    bad = False
732

    
733
    # full package version
734
    if constants.RELEASE_VERSION != remote_version[1]:
735
      feedback_fn("  - WARNING: software version mismatch: master %s,"
736
                  " node %s %s" %
737
                  (constants.RELEASE_VERSION, node, remote_version[1]))
738

    
739
    # checks vg existence and size > 20G
740
    if vg_name is not None:
741
      vglist = node_result.get(constants.NV_VGLIST, None)
742
      if not vglist:
743
        feedback_fn("  - ERROR: unable to check volume groups on node %s." %
744
                        (node,))
745
        bad = True
746
      else:
747
        vgstatus = utils.CheckVolumeGroupSize(vglist, vg_name,
748
                                              constants.MIN_VG_SIZE)
749
        if vgstatus:
750
          feedback_fn("  - ERROR: %s on node %s" % (vgstatus, node))
751
          bad = True
752

    
753
    # checks config file checksum
754

    
755
    remote_cksum = node_result.get(constants.NV_FILELIST, None)
756
    if not isinstance(remote_cksum, dict):
757
      bad = True
758
      feedback_fn("  - ERROR: node hasn't returned file checksum data")
759
    else:
760
      for file_name in file_list:
761
        node_is_mc = nodeinfo.master_candidate
762
        must_have_file = file_name not in master_files
763
        if file_name not in remote_cksum:
764
          if node_is_mc or must_have_file:
765
            bad = True
766
            feedback_fn("  - ERROR: file '%s' missing" % file_name)
767
        elif remote_cksum[file_name] != local_cksum[file_name]:
768
          if node_is_mc or must_have_file:
769
            bad = True
770
            feedback_fn("  - ERROR: file '%s' has wrong checksum" % file_name)
771
          else:
772
            # not candidate and this is not a must-have file
773
            bad = True
774
            feedback_fn("  - ERROR: non master-candidate has old/wrong file"
775
                        " '%s'" % file_name)
776
        else:
777
          # all good, except non-master/non-must have combination
778
          if not node_is_mc and not must_have_file:
779
            feedback_fn("  - ERROR: file '%s' should not exist on non master"
780
                        " candidates" % file_name)
781

    
782
    # checks ssh to any
783

    
784
    if constants.NV_NODELIST not in node_result:
785
      bad = True
786
      feedback_fn("  - ERROR: node hasn't returned node ssh connectivity data")
787
    else:
788
      if node_result[constants.NV_NODELIST]:
789
        bad = True
790
        for node in node_result[constants.NV_NODELIST]:
791
          feedback_fn("  - ERROR: ssh communication with node '%s': %s" %
792
                          (node, node_result[constants.NV_NODELIST][node]))
793

    
794
    if constants.NV_NODENETTEST not in node_result:
795
      bad = True
796
      feedback_fn("  - ERROR: node hasn't returned node tcp connectivity data")
797
    else:
798
      if node_result[constants.NV_NODENETTEST]:
799
        bad = True
800
        nlist = utils.NiceSort(node_result[constants.NV_NODENETTEST].keys())
801
        for node in nlist:
802
          feedback_fn("  - ERROR: tcp communication with node '%s': %s" %
803
                          (node, node_result[constants.NV_NODENETTEST][node]))
804

    
805
    hyp_result = node_result.get(constants.NV_HYPERVISOR, None)
806
    if isinstance(hyp_result, dict):
807
      for hv_name, hv_result in hyp_result.iteritems():
808
        if hv_result is not None:
809
          feedback_fn("  - ERROR: hypervisor %s verify failure: '%s'" %
810
                      (hv_name, hv_result))
811

    
812
    # check used drbd list
813
    if vg_name is not None:
814
      used_minors = node_result.get(constants.NV_DRBDLIST, [])
815
      if not isinstance(used_minors, (tuple, list)):
816
        feedback_fn("  - ERROR: cannot parse drbd status file: %s" %
817
                    str(used_minors))
818
      else:
819
        for minor, (iname, must_exist) in drbd_map.items():
820
          if minor not in used_minors and must_exist:
821
            feedback_fn("  - ERROR: drbd minor %d of instance %s is"
822
                        " not active" % (minor, iname))
823
            bad = True
824
        for minor in used_minors:
825
          if minor not in drbd_map:
826
            feedback_fn("  - ERROR: unallocated drbd minor %d is in use" %
827
                        minor)
828
            bad = True
829

    
830
    return bad
831

    
832
  def _VerifyInstance(self, instance, instanceconfig, node_vol_is,
833
                      node_instance, feedback_fn, n_offline):
834
    """Verify an instance.
835

836
    This function checks to see if the required block devices are
837
    available on the instance's node.
838

839
    """
840
    bad = False
841

    
842
    node_current = instanceconfig.primary_node
843

    
844
    node_vol_should = {}
845
    instanceconfig.MapLVsByNode(node_vol_should)
846

    
847
    for node in node_vol_should:
848
      if node in n_offline:
849
        # ignore missing volumes on offline nodes
850
        continue
851
      for volume in node_vol_should[node]:
852
        if node not in node_vol_is or volume not in node_vol_is[node]:
853
          feedback_fn("  - ERROR: volume %s missing on node %s" %
854
                          (volume, node))
855
          bad = True
856

    
857
    if instanceconfig.admin_up:
858
      if ((node_current not in node_instance or
859
          not instance in node_instance[node_current]) and
860
          node_current not in n_offline):
861
        feedback_fn("  - ERROR: instance %s not running on node %s" %
862
                        (instance, node_current))
863
        bad = True
864

    
865
    for node in node_instance:
866
      if (not node == node_current):
867
        if instance in node_instance[node]:
868
          feedback_fn("  - ERROR: instance %s should not run on node %s" %
869
                          (instance, node))
870
          bad = True
871

    
872
    return bad
873

    
874
  def _VerifyOrphanVolumes(self, node_vol_should, node_vol_is, feedback_fn):
875
    """Verify if there are any unknown volumes in the cluster.
876

877
    The .os, .swap and backup volumes are ignored. All other volumes are
878
    reported as unknown.
879

880
    """
881
    bad = False
882

    
883
    for node in node_vol_is:
884
      for volume in node_vol_is[node]:
885
        if node not in node_vol_should or volume not in node_vol_should[node]:
886
          feedback_fn("  - ERROR: volume %s on node %s should not exist" %
887
                      (volume, node))
888
          bad = True
889
    return bad
890

    
891
  def _VerifyOrphanInstances(self, instancelist, node_instance, feedback_fn):
892
    """Verify the list of running instances.
893

894
    This checks what instances are running but unknown to the cluster.
895

896
    """
897
    bad = False
898
    for node in node_instance:
899
      for runninginstance in node_instance[node]:
900
        if runninginstance not in instancelist:
901
          feedback_fn("  - ERROR: instance %s on node %s should not exist" %
902
                          (runninginstance, node))
903
          bad = True
904
    return bad
905

    
906
  def _VerifyNPlusOneMemory(self, node_info, instance_cfg, feedback_fn):
907
    """Verify N+1 Memory Resilience.
908

909
    Check that if one single node dies we can still start all the instances it
910
    was primary for.
911

912
    """
913
    bad = False
914

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

    
936
  def CheckPrereq(self):
937
    """Check prerequisites.
938

939
    Transform the list of checks we're going to skip into a set and check that
940
    all its members are valid.
941

942
    """
943
    self.skip_set = frozenset(self.op.skip_checks)
944
    if not constants.VERIFY_OPTIONAL_CHECKS.issuperset(self.skip_set):
945
      raise errors.OpPrereqError("Invalid checks to be skipped specified")
946

    
947
  def BuildHooksEnv(self):
948
    """Build hooks env.
949

950
    Cluster-Verify hooks just rone in the post phase and their failure makes
951
    the output be logged in the verify output and the verification to fail.
952

953
    """
954
    all_nodes = self.cfg.GetNodeList()
955
    env = {
956
      "CLUSTER_TAGS": " ".join(self.cfg.GetClusterInfo().GetTags())
957
      }
958
    for node in self.cfg.GetAllNodesInfo().values():
959
      env["NODE_TAGS_%s" % node.name] = " ".join(node.GetTags())
960

    
961
    return env, [], all_nodes
962

    
963
  def Exec(self, feedback_fn):
964
    """Verify integrity of cluster, performing various test on nodes.
965

966
    """
967
    bad = False
968
    feedback_fn("* Verifying global settings")
969
    for msg in self.cfg.VerifyConfig():
970
      feedback_fn("  - ERROR: %s" % msg)
971

    
972
    vg_name = self.cfg.GetVGName()
973
    hypervisors = self.cfg.GetClusterInfo().enabled_hypervisors
974
    nodelist = utils.NiceSort(self.cfg.GetNodeList())
975
    nodeinfo = [self.cfg.GetNodeInfo(nname) for nname in nodelist]
976
    instancelist = utils.NiceSort(self.cfg.GetInstanceList())
977
    instanceinfo = dict((iname, self.cfg.GetInstanceInfo(iname))
978
                        for iname in instancelist)
979
    i_non_redundant = [] # Non redundant instances
980
    i_non_a_balanced = [] # Non auto-balanced instances
981
    n_offline = [] # List of offline nodes
982
    n_drained = [] # List of nodes being drained
983
    node_volume = {}
984
    node_instance = {}
985
    node_info = {}
986
    instance_cfg = {}
987

    
988
    # FIXME: verify OS list
989
    # do local checksums
990
    master_files = [constants.CLUSTER_CONF_FILE]
991

    
992
    file_names = ssconf.SimpleStore().GetFileList()
993
    file_names.append(constants.SSL_CERT_FILE)
994
    file_names.append(constants.RAPI_CERT_FILE)
995
    file_names.extend(master_files)
996

    
997
    local_checksums = utils.FingerprintFiles(file_names)
998

    
999
    feedback_fn("* Gathering data (%d nodes)" % len(nodelist))
1000
    node_verify_param = {
1001
      constants.NV_FILELIST: file_names,
1002
      constants.NV_NODELIST: [node.name for node in nodeinfo
1003
                              if not node.offline],
1004
      constants.NV_HYPERVISOR: hypervisors,
1005
      constants.NV_NODENETTEST: [(node.name, node.primary_ip,
1006
                                  node.secondary_ip) for node in nodeinfo
1007
                                 if not node.offline],
1008
      constants.NV_INSTANCELIST: hypervisors,
1009
      constants.NV_VERSION: None,
1010
      constants.NV_HVINFO: self.cfg.GetHypervisorType(),
1011
      }
1012
    if vg_name is not None:
1013
      node_verify_param[constants.NV_VGLIST] = None
1014
      node_verify_param[constants.NV_LVLIST] = vg_name
1015
      node_verify_param[constants.NV_DRBDLIST] = None
1016
    all_nvinfo = self.rpc.call_node_verify(nodelist, node_verify_param,
1017
                                           self.cfg.GetClusterName())
1018

    
1019
    cluster = self.cfg.GetClusterInfo()
1020
    master_node = self.cfg.GetMasterNode()
1021
    all_drbd_map = self.cfg.ComputeDRBDMap()
1022

    
1023
    for node_i in nodeinfo:
1024
      node = node_i.name
1025

    
1026
      if node_i.offline:
1027
        feedback_fn("* Skipping offline node %s" % (node,))
1028
        n_offline.append(node)
1029
        continue
1030

    
1031
      if node == master_node:
1032
        ntype = "master"
1033
      elif node_i.master_candidate:
1034
        ntype = "master candidate"
1035
      elif node_i.drained:
1036
        ntype = "drained"
1037
        n_drained.append(node)
1038
      else:
1039
        ntype = "regular"
1040
      feedback_fn("* Verifying node %s (%s)" % (node, ntype))
1041

    
1042
      msg = all_nvinfo[node].RemoteFailMsg()
1043
      if msg:
1044
        feedback_fn("  - ERROR: while contacting node %s: %s" % (node, msg))
1045
        bad = True
1046
        continue
1047

    
1048
      nresult = all_nvinfo[node].payload
1049
      node_drbd = {}
1050
      for minor, instance in all_drbd_map[node].items():
1051
        if instance not in instanceinfo:
1052
          feedback_fn("  - ERROR: ghost instance '%s' in temporary DRBD map" %
1053
                      instance)
1054
          # ghost instance should not be running, but otherwise we
1055
          # don't give double warnings (both ghost instance and
1056
          # unallocated minor in use)
1057
          node_drbd[minor] = (instance, False)
1058
        else:
1059
          instance = instanceinfo[instance]
1060
          node_drbd[minor] = (instance.name, instance.admin_up)
1061
      result = self._VerifyNode(node_i, file_names, local_checksums,
1062
                                nresult, feedback_fn, master_files,
1063
                                node_drbd, vg_name)
1064
      bad = bad or result
1065

    
1066
      lvdata = nresult.get(constants.NV_LVLIST, "Missing LV data")
1067
      if vg_name is None:
1068
        node_volume[node] = {}
1069
      elif isinstance(lvdata, basestring):
1070
        feedback_fn("  - ERROR: LVM problem on node %s: %s" %
1071
                    (node, utils.SafeEncode(lvdata)))
1072
        bad = True
1073
        node_volume[node] = {}
1074
      elif not isinstance(lvdata, dict):
1075
        feedback_fn("  - ERROR: connection to %s failed (lvlist)" % (node,))
1076
        bad = True
1077
        continue
1078
      else:
1079
        node_volume[node] = lvdata
1080

    
1081
      # node_instance
1082
      idata = nresult.get(constants.NV_INSTANCELIST, None)
1083
      if not isinstance(idata, list):
1084
        feedback_fn("  - ERROR: connection to %s failed (instancelist)" %
1085
                    (node,))
1086
        bad = True
1087
        continue
1088

    
1089
      node_instance[node] = idata
1090

    
1091
      # node_info
1092
      nodeinfo = nresult.get(constants.NV_HVINFO, None)
1093
      if not isinstance(nodeinfo, dict):
1094
        feedback_fn("  - ERROR: connection to %s failed (hvinfo)" % (node,))
1095
        bad = True
1096
        continue
1097

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

    
1127
    node_vol_should = {}
1128

    
1129
    for instance in instancelist:
1130
      feedback_fn("* Verifying instance %s" % instance)
1131
      inst_config = instanceinfo[instance]
1132
      result =  self._VerifyInstance(instance, inst_config, node_volume,
1133
                                     node_instance, feedback_fn, n_offline)
1134
      bad = bad or result
1135
      inst_nodes_offline = []
1136

    
1137
      inst_config.MapLVsByNode(node_vol_should)
1138

    
1139
      instance_cfg[instance] = inst_config
1140

    
1141
      pnode = inst_config.primary_node
1142
      if pnode in node_info:
1143
        node_info[pnode]['pinst'].append(instance)
1144
      elif pnode not in n_offline:
1145
        feedback_fn("  - ERROR: instance %s, connection to primary node"
1146
                    " %s failed" % (instance, pnode))
1147
        bad = True
1148

    
1149
      if pnode in n_offline:
1150
        inst_nodes_offline.append(pnode)
1151

    
1152
      # If the instance is non-redundant we cannot survive losing its primary
1153
      # node, so we are not N+1 compliant. On the other hand we have no disk
1154
      # templates with more than one secondary so that situation is not well
1155
      # supported either.
1156
      # FIXME: does not support file-backed instances
1157
      if len(inst_config.secondary_nodes) == 0:
1158
        i_non_redundant.append(instance)
1159
      elif len(inst_config.secondary_nodes) > 1:
1160
        feedback_fn("  - WARNING: multiple secondaries for instance %s"
1161
                    % instance)
1162

    
1163
      if not cluster.FillBE(inst_config)[constants.BE_AUTO_BALANCE]:
1164
        i_non_a_balanced.append(instance)
1165

    
1166
      for snode in inst_config.secondary_nodes:
1167
        if snode in node_info:
1168
          node_info[snode]['sinst'].append(instance)
1169
          if pnode not in node_info[snode]['sinst-by-pnode']:
1170
            node_info[snode]['sinst-by-pnode'][pnode] = []
1171
          node_info[snode]['sinst-by-pnode'][pnode].append(instance)
1172
        elif snode not in n_offline:
1173
          feedback_fn("  - ERROR: instance %s, connection to secondary node"
1174
                      " %s failed" % (instance, snode))
1175
          bad = True
1176
        if snode in n_offline:
1177
          inst_nodes_offline.append(snode)
1178

    
1179
      if inst_nodes_offline:
1180
        # warn that the instance lives on offline nodes, and set bad=True
1181
        feedback_fn("  - ERROR: instance lives on offline node(s) %s" %
1182
                    ", ".join(inst_nodes_offline))
1183
        bad = True
1184

    
1185
    feedback_fn("* Verifying orphan volumes")
1186
    result = self._VerifyOrphanVolumes(node_vol_should, node_volume,
1187
                                       feedback_fn)
1188
    bad = bad or result
1189

    
1190
    feedback_fn("* Verifying remaining instances")
1191
    result = self._VerifyOrphanInstances(instancelist, node_instance,
1192
                                         feedback_fn)
1193
    bad = bad or result
1194

    
1195
    if constants.VERIFY_NPLUSONE_MEM not in self.skip_set:
1196
      feedback_fn("* Verifying N+1 Memory redundancy")
1197
      result = self._VerifyNPlusOneMemory(node_info, instance_cfg, feedback_fn)
1198
      bad = bad or result
1199

    
1200
    feedback_fn("* Other Notes")
1201
    if i_non_redundant:
1202
      feedback_fn("  - NOTICE: %d non-redundant instance(s) found."
1203
                  % len(i_non_redundant))
1204

    
1205
    if i_non_a_balanced:
1206
      feedback_fn("  - NOTICE: %d non-auto-balanced instance(s) found."
1207
                  % len(i_non_a_balanced))
1208

    
1209
    if n_offline:
1210
      feedback_fn("  - NOTICE: %d offline node(s) found." % len(n_offline))
1211

    
1212
    if n_drained:
1213
      feedback_fn("  - NOTICE: %d drained node(s) found." % len(n_drained))
1214

    
1215
    return not bad
1216

    
1217
  def HooksCallBack(self, phase, hooks_results, feedback_fn, lu_result):
1218
    """Analize the post-hooks' result
1219

1220
    This method analyses the hook result, handles it, and sends some
1221
    nicely-formatted feedback back to the user.
1222

1223
    @param phase: one of L{constants.HOOKS_PHASE_POST} or
1224
        L{constants.HOOKS_PHASE_PRE}; it denotes the hooks phase
1225
    @param hooks_results: the results of the multi-node hooks rpc call
1226
    @param feedback_fn: function used send feedback back to the caller
1227
    @param lu_result: previous Exec result
1228
    @return: the new Exec result, based on the previous result
1229
        and hook results
1230

1231
    """
1232
    # We only really run POST phase hooks, and are only interested in
1233
    # their results
1234
    if phase == constants.HOOKS_PHASE_POST:
1235
      # Used to change hooks' output to proper indentation
1236
      indent_re = re.compile('^', re.M)
1237
      feedback_fn("* Hooks Results")
1238
      if not hooks_results:
1239
        feedback_fn("  - ERROR: general communication failure")
1240
        lu_result = 1
1241
      else:
1242
        for node_name in hooks_results:
1243
          show_node_header = True
1244
          res = hooks_results[node_name]
1245
          msg = res.RemoteFailMsg()
1246
          if msg:
1247
            if res.offline:
1248
              # no need to warn or set fail return value
1249
              continue
1250
            feedback_fn("    Communication failure in hooks execution: %s" %
1251
                        msg)
1252
            lu_result = 1
1253
            continue
1254
          for script, hkr, output in res.payload:
1255
            if hkr == constants.HKR_FAIL:
1256
              # The node header is only shown once, if there are
1257
              # failing hooks on that node
1258
              if show_node_header:
1259
                feedback_fn("  Node %s:" % node_name)
1260
                show_node_header = False
1261
              feedback_fn("    ERROR: Script %s failed, output:" % script)
1262
              output = indent_re.sub('      ', output)
1263
              feedback_fn("%s" % output)
1264
              lu_result = 1
1265

    
1266
      return lu_result
1267

    
1268

    
1269
class LUVerifyDisks(NoHooksLU):
1270
  """Verifies the cluster disks status.
1271

1272
  """
1273
  _OP_REQP = []
1274
  REQ_BGL = False
1275

    
1276
  def ExpandNames(self):
1277
    self.needed_locks = {
1278
      locking.LEVEL_NODE: locking.ALL_SET,
1279
      locking.LEVEL_INSTANCE: locking.ALL_SET,
1280
    }
1281
    self.share_locks = dict(((i, 1) for i in locking.LEVELS))
1282

    
1283
  def CheckPrereq(self):
1284
    """Check prerequisites.
1285

1286
    This has no prerequisites.
1287

1288
    """
1289
    pass
1290

    
1291
  def Exec(self, feedback_fn):
1292
    """Verify integrity of cluster disks.
1293

1294
    @rtype: tuple of three items
1295
    @return: a tuple of (dict of node-to-node_error, list of instances
1296
        which need activate-disks, dict of instance: (node, volume) for
1297
        missing volumes
1298

1299
    """
1300
    result = res_nodes, res_instances, res_missing = {}, [], {}
1301

    
1302
    vg_name = self.cfg.GetVGName()
1303
    nodes = utils.NiceSort(self.cfg.GetNodeList())
1304
    instances = [self.cfg.GetInstanceInfo(name)
1305
                 for name in self.cfg.GetInstanceList()]
1306

    
1307
    nv_dict = {}
1308
    for inst in instances:
1309
      inst_lvs = {}
1310
      if (not inst.admin_up or
1311
          inst.disk_template not in constants.DTS_NET_MIRROR):
1312
        continue
1313
      inst.MapLVsByNode(inst_lvs)
1314
      # transform { iname: {node: [vol,],},} to {(node, vol): iname}
1315
      for node, vol_list in inst_lvs.iteritems():
1316
        for vol in vol_list:
1317
          nv_dict[(node, vol)] = inst
1318

    
1319
    if not nv_dict:
1320
      return result
1321

    
1322
    node_lvs = self.rpc.call_volume_list(nodes, vg_name)
1323

    
1324
    to_act = set()
1325
    for node in nodes:
1326
      # node_volume
1327
      node_res = node_lvs[node]
1328
      if node_res.offline:
1329
        continue
1330
      msg = node_res.RemoteFailMsg()
1331
      if msg:
1332
        logging.warning("Error enumerating LVs on node %s: %s", node, msg)
1333
        res_nodes[node] = msg
1334
        continue
1335

    
1336
      lvs = node_res.payload
1337
      for lv_name, (_, lv_inactive, lv_online) in lvs.items():
1338
        inst = nv_dict.pop((node, lv_name), None)
1339
        if (not lv_online and inst is not None
1340
            and inst.name not in res_instances):
1341
          res_instances.append(inst.name)
1342

    
1343
    # any leftover items in nv_dict are missing LVs, let's arrange the
1344
    # data better
1345
    for key, inst in nv_dict.iteritems():
1346
      if inst.name not in res_missing:
1347
        res_missing[inst.name] = []
1348
      res_missing[inst.name].append(key)
1349

    
1350
    return result
1351

    
1352

    
1353
class LURenameCluster(LogicalUnit):
1354
  """Rename the cluster.
1355

1356
  """
1357
  HPATH = "cluster-rename"
1358
  HTYPE = constants.HTYPE_CLUSTER
1359
  _OP_REQP = ["name"]
1360

    
1361
  def BuildHooksEnv(self):
1362
    """Build hooks env.
1363

1364
    """
1365
    env = {
1366
      "OP_TARGET": self.cfg.GetClusterName(),
1367
      "NEW_NAME": self.op.name,
1368
      }
1369
    mn = self.cfg.GetMasterNode()
1370
    return env, [mn], [mn]
1371

    
1372
  def CheckPrereq(self):
1373
    """Verify that the passed name is a valid one.
1374

1375
    """
1376
    hostname = utils.HostInfo(self.op.name)
1377

    
1378
    new_name = hostname.name
1379
    self.ip = new_ip = hostname.ip
1380
    old_name = self.cfg.GetClusterName()
1381
    old_ip = self.cfg.GetMasterIP()
1382
    if new_name == old_name and new_ip == old_ip:
1383
      raise errors.OpPrereqError("Neither the name nor the IP address of the"
1384
                                 " cluster has changed")
1385
    if new_ip != old_ip:
1386
      if utils.TcpPing(new_ip, constants.DEFAULT_NODED_PORT):
1387
        raise errors.OpPrereqError("The given cluster IP address (%s) is"
1388
                                   " reachable on the network. Aborting." %
1389
                                   new_ip)
1390

    
1391
    self.op.name = new_name
1392

    
1393
  def Exec(self, feedback_fn):
1394
    """Rename the cluster.
1395

1396
    """
1397
    clustername = self.op.name
1398
    ip = self.ip
1399

    
1400
    # shutdown the master IP
1401
    master = self.cfg.GetMasterNode()
1402
    result = self.rpc.call_node_stop_master(master, False)
1403
    msg = result.RemoteFailMsg()
1404
    if msg:
1405
      raise errors.OpExecError("Could not disable the master role: %s" % msg)
1406

    
1407
    try:
1408
      cluster = self.cfg.GetClusterInfo()
1409
      cluster.cluster_name = clustername
1410
      cluster.master_ip = ip
1411
      self.cfg.Update(cluster)
1412

    
1413
      # update the known hosts file
1414
      ssh.WriteKnownHostsFile(self.cfg, constants.SSH_KNOWN_HOSTS_FILE)
1415
      node_list = self.cfg.GetNodeList()
1416
      try:
1417
        node_list.remove(master)
1418
      except ValueError:
1419
        pass
1420
      result = self.rpc.call_upload_file(node_list,
1421
                                         constants.SSH_KNOWN_HOSTS_FILE)
1422
      for to_node, to_result in result.iteritems():
1423
         msg = to_result.RemoteFailMsg()
1424
         if msg:
1425
           msg = ("Copy of file %s to node %s failed: %s" %
1426
                   (constants.SSH_KNOWN_HOSTS_FILE, to_node, msg))
1427
           self.proc.LogWarning(msg)
1428

    
1429
    finally:
1430
      result = self.rpc.call_node_start_master(master, False)
1431
      msg = result.RemoteFailMsg()
1432
      if msg:
1433
        self.LogWarning("Could not re-enable the master role on"
1434
                        " the master, please restart manually: %s", msg)
1435

    
1436

    
1437
def _RecursiveCheckIfLVMBased(disk):
1438
  """Check if the given disk or its children are lvm-based.
1439

1440
  @type disk: L{objects.Disk}
1441
  @param disk: the disk to check
1442
  @rtype: booleean
1443
  @return: boolean indicating whether a LD_LV dev_type was found or not
1444

1445
  """
1446
  if disk.children:
1447
    for chdisk in disk.children:
1448
      if _RecursiveCheckIfLVMBased(chdisk):
1449
        return True
1450
  return disk.dev_type == constants.LD_LV
1451

    
1452

    
1453
class LUSetClusterParams(LogicalUnit):
1454
  """Change the parameters of the cluster.
1455

1456
  """
1457
  HPATH = "cluster-modify"
1458
  HTYPE = constants.HTYPE_CLUSTER
1459
  _OP_REQP = []
1460
  REQ_BGL = False
1461

    
1462
  def CheckArguments(self):
1463
    """Check parameters
1464

1465
    """
1466
    if not hasattr(self.op, "candidate_pool_size"):
1467
      self.op.candidate_pool_size = None
1468
    if self.op.candidate_pool_size is not None:
1469
      try:
1470
        self.op.candidate_pool_size = int(self.op.candidate_pool_size)
1471
      except (ValueError, TypeError), err:
1472
        raise errors.OpPrereqError("Invalid candidate_pool_size value: %s" %
1473
                                   str(err))
1474
      if self.op.candidate_pool_size < 1:
1475
        raise errors.OpPrereqError("At least one master candidate needed")
1476

    
1477
  def ExpandNames(self):
1478
    # FIXME: in the future maybe other cluster params won't require checking on
1479
    # all nodes to be modified.
1480
    self.needed_locks = {
1481
      locking.LEVEL_NODE: locking.ALL_SET,
1482
    }
1483
    self.share_locks[locking.LEVEL_NODE] = 1
1484

    
1485
  def BuildHooksEnv(self):
1486
    """Build hooks env.
1487

1488
    """
1489
    env = {
1490
      "OP_TARGET": self.cfg.GetClusterName(),
1491
      "NEW_VG_NAME": self.op.vg_name,
1492
      }
1493
    mn = self.cfg.GetMasterNode()
1494
    return env, [mn], [mn]
1495

    
1496
  def CheckPrereq(self):
1497
    """Check prerequisites.
1498

1499
    This checks whether the given params don't conflict and
1500
    if the given volume group is valid.
1501

1502
    """
1503
    if self.op.vg_name is not None and not self.op.vg_name:
1504
      instances = self.cfg.GetAllInstancesInfo().values()
1505
      for inst in instances:
1506
        for disk in inst.disks:
1507
          if _RecursiveCheckIfLVMBased(disk):
1508
            raise errors.OpPrereqError("Cannot disable lvm storage while"
1509
                                       " lvm-based instances exist")
1510

    
1511
    node_list = self.acquired_locks[locking.LEVEL_NODE]
1512

    
1513
    # if vg_name not None, checks given volume group on all nodes
1514
    if self.op.vg_name:
1515
      vglist = self.rpc.call_vg_list(node_list)
1516
      for node in node_list:
1517
        msg = vglist[node].RemoteFailMsg()
1518
        if msg:
1519
          # ignoring down node
1520
          self.LogWarning("Error while gathering data on node %s"
1521
                          " (ignoring node): %s", node, msg)
1522
          continue
1523
        vgstatus = utils.CheckVolumeGroupSize(vglist[node].payload,
1524
                                              self.op.vg_name,
1525
                                              constants.MIN_VG_SIZE)
1526
        if vgstatus:
1527
          raise errors.OpPrereqError("Error on node '%s': %s" %
1528
                                     (node, vgstatus))
1529

    
1530
    self.cluster = cluster = self.cfg.GetClusterInfo()
1531
    # validate params changes
1532
    if self.op.beparams:
1533
      utils.ForceDictType(self.op.beparams, constants.BES_PARAMETER_TYPES)
1534
      self.new_beparams = objects.FillDict(
1535
        cluster.beparams[constants.PP_DEFAULT], self.op.beparams)
1536

    
1537
    if self.op.nicparams:
1538
      utils.ForceDictType(self.op.nicparams, constants.NICS_PARAMETER_TYPES)
1539
      self.new_nicparams = objects.FillDict(
1540
        cluster.nicparams[constants.PP_DEFAULT], self.op.nicparams)
1541
      objects.NIC.CheckParameterSyntax(self.new_nicparams)
1542

    
1543
    # hypervisor list/parameters
1544
    self.new_hvparams = objects.FillDict(cluster.hvparams, {})
1545
    if self.op.hvparams:
1546
      if not isinstance(self.op.hvparams, dict):
1547
        raise errors.OpPrereqError("Invalid 'hvparams' parameter on input")
1548
      for hv_name, hv_dict in self.op.hvparams.items():
1549
        if hv_name not in self.new_hvparams:
1550
          self.new_hvparams[hv_name] = hv_dict
1551
        else:
1552
          self.new_hvparams[hv_name].update(hv_dict)
1553

    
1554
    if self.op.enabled_hypervisors is not None:
1555
      self.hv_list = self.op.enabled_hypervisors
1556
    else:
1557
      self.hv_list = cluster.enabled_hypervisors
1558

    
1559
    if self.op.hvparams or self.op.enabled_hypervisors is not None:
1560
      # either the enabled list has changed, or the parameters have, validate
1561
      for hv_name, hv_params in self.new_hvparams.items():
1562
        if ((self.op.hvparams and hv_name in self.op.hvparams) or
1563
            (self.op.enabled_hypervisors and
1564
             hv_name in self.op.enabled_hypervisors)):
1565
          # either this is a new hypervisor, or its parameters have changed
1566
          hv_class = hypervisor.GetHypervisor(hv_name)
1567
          utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
1568
          hv_class.CheckParameterSyntax(hv_params)
1569
          _CheckHVParams(self, node_list, hv_name, hv_params)
1570

    
1571
  def Exec(self, feedback_fn):
1572
    """Change the parameters of the cluster.
1573

1574
    """
1575
    if self.op.vg_name is not None:
1576
      new_volume = self.op.vg_name
1577
      if not new_volume:
1578
        new_volume = None
1579
      if new_volume != self.cfg.GetVGName():
1580
        self.cfg.SetVGName(new_volume)
1581
      else:
1582
        feedback_fn("Cluster LVM configuration already in desired"
1583
                    " state, not changing")
1584
    if self.op.hvparams:
1585
      self.cluster.hvparams = self.new_hvparams
1586
    if self.op.enabled_hypervisors is not None:
1587
      self.cluster.enabled_hypervisors = self.op.enabled_hypervisors
1588
    if self.op.beparams:
1589
      self.cluster.beparams[constants.PP_DEFAULT] = self.new_beparams
1590
    if self.op.nicparams:
1591
      self.cluster.nicparams[constants.PP_DEFAULT] = self.new_nicparams
1592

    
1593
    if self.op.candidate_pool_size is not None:
1594
      self.cluster.candidate_pool_size = self.op.candidate_pool_size
1595

    
1596
    self.cfg.Update(self.cluster)
1597

    
1598
    # we want to update nodes after the cluster so that if any errors
1599
    # happen, we have recorded and saved the cluster info
1600
    if self.op.candidate_pool_size is not None:
1601
      _AdjustCandidatePool(self)
1602

    
1603

    
1604
def _RedistributeAncillaryFiles(lu, additional_nodes=None):
1605
  """Distribute additional files which are part of the cluster configuration.
1606

1607
  ConfigWriter takes care of distributing the config and ssconf files, but
1608
  there are more files which should be distributed to all nodes. This function
1609
  makes sure those are copied.
1610

1611
  @param lu: calling logical unit
1612
  @param additional_nodes: list of nodes not in the config to distribute to
1613

1614
  """
1615
  # 1. Gather target nodes
1616
  myself = lu.cfg.GetNodeInfo(lu.cfg.GetMasterNode())
1617
  dist_nodes = lu.cfg.GetNodeList()
1618
  if additional_nodes is not None:
1619
    dist_nodes.extend(additional_nodes)
1620
  if myself.name in dist_nodes:
1621
    dist_nodes.remove(myself.name)
1622
  # 2. Gather files to distribute
1623
  dist_files = set([constants.ETC_HOSTS,
1624
                    constants.SSH_KNOWN_HOSTS_FILE,
1625
                    constants.RAPI_CERT_FILE,
1626
                    constants.RAPI_USERS_FILE,
1627
                   ])
1628

    
1629
  enabled_hypervisors = lu.cfg.GetClusterInfo().enabled_hypervisors
1630
  for hv_name in enabled_hypervisors:
1631
    hv_class = hypervisor.GetHypervisor(hv_name)
1632
    dist_files.update(hv_class.GetAncillaryFiles())
1633

    
1634
  # 3. Perform the files upload
1635
  for fname in dist_files:
1636
    if os.path.exists(fname):
1637
      result = lu.rpc.call_upload_file(dist_nodes, fname)
1638
      for to_node, to_result in result.items():
1639
         msg = to_result.RemoteFailMsg()
1640
         if msg:
1641
           msg = ("Copy of file %s to node %s failed: %s" %
1642
                   (fname, to_node, msg))
1643
           lu.proc.LogWarning(msg)
1644

    
1645

    
1646
class LURedistributeConfig(NoHooksLU):
1647
  """Force the redistribution of cluster configuration.
1648

1649
  This is a very simple LU.
1650

1651
  """
1652
  _OP_REQP = []
1653
  REQ_BGL = False
1654

    
1655
  def ExpandNames(self):
1656
    self.needed_locks = {
1657
      locking.LEVEL_NODE: locking.ALL_SET,
1658
    }
1659
    self.share_locks[locking.LEVEL_NODE] = 1
1660

    
1661
  def CheckPrereq(self):
1662
    """Check prerequisites.
1663

1664
    """
1665

    
1666
  def Exec(self, feedback_fn):
1667
    """Redistribute the configuration.
1668

1669
    """
1670
    self.cfg.Update(self.cfg.GetClusterInfo())
1671
    _RedistributeAncillaryFiles(self)
1672

    
1673

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

1677
  """
1678
  if not instance.disks:
1679
    return True
1680

    
1681
  if not oneshot:
1682
    lu.proc.LogInfo("Waiting for instance %s to sync disks." % instance.name)
1683

    
1684
  node = instance.primary_node
1685

    
1686
  for dev in instance.disks:
1687
    lu.cfg.SetDiskID(dev, node)
1688

    
1689
  retries = 0
1690
  while True:
1691
    max_time = 0
1692
    done = True
1693
    cumul_degraded = False
1694
    rstats = lu.rpc.call_blockdev_getmirrorstatus(node, instance.disks)
1695
    msg = rstats.RemoteFailMsg()
1696
    if msg:
1697
      lu.LogWarning("Can't get any data from node %s: %s", node, msg)
1698
      retries += 1
1699
      if retries >= 10:
1700
        raise errors.RemoteError("Can't contact node %s for mirror data,"
1701
                                 " aborting." % node)
1702
      time.sleep(6)
1703
      continue
1704
    rstats = rstats.payload
1705
    retries = 0
1706
    for i, mstat in enumerate(rstats):
1707
      if mstat is None:
1708
        lu.LogWarning("Can't compute data for node %s/%s",
1709
                           node, instance.disks[i].iv_name)
1710
        continue
1711
      # we ignore the ldisk parameter
1712
      perc_done, est_time, is_degraded, _ = mstat
1713
      cumul_degraded = cumul_degraded or (is_degraded and perc_done is None)
1714
      if perc_done is not None:
1715
        done = False
1716
        if est_time is not None:
1717
          rem_time = "%d estimated seconds remaining" % est_time
1718
          max_time = est_time
1719
        else:
1720
          rem_time = "no time estimate"
1721
        lu.proc.LogInfo("- device %s: %5.2f%% done, %s" %
1722
                        (instance.disks[i].iv_name, perc_done, rem_time))
1723
    if done or oneshot:
1724
      break
1725

    
1726
    time.sleep(min(60, max_time))
1727

    
1728
  if done:
1729
    lu.proc.LogInfo("Instance %s's disks are in sync." % instance.name)
1730
  return not cumul_degraded
1731

    
1732

    
1733
def _CheckDiskConsistency(lu, dev, node, on_primary, ldisk=False):
1734
  """Check that mirrors are not degraded.
1735

1736
  The ldisk parameter, if True, will change the test from the
1737
  is_degraded attribute (which represents overall non-ok status for
1738
  the device(s)) to the ldisk (representing the local storage status).
1739

1740
  """
1741
  lu.cfg.SetDiskID(dev, node)
1742
  if ldisk:
1743
    idx = 6
1744
  else:
1745
    idx = 5
1746

    
1747
  result = True
1748
  if on_primary or dev.AssembleOnSecondary():
1749
    rstats = lu.rpc.call_blockdev_find(node, dev)
1750
    msg = rstats.RemoteFailMsg()
1751
    if msg:
1752
      lu.LogWarning("Can't find disk on node %s: %s", node, msg)
1753
      result = False
1754
    elif not rstats.payload:
1755
      lu.LogWarning("Can't find disk on node %s", node)
1756
      result = False
1757
    else:
1758
      result = result and (not rstats.payload[idx])
1759
  if dev.children:
1760
    for child in dev.children:
1761
      result = result and _CheckDiskConsistency(lu, child, node, on_primary)
1762

    
1763
  return result
1764

    
1765

    
1766
class LUDiagnoseOS(NoHooksLU):
1767
  """Logical unit for OS diagnose/query.
1768

1769
  """
1770
  _OP_REQP = ["output_fields", "names"]
1771
  REQ_BGL = False
1772
  _FIELDS_STATIC = utils.FieldSet()
1773
  _FIELDS_DYNAMIC = utils.FieldSet("name", "valid", "node_status")
1774

    
1775
  def ExpandNames(self):
1776
    if self.op.names:
1777
      raise errors.OpPrereqError("Selective OS query not supported")
1778

    
1779
    _CheckOutputFields(static=self._FIELDS_STATIC,
1780
                       dynamic=self._FIELDS_DYNAMIC,
1781
                       selected=self.op.output_fields)
1782

    
1783
    # Lock all nodes, in shared mode
1784
    # Temporary removal of locks, should be reverted later
1785
    # TODO: reintroduce locks when they are lighter-weight
1786
    self.needed_locks = {}
1787
    #self.share_locks[locking.LEVEL_NODE] = 1
1788
    #self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
1789

    
1790
  def CheckPrereq(self):
1791
    """Check prerequisites.
1792

1793
    """
1794

    
1795
  @staticmethod
1796
  def _DiagnoseByOS(node_list, rlist):
1797
    """Remaps a per-node return list into an a per-os per-node dictionary
1798

1799
    @param node_list: a list with the names of all nodes
1800
    @param rlist: a map with node names as keys and OS objects as values
1801

1802
    @rtype: dict
1803
    @return: a dictionary with osnames as keys and as value another map, with
1804
        nodes as keys and list of OS objects as values, eg::
1805

1806
          {"debian-etch": {"node1": [<object>,...],
1807
                           "node2": [<object>,]}
1808
          }
1809

1810
    """
1811
    all_os = {}
1812
    # we build here the list of nodes that didn't fail the RPC (at RPC
1813
    # level), so that nodes with a non-responding node daemon don't
1814
    # make all OSes invalid
1815
    good_nodes = [node_name for node_name in rlist
1816
                  if not rlist[node_name].RemoteFailMsg()]
1817
    for node_name, nr in rlist.items():
1818
      if nr.RemoteFailMsg() or not nr.payload:
1819
        continue
1820
      for os_serialized in nr.payload:
1821
        os_obj = objects.OS.FromDict(os_serialized)
1822
        if os_obj.name not in all_os:
1823
          # build a list of nodes for this os containing empty lists
1824
          # for each node in node_list
1825
          all_os[os_obj.name] = {}
1826
          for nname in good_nodes:
1827
            all_os[os_obj.name][nname] = []
1828
        all_os[os_obj.name][node_name].append(os_obj)
1829
    return all_os
1830

    
1831
  def Exec(self, feedback_fn):
1832
    """Compute the list of OSes.
1833

1834
    """
1835
    valid_nodes = [node for node in self.cfg.GetOnlineNodeList()]
1836
    node_data = self.rpc.call_os_diagnose(valid_nodes)
1837
    pol = self._DiagnoseByOS(valid_nodes, node_data)
1838
    output = []
1839
    for os_name, os_data in pol.items():
1840
      row = []
1841
      for field in self.op.output_fields:
1842
        if field == "name":
1843
          val = os_name
1844
        elif field == "valid":
1845
          val = utils.all([osl and osl[0] for osl in os_data.values()])
1846
        elif field == "node_status":
1847
          val = {}
1848
          for node_name, nos_list in os_data.iteritems():
1849
            val[node_name] = [(v.status, v.path) for v in nos_list]
1850
        else:
1851
          raise errors.ParameterError(field)
1852
        row.append(val)
1853
      output.append(row)
1854

    
1855
    return output
1856

    
1857

    
1858
class LURemoveNode(LogicalUnit):
1859
  """Logical unit for removing a node.
1860

1861
  """
1862
  HPATH = "node-remove"
1863
  HTYPE = constants.HTYPE_NODE
1864
  _OP_REQP = ["node_name"]
1865

    
1866
  def BuildHooksEnv(self):
1867
    """Build hooks env.
1868

1869
    This doesn't run on the target node in the pre phase as a failed
1870
    node would then be impossible to remove.
1871

1872
    """
1873
    env = {
1874
      "OP_TARGET": self.op.node_name,
1875
      "NODE_NAME": self.op.node_name,
1876
      }
1877
    all_nodes = self.cfg.GetNodeList()
1878
    all_nodes.remove(self.op.node_name)
1879
    return env, all_nodes, all_nodes
1880

    
1881
  def CheckPrereq(self):
1882
    """Check prerequisites.
1883

1884
    This checks:
1885
     - the node exists in the configuration
1886
     - it does not have primary or secondary instances
1887
     - it's not the master
1888

1889
    Any errors are signalled by raising errors.OpPrereqError.
1890

1891
    """
1892
    node = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.node_name))
1893
    if node is None:
1894
      raise errors.OpPrereqError, ("Node '%s' is unknown." % self.op.node_name)
1895

    
1896
    instance_list = self.cfg.GetInstanceList()
1897

    
1898
    masternode = self.cfg.GetMasterNode()
1899
    if node.name == masternode:
1900
      raise errors.OpPrereqError("Node is the master node,"
1901
                                 " you need to failover first.")
1902

    
1903
    for instance_name in instance_list:
1904
      instance = self.cfg.GetInstanceInfo(instance_name)
1905
      if node.name in instance.all_nodes:
1906
        raise errors.OpPrereqError("Instance %s is still running on the node,"
1907
                                   " please remove first." % instance_name)
1908
    self.op.node_name = node.name
1909
    self.node = node
1910

    
1911
  def Exec(self, feedback_fn):
1912
    """Removes the node from the cluster.
1913

1914
    """
1915
    node = self.node
1916
    logging.info("Stopping the node daemon and removing configs from node %s",
1917
                 node.name)
1918

    
1919
    self.context.RemoveNode(node.name)
1920

    
1921
    result = self.rpc.call_node_leave_cluster(node.name)
1922
    msg = result.RemoteFailMsg()
1923
    if msg:
1924
      self.LogWarning("Errors encountered on the remote node while leaving"
1925
                      " the cluster: %s", msg)
1926

    
1927
    # Promote nodes to master candidate as needed
1928
    _AdjustCandidatePool(self)
1929

    
1930

    
1931
class LUQueryNodes(NoHooksLU):
1932
  """Logical unit for querying nodes.
1933

1934
  """
1935
  _OP_REQP = ["output_fields", "names", "use_locking"]
1936
  REQ_BGL = False
1937
  _FIELDS_DYNAMIC = utils.FieldSet(
1938
    "dtotal", "dfree",
1939
    "mtotal", "mnode", "mfree",
1940
    "bootid",
1941
    "ctotal", "cnodes", "csockets",
1942
    )
1943

    
1944
  _FIELDS_STATIC = utils.FieldSet(
1945
    "name", "pinst_cnt", "sinst_cnt",
1946
    "pinst_list", "sinst_list",
1947
    "pip", "sip", "tags",
1948
    "serial_no",
1949
    "master_candidate",
1950
    "master",
1951
    "offline",
1952
    "drained",
1953
    )
1954

    
1955
  def ExpandNames(self):
1956
    _CheckOutputFields(static=self._FIELDS_STATIC,
1957
                       dynamic=self._FIELDS_DYNAMIC,
1958
                       selected=self.op.output_fields)
1959

    
1960
    self.needed_locks = {}
1961
    self.share_locks[locking.LEVEL_NODE] = 1
1962

    
1963
    if self.op.names:
1964
      self.wanted = _GetWantedNodes(self, self.op.names)
1965
    else:
1966
      self.wanted = locking.ALL_SET
1967

    
1968
    self.do_node_query = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
1969
    self.do_locking = self.do_node_query and self.op.use_locking
1970
    if self.do_locking:
1971
      # if we don't request only static fields, we need to lock the nodes
1972
      self.needed_locks[locking.LEVEL_NODE] = self.wanted
1973

    
1974

    
1975
  def CheckPrereq(self):
1976
    """Check prerequisites.
1977

1978
    """
1979
    # The validation of the node list is done in the _GetWantedNodes,
1980
    # if non empty, and if empty, there's no validation to do
1981
    pass
1982

    
1983
  def Exec(self, feedback_fn):
1984
    """Computes the list of nodes and their attributes.
1985

1986
    """
1987
    all_info = self.cfg.GetAllNodesInfo()
1988
    if self.do_locking:
1989
      nodenames = self.acquired_locks[locking.LEVEL_NODE]
1990
    elif self.wanted != locking.ALL_SET:
1991
      nodenames = self.wanted
1992
      missing = set(nodenames).difference(all_info.keys())
1993
      if missing:
1994
        raise errors.OpExecError(
1995
          "Some nodes were removed before retrieving their data: %s" % missing)
1996
    else:
1997
      nodenames = all_info.keys()
1998

    
1999
    nodenames = utils.NiceSort(nodenames)
2000
    nodelist = [all_info[name] for name in nodenames]
2001

    
2002
    # begin data gathering
2003

    
2004
    if self.do_node_query:
2005
      live_data = {}
2006
      node_data = self.rpc.call_node_info(nodenames, self.cfg.GetVGName(),
2007
                                          self.cfg.GetHypervisorType())
2008
      for name in nodenames:
2009
        nodeinfo = node_data[name]
2010
        if not nodeinfo.RemoteFailMsg() and nodeinfo.payload:
2011
          nodeinfo = nodeinfo.payload
2012
          fn = utils.TryConvert
2013
          live_data[name] = {
2014
            "mtotal": fn(int, nodeinfo.get('memory_total', None)),
2015
            "mnode": fn(int, nodeinfo.get('memory_dom0', None)),
2016
            "mfree": fn(int, nodeinfo.get('memory_free', None)),
2017
            "dtotal": fn(int, nodeinfo.get('vg_size', None)),
2018
            "dfree": fn(int, nodeinfo.get('vg_free', None)),
2019
            "ctotal": fn(int, nodeinfo.get('cpu_total', None)),
2020
            "bootid": nodeinfo.get('bootid', None),
2021
            "cnodes": fn(int, nodeinfo.get('cpu_nodes', None)),
2022
            "csockets": fn(int, nodeinfo.get('cpu_sockets', None)),
2023
            }
2024
        else:
2025
          live_data[name] = {}
2026
    else:
2027
      live_data = dict.fromkeys(nodenames, {})
2028

    
2029
    node_to_primary = dict([(name, set()) for name in nodenames])
2030
    node_to_secondary = dict([(name, set()) for name in nodenames])
2031

    
2032
    inst_fields = frozenset(("pinst_cnt", "pinst_list",
2033
                             "sinst_cnt", "sinst_list"))
2034
    if inst_fields & frozenset(self.op.output_fields):
2035
      instancelist = self.cfg.GetInstanceList()
2036

    
2037
      for instance_name in instancelist:
2038
        inst = self.cfg.GetInstanceInfo(instance_name)
2039
        if inst.primary_node in node_to_primary:
2040
          node_to_primary[inst.primary_node].add(inst.name)
2041
        for secnode in inst.secondary_nodes:
2042
          if secnode in node_to_secondary:
2043
            node_to_secondary[secnode].add(inst.name)
2044

    
2045
    master_node = self.cfg.GetMasterNode()
2046

    
2047
    # end data gathering
2048

    
2049
    output = []
2050
    for node in nodelist:
2051
      node_output = []
2052
      for field in self.op.output_fields:
2053
        if field == "name":
2054
          val = node.name
2055
        elif field == "pinst_list":
2056
          val = list(node_to_primary[node.name])
2057
        elif field == "sinst_list":
2058
          val = list(node_to_secondary[node.name])
2059
        elif field == "pinst_cnt":
2060
          val = len(node_to_primary[node.name])
2061
        elif field == "sinst_cnt":
2062
          val = len(node_to_secondary[node.name])
2063
        elif field == "pip":
2064
          val = node.primary_ip
2065
        elif field == "sip":
2066
          val = node.secondary_ip
2067
        elif field == "tags":
2068
          val = list(node.GetTags())
2069
        elif field == "serial_no":
2070
          val = node.serial_no
2071
        elif field == "master_candidate":
2072
          val = node.master_candidate
2073
        elif field == "master":
2074
          val = node.name == master_node
2075
        elif field == "offline":
2076
          val = node.offline
2077
        elif field == "drained":
2078
          val = node.drained
2079
        elif self._FIELDS_DYNAMIC.Matches(field):
2080
          val = live_data[node.name].get(field, None)
2081
        else:
2082
          raise errors.ParameterError(field)
2083
        node_output.append(val)
2084
      output.append(node_output)
2085

    
2086
    return output
2087

    
2088

    
2089
class LUQueryNodeVolumes(NoHooksLU):
2090
  """Logical unit for getting volumes on node(s).
2091

2092
  """
2093
  _OP_REQP = ["nodes", "output_fields"]
2094
  REQ_BGL = False
2095
  _FIELDS_DYNAMIC = utils.FieldSet("phys", "vg", "name", "size", "instance")
2096
  _FIELDS_STATIC = utils.FieldSet("node")
2097

    
2098
  def ExpandNames(self):
2099
    _CheckOutputFields(static=self._FIELDS_STATIC,
2100
                       dynamic=self._FIELDS_DYNAMIC,
2101
                       selected=self.op.output_fields)
2102

    
2103
    self.needed_locks = {}
2104
    self.share_locks[locking.LEVEL_NODE] = 1
2105
    if not self.op.nodes:
2106
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
2107
    else:
2108
      self.needed_locks[locking.LEVEL_NODE] = \
2109
        _GetWantedNodes(self, self.op.nodes)
2110

    
2111
  def CheckPrereq(self):
2112
    """Check prerequisites.
2113

2114
    This checks that the fields required are valid output fields.
2115

2116
    """
2117
    self.nodes = self.acquired_locks[locking.LEVEL_NODE]
2118

    
2119
  def Exec(self, feedback_fn):
2120
    """Computes the list of nodes and their attributes.
2121

2122
    """
2123
    nodenames = self.nodes
2124
    volumes = self.rpc.call_node_volumes(nodenames)
2125

    
2126
    ilist = [self.cfg.GetInstanceInfo(iname) for iname
2127
             in self.cfg.GetInstanceList()]
2128

    
2129
    lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
2130

    
2131
    output = []
2132
    for node in nodenames:
2133
      nresult = volumes[node]
2134
      if nresult.offline:
2135
        continue
2136
      msg = nresult.RemoteFailMsg()
2137
      if msg:
2138
        self.LogWarning("Can't compute volume data on node %s: %s", node, msg)
2139
        continue
2140

    
2141
      node_vols = nresult.payload[:]
2142
      node_vols.sort(key=lambda vol: vol['dev'])
2143

    
2144
      for vol in node_vols:
2145
        node_output = []
2146
        for field in self.op.output_fields:
2147
          if field == "node":
2148
            val = node
2149
          elif field == "phys":
2150
            val = vol['dev']
2151
          elif field == "vg":
2152
            val = vol['vg']
2153
          elif field == "name":
2154
            val = vol['name']
2155
          elif field == "size":
2156
            val = int(float(vol['size']))
2157
          elif field == "instance":
2158
            for inst in ilist:
2159
              if node not in lv_by_node[inst]:
2160
                continue
2161
              if vol['name'] in lv_by_node[inst][node]:
2162
                val = inst.name
2163
                break
2164
            else:
2165
              val = '-'
2166
          else:
2167
            raise errors.ParameterError(field)
2168
          node_output.append(str(val))
2169

    
2170
        output.append(node_output)
2171

    
2172
    return output
2173

    
2174

    
2175
class LUAddNode(LogicalUnit):
2176
  """Logical unit for adding node to the cluster.
2177

2178
  """
2179
  HPATH = "node-add"
2180
  HTYPE = constants.HTYPE_NODE
2181
  _OP_REQP = ["node_name"]
2182

    
2183
  def BuildHooksEnv(self):
2184
    """Build hooks env.
2185

2186
    This will run on all nodes before, and on all nodes + the new node after.
2187

2188
    """
2189
    env = {
2190
      "OP_TARGET": self.op.node_name,
2191
      "NODE_NAME": self.op.node_name,
2192
      "NODE_PIP": self.op.primary_ip,
2193
      "NODE_SIP": self.op.secondary_ip,
2194
      }
2195
    nodes_0 = self.cfg.GetNodeList()
2196
    nodes_1 = nodes_0 + [self.op.node_name, ]
2197
    return env, nodes_0, nodes_1
2198

    
2199
  def CheckPrereq(self):
2200
    """Check prerequisites.
2201

2202
    This checks:
2203
     - the new node is not already in the config
2204
     - it is resolvable
2205
     - its parameters (single/dual homed) matches the cluster
2206

2207
    Any errors are signalled by raising errors.OpPrereqError.
2208

2209
    """
2210
    node_name = self.op.node_name
2211
    cfg = self.cfg
2212

    
2213
    dns_data = utils.HostInfo(node_name)
2214

    
2215
    node = dns_data.name
2216
    primary_ip = self.op.primary_ip = dns_data.ip
2217
    secondary_ip = getattr(self.op, "secondary_ip", None)
2218
    if secondary_ip is None:
2219
      secondary_ip = primary_ip
2220
    if not utils.IsValidIP(secondary_ip):
2221
      raise errors.OpPrereqError("Invalid secondary IP given")
2222
    self.op.secondary_ip = secondary_ip
2223

    
2224
    node_list = cfg.GetNodeList()
2225
    if not self.op.readd and node in node_list:
2226
      raise errors.OpPrereqError("Node %s is already in the configuration" %
2227
                                 node)
2228
    elif self.op.readd and node not in node_list:
2229
      raise errors.OpPrereqError("Node %s is not in the configuration" % node)
2230

    
2231
    for existing_node_name in node_list:
2232
      existing_node = cfg.GetNodeInfo(existing_node_name)
2233

    
2234
      if self.op.readd and node == existing_node_name:
2235
        if (existing_node.primary_ip != primary_ip or
2236
            existing_node.secondary_ip != secondary_ip):
2237
          raise errors.OpPrereqError("Readded node doesn't have the same IP"
2238
                                     " address configuration as before")
2239
        continue
2240

    
2241
      if (existing_node.primary_ip == primary_ip or
2242
          existing_node.secondary_ip == primary_ip or
2243
          existing_node.primary_ip == secondary_ip or
2244
          existing_node.secondary_ip == secondary_ip):
2245
        raise errors.OpPrereqError("New node ip address(es) conflict with"
2246
                                   " existing node %s" % existing_node.name)
2247

    
2248
    # check that the type of the node (single versus dual homed) is the
2249
    # same as for the master
2250
    myself = cfg.GetNodeInfo(self.cfg.GetMasterNode())
2251
    master_singlehomed = myself.secondary_ip == myself.primary_ip
2252
    newbie_singlehomed = secondary_ip == primary_ip
2253
    if master_singlehomed != newbie_singlehomed:
2254
      if master_singlehomed:
2255
        raise errors.OpPrereqError("The master has no private ip but the"
2256
                                   " new node has one")
2257
      else:
2258
        raise errors.OpPrereqError("The master has a private ip but the"
2259
                                   " new node doesn't have one")
2260

    
2261
    # checks reachablity
2262
    if not utils.TcpPing(primary_ip, constants.DEFAULT_NODED_PORT):
2263
      raise errors.OpPrereqError("Node not reachable by ping")
2264

    
2265
    if not newbie_singlehomed:
2266
      # check reachability from my secondary ip to newbie's secondary ip
2267
      if not utils.TcpPing(secondary_ip, constants.DEFAULT_NODED_PORT,
2268
                           source=myself.secondary_ip):
2269
        raise errors.OpPrereqError("Node secondary ip not reachable by TCP"
2270
                                   " based ping to noded port")
2271

    
2272
    cp_size = self.cfg.GetClusterInfo().candidate_pool_size
2273
    mc_now, _ = self.cfg.GetMasterCandidateStats()
2274
    master_candidate = mc_now < cp_size
2275

    
2276
    self.new_node = objects.Node(name=node,
2277
                                 primary_ip=primary_ip,
2278
                                 secondary_ip=secondary_ip,
2279
                                 master_candidate=master_candidate,
2280
                                 offline=False, drained=False)
2281

    
2282
  def Exec(self, feedback_fn):
2283
    """Adds the new node to the cluster.
2284

2285
    """
2286
    new_node = self.new_node
2287
    node = new_node.name
2288

    
2289
    # check connectivity
2290
    result = self.rpc.call_version([node])[node]
2291
    msg = result.RemoteFailMsg()
2292
    if msg:
2293
      raise errors.OpExecError("Can't get version information from"
2294
                               " node %s: %s" % (node, msg))
2295
    if constants.PROTOCOL_VERSION == result.payload:
2296
      logging.info("Communication to node %s fine, sw version %s match",
2297
                   node, result.payload)
2298
    else:
2299
      raise errors.OpExecError("Version mismatch master version %s,"
2300
                               " node version %s" %
2301
                               (constants.PROTOCOL_VERSION, result.payload))
2302

    
2303
    # setup ssh on node
2304
    logging.info("Copy ssh key to node %s", node)
2305
    priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
2306
    keyarray = []
2307
    keyfiles = [constants.SSH_HOST_DSA_PRIV, constants.SSH_HOST_DSA_PUB,
2308
                constants.SSH_HOST_RSA_PRIV, constants.SSH_HOST_RSA_PUB,
2309
                priv_key, pub_key]
2310

    
2311
    for i in keyfiles:
2312
      f = open(i, 'r')
2313
      try:
2314
        keyarray.append(f.read())
2315
      finally:
2316
        f.close()
2317

    
2318
    result = self.rpc.call_node_add(node, keyarray[0], keyarray[1],
2319
                                    keyarray[2],
2320
                                    keyarray[3], keyarray[4], keyarray[5])
2321

    
2322
    msg = result.RemoteFailMsg()
2323
    if msg:
2324
      raise errors.OpExecError("Cannot transfer ssh keys to the"
2325
                               " new node: %s" % msg)
2326

    
2327
    # Add node to our /etc/hosts, and add key to known_hosts
2328
    if self.cfg.GetClusterInfo().modify_etc_hosts:
2329
      utils.AddHostToEtcHosts(new_node.name)
2330

    
2331
    if new_node.secondary_ip != new_node.primary_ip:
2332
      result = self.rpc.call_node_has_ip_address(new_node.name,
2333
                                                 new_node.secondary_ip)
2334
      msg = result.RemoteFailMsg()
2335
      if msg:
2336
        raise errors.OpPrereqError("Failure checking secondary ip"
2337
                                   " on node %s: %s" % (new_node.name, msg))
2338
      if not result.payload:
2339
        raise errors.OpExecError("Node claims it doesn't have the secondary ip"
2340
                                 " you gave (%s). Please fix and re-run this"
2341
                                 " command." % new_node.secondary_ip)
2342

    
2343
    node_verify_list = [self.cfg.GetMasterNode()]
2344
    node_verify_param = {
2345
      'nodelist': [node],
2346
      # TODO: do a node-net-test as well?
2347
    }
2348

    
2349
    result = self.rpc.call_node_verify(node_verify_list, node_verify_param,
2350
                                       self.cfg.GetClusterName())
2351
    for verifier in node_verify_list:
2352
      msg = result[verifier].RemoteFailMsg()
2353
      if msg:
2354
        raise errors.OpExecError("Cannot communicate with node %s: %s" %
2355
                                 (verifier, msg))
2356
      nl_payload = result[verifier].payload['nodelist']
2357
      if nl_payload:
2358
        for failed in nl_payload:
2359
          feedback_fn("ssh/hostname verification failed %s -> %s" %
2360
                      (verifier, nl_payload[failed]))
2361
        raise errors.OpExecError("ssh/hostname verification failed.")
2362

    
2363
    if self.op.readd:
2364
      _RedistributeAncillaryFiles(self)
2365
      self.context.ReaddNode(new_node)
2366
    else:
2367
      _RedistributeAncillaryFiles(self, additional_nodes=[node])
2368
      self.context.AddNode(new_node)
2369

    
2370

    
2371
class LUSetNodeParams(LogicalUnit):
2372
  """Modifies the parameters of a node.
2373

2374
  """
2375
  HPATH = "node-modify"
2376
  HTYPE = constants.HTYPE_NODE
2377
  _OP_REQP = ["node_name"]
2378
  REQ_BGL = False
2379

    
2380
  def CheckArguments(self):
2381
    node_name = self.cfg.ExpandNodeName(self.op.node_name)
2382
    if node_name is None:
2383
      raise errors.OpPrereqError("Invalid node name '%s'" % self.op.node_name)
2384
    self.op.node_name = node_name
2385
    _CheckBooleanOpField(self.op, 'master_candidate')
2386
    _CheckBooleanOpField(self.op, 'offline')
2387
    _CheckBooleanOpField(self.op, 'drained')
2388
    all_mods = [self.op.offline, self.op.master_candidate, self.op.drained]
2389
    if all_mods.count(None) == 3:
2390
      raise errors.OpPrereqError("Please pass at least one modification")
2391
    if all_mods.count(True) > 1:
2392
      raise errors.OpPrereqError("Can't set the node into more than one"
2393
                                 " state at the same time")
2394

    
2395
  def ExpandNames(self):
2396
    self.needed_locks = {locking.LEVEL_NODE: self.op.node_name}
2397

    
2398
  def BuildHooksEnv(self):
2399
    """Build hooks env.
2400

2401
    This runs on the master node.
2402

2403
    """
2404
    env = {
2405
      "OP_TARGET": self.op.node_name,
2406
      "MASTER_CANDIDATE": str(self.op.master_candidate),
2407
      "OFFLINE": str(self.op.offline),
2408
      "DRAINED": str(self.op.drained),
2409
      }
2410
    nl = [self.cfg.GetMasterNode(),
2411
          self.op.node_name]
2412
    return env, nl, nl
2413

    
2414
  def CheckPrereq(self):
2415
    """Check prerequisites.
2416

2417
    This only checks the instance list against the existing names.
2418

2419
    """
2420
    node = self.node = self.cfg.GetNodeInfo(self.op.node_name)
2421

    
2422
    if ((self.op.master_candidate == False or self.op.offline == True or
2423
         self.op.drained == True) and node.master_candidate):
2424
      # we will demote the node from master_candidate
2425
      if self.op.node_name == self.cfg.GetMasterNode():
2426
        raise errors.OpPrereqError("The master node has to be a"
2427
                                   " master candidate, online and not drained")
2428
      cp_size = self.cfg.GetClusterInfo().candidate_pool_size
2429
      num_candidates, _ = self.cfg.GetMasterCandidateStats()
2430
      if num_candidates <= cp_size:
2431
        msg = ("Not enough master candidates (desired"
2432
               " %d, new value will be %d)" % (cp_size, num_candidates-1))
2433
        if self.op.force:
2434
          self.LogWarning(msg)
2435
        else:
2436
          raise errors.OpPrereqError(msg)
2437

    
2438
    if (self.op.master_candidate == True and
2439
        ((node.offline and not self.op.offline == False) or
2440
         (node.drained and not self.op.drained == False))):
2441
      raise errors.OpPrereqError("Node '%s' is offline or drained, can't set"
2442
                                 " to master_candidate" % node.name)
2443

    
2444
    return
2445

    
2446
  def Exec(self, feedback_fn):
2447
    """Modifies a node.
2448

2449
    """
2450
    node = self.node
2451

    
2452
    result = []
2453
    changed_mc = False
2454

    
2455
    if self.op.offline is not None:
2456
      node.offline = self.op.offline
2457
      result.append(("offline", str(self.op.offline)))
2458
      if self.op.offline == True:
2459
        if node.master_candidate:
2460
          node.master_candidate = False
2461
          changed_mc = True
2462
          result.append(("master_candidate", "auto-demotion due to offline"))
2463
        if node.drained:
2464
          node.drained = False
2465
          result.append(("drained", "clear drained status due to offline"))
2466

    
2467
    if self.op.master_candidate is not None:
2468
      node.master_candidate = self.op.master_candidate
2469
      changed_mc = True
2470
      result.append(("master_candidate", str(self.op.master_candidate)))
2471
      if self.op.master_candidate == False:
2472
        rrc = self.rpc.call_node_demote_from_mc(node.name)
2473
        msg = rrc.RemoteFailMsg()
2474
        if msg:
2475
          self.LogWarning("Node failed to demote itself: %s" % msg)
2476

    
2477
    if self.op.drained is not None:
2478
      node.drained = self.op.drained
2479
      result.append(("drained", str(self.op.drained)))
2480
      if self.op.drained == True:
2481
        if node.master_candidate:
2482
          node.master_candidate = False
2483
          changed_mc = True
2484
          result.append(("master_candidate", "auto-demotion due to drain"))
2485
        if node.offline:
2486
          node.offline = False
2487
          result.append(("offline", "clear offline status due to drain"))
2488

    
2489
    # this will trigger configuration file update, if needed
2490
    self.cfg.Update(node)
2491
    # this will trigger job queue propagation or cleanup
2492
    if changed_mc:
2493
      self.context.ReaddNode(node)
2494

    
2495
    return result
2496

    
2497

    
2498
class LUPowercycleNode(NoHooksLU):
2499
  """Powercycles a node.
2500

2501
  """
2502
  _OP_REQP = ["node_name", "force"]
2503
  REQ_BGL = False
2504

    
2505
  def CheckArguments(self):
2506
    node_name = self.cfg.ExpandNodeName(self.op.node_name)
2507
    if node_name is None:
2508
      raise errors.OpPrereqError("Invalid node name '%s'" % self.op.node_name)
2509
    self.op.node_name = node_name
2510
    if node_name == self.cfg.GetMasterNode() and not self.op.force:
2511
      raise errors.OpPrereqError("The node is the master and the force"
2512
                                 " parameter was not set")
2513

    
2514
  def ExpandNames(self):
2515
    """Locking for PowercycleNode.
2516

2517
    This is a last-resource option and shouldn't block on other
2518
    jobs. Therefore, we grab no locks.
2519

2520
    """
2521
    self.needed_locks = {}
2522

    
2523
  def CheckPrereq(self):
2524
    """Check prerequisites.
2525

2526
    This LU has no prereqs.
2527

2528
    """
2529
    pass
2530

    
2531
  def Exec(self, feedback_fn):
2532
    """Reboots a node.
2533

2534
    """
2535
    result = self.rpc.call_node_powercycle(self.op.node_name,
2536
                                           self.cfg.GetHypervisorType())
2537
    msg = result.RemoteFailMsg()
2538
    if msg:
2539
      raise errors.OpExecError("Failed to schedule the reboot: %s" % msg)
2540
    return result.payload
2541

    
2542

    
2543
class LUQueryClusterInfo(NoHooksLU):
2544
  """Query cluster configuration.
2545

2546
  """
2547
  _OP_REQP = []
2548
  REQ_BGL = False
2549

    
2550
  def ExpandNames(self):
2551
    self.needed_locks = {}
2552

    
2553
  def CheckPrereq(self):
2554
    """No prerequsites needed for this LU.
2555

2556
    """
2557
    pass
2558

    
2559
  def Exec(self, feedback_fn):
2560
    """Return cluster config.
2561

2562
    """
2563
    cluster = self.cfg.GetClusterInfo()
2564
    result = {
2565
      "software_version": constants.RELEASE_VERSION,
2566
      "protocol_version": constants.PROTOCOL_VERSION,
2567
      "config_version": constants.CONFIG_VERSION,
2568
      "os_api_version": constants.OS_API_VERSION,
2569
      "export_version": constants.EXPORT_VERSION,
2570
      "architecture": (platform.architecture()[0], platform.machine()),
2571
      "name": cluster.cluster_name,
2572
      "master": cluster.master_node,
2573
      "default_hypervisor": cluster.default_hypervisor,
2574
      "enabled_hypervisors": cluster.enabled_hypervisors,
2575
      "hvparams": dict([(hypervisor, cluster.hvparams[hypervisor])
2576
                        for hypervisor in cluster.enabled_hypervisors]),
2577
      "beparams": cluster.beparams,
2578
      "nicparams": cluster.nicparams,
2579
      "candidate_pool_size": cluster.candidate_pool_size,
2580
      "master_netdev": cluster.master_netdev,
2581
      "volume_group_name": cluster.volume_group_name,
2582
      "file_storage_dir": cluster.file_storage_dir,
2583
      }
2584

    
2585
    return result
2586

    
2587

    
2588
class LUQueryConfigValues(NoHooksLU):
2589
  """Return configuration values.
2590

2591
  """
2592
  _OP_REQP = []
2593
  REQ_BGL = False
2594
  _FIELDS_DYNAMIC = utils.FieldSet()
2595
  _FIELDS_STATIC = utils.FieldSet("cluster_name", "master_node", "drain_flag")
2596

    
2597
  def ExpandNames(self):
2598
    self.needed_locks = {}
2599

    
2600
    _CheckOutputFields(static=self._FIELDS_STATIC,
2601
                       dynamic=self._FIELDS_DYNAMIC,
2602
                       selected=self.op.output_fields)
2603

    
2604
  def CheckPrereq(self):
2605
    """No prerequisites.
2606

2607
    """
2608
    pass
2609

    
2610
  def Exec(self, feedback_fn):
2611
    """Dump a representation of the cluster config to the standard output.
2612

2613
    """
2614
    values = []
2615
    for field in self.op.output_fields:
2616
      if field == "cluster_name":
2617
        entry = self.cfg.GetClusterName()
2618
      elif field == "master_node":
2619
        entry = self.cfg.GetMasterNode()
2620
      elif field == "drain_flag":
2621
        entry = os.path.exists(constants.JOB_QUEUE_DRAIN_FILE)
2622
      else:
2623
        raise errors.ParameterError(field)
2624
      values.append(entry)
2625
    return values
2626

    
2627

    
2628
class LUActivateInstanceDisks(NoHooksLU):
2629
  """Bring up an instance's disks.
2630

2631
  """
2632
  _OP_REQP = ["instance_name"]
2633
  REQ_BGL = False
2634

    
2635
  def ExpandNames(self):
2636
    self._ExpandAndLockInstance()
2637
    self.needed_locks[locking.LEVEL_NODE] = []
2638
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2639

    
2640
  def DeclareLocks(self, level):
2641
    if level == locking.LEVEL_NODE:
2642
      self._LockInstancesNodes()
2643

    
2644
  def CheckPrereq(self):
2645
    """Check prerequisites.
2646

2647
    This checks that the instance is in the cluster.
2648

2649
    """
2650
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2651
    assert self.instance is not None, \
2652
      "Cannot retrieve locked instance %s" % self.op.instance_name
2653
    _CheckNodeOnline(self, self.instance.primary_node)
2654

    
2655
  def Exec(self, feedback_fn):
2656
    """Activate the disks.
2657

2658
    """
2659
    disks_ok, disks_info = _AssembleInstanceDisks(self, self.instance)
2660
    if not disks_ok:
2661
      raise errors.OpExecError("Cannot activate block devices")
2662

    
2663
    return disks_info
2664

    
2665

    
2666
def _AssembleInstanceDisks(lu, instance, ignore_secondaries=False):
2667
  """Prepare the block devices for an instance.
2668

2669
  This sets up the block devices on all nodes.
2670

2671
  @type lu: L{LogicalUnit}
2672
  @param lu: the logical unit on whose behalf we execute
2673
  @type instance: L{objects.Instance}
2674
  @param instance: the instance for whose disks we assemble
2675
  @type ignore_secondaries: boolean
2676
  @param ignore_secondaries: if true, errors on secondary nodes
2677
      won't result in an error return from the function
2678
  @return: False if the operation failed, otherwise a list of
2679
      (host, instance_visible_name, node_visible_name)
2680
      with the mapping from node devices to instance devices
2681

2682
  """
2683
  device_info = []
2684
  disks_ok = True
2685
  iname = instance.name
2686
  # With the two passes mechanism we try to reduce the window of
2687
  # opportunity for the race condition of switching DRBD to primary
2688
  # before handshaking occured, but we do not eliminate it
2689

    
2690
  # The proper fix would be to wait (with some limits) until the
2691
  # connection has been made and drbd transitions from WFConnection
2692
  # into any other network-connected state (Connected, SyncTarget,
2693
  # SyncSource, etc.)
2694

    
2695
  # 1st pass, assemble on all nodes in secondary mode
2696
  for inst_disk in instance.disks:
2697
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
2698
      lu.cfg.SetDiskID(node_disk, node)
2699
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, False)
2700
      msg = result.RemoteFailMsg()
2701
      if msg:
2702
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
2703
                           " (is_primary=False, pass=1): %s",
2704
                           inst_disk.iv_name, node, msg)
2705
        if not ignore_secondaries:
2706
          disks_ok = False
2707

    
2708
  # FIXME: race condition on drbd migration to primary
2709

    
2710
  # 2nd pass, do only the primary node
2711
  for inst_disk in instance.disks:
2712
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
2713
      if node != instance.primary_node:
2714
        continue
2715
      lu.cfg.SetDiskID(node_disk, node)
2716
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, True)
2717
      msg = result.RemoteFailMsg()
2718
      if msg:
2719
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
2720
                           " (is_primary=True, pass=2): %s",
2721
                           inst_disk.iv_name, node, msg)
2722
        disks_ok = False
2723
    device_info.append((instance.primary_node, inst_disk.iv_name,
2724
                        result.payload))
2725

    
2726
  # leave the disks configured for the primary node
2727
  # this is a workaround that would be fixed better by
2728
  # improving the logical/physical id handling
2729
  for disk in instance.disks:
2730
    lu.cfg.SetDiskID(disk, instance.primary_node)
2731

    
2732
  return disks_ok, device_info
2733

    
2734

    
2735
def _StartInstanceDisks(lu, instance, force):
2736
  """Start the disks of an instance.
2737

2738
  """
2739
  disks_ok, dummy = _AssembleInstanceDisks(lu, instance,
2740
                                           ignore_secondaries=force)
2741
  if not disks_ok:
2742
    _ShutdownInstanceDisks(lu, instance)
2743
    if force is not None and not force:
2744
      lu.proc.LogWarning("", hint="If the message above refers to a"
2745
                         " secondary node,"
2746
                         " you can retry the operation using '--force'.")
2747
    raise errors.OpExecError("Disk consistency error")
2748

    
2749

    
2750
class LUDeactivateInstanceDisks(NoHooksLU):
2751
  """Shutdown an instance's disks.
2752

2753
  """
2754
  _OP_REQP = ["instance_name"]
2755
  REQ_BGL = False
2756

    
2757
  def ExpandNames(self):
2758
    self._ExpandAndLockInstance()
2759
    self.needed_locks[locking.LEVEL_NODE] = []
2760
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2761

    
2762
  def DeclareLocks(self, level):
2763
    if level == locking.LEVEL_NODE:
2764
      self._LockInstancesNodes()
2765

    
2766
  def CheckPrereq(self):
2767
    """Check prerequisites.
2768

2769
    This checks that the instance is in the cluster.
2770

2771
    """
2772
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2773
    assert self.instance is not None, \
2774
      "Cannot retrieve locked instance %s" % self.op.instance_name
2775

    
2776
  def Exec(self, feedback_fn):
2777
    """Deactivate the disks
2778

2779
    """
2780
    instance = self.instance
2781
    _SafeShutdownInstanceDisks(self, instance)
2782

    
2783

    
2784
def _SafeShutdownInstanceDisks(lu, instance):
2785
  """Shutdown block devices of an instance.
2786

2787
  This function checks if an instance is running, before calling
2788
  _ShutdownInstanceDisks.
2789

2790
  """
2791
  pnode = instance.primary_node
2792
  ins_l = lu.rpc.call_instance_list([pnode], [instance.hypervisor])
2793
  ins_l = ins_l[pnode]
2794
  msg = ins_l.RemoteFailMsg()
2795
  if msg:
2796
    raise errors.OpExecError("Can't contact node %s: %s" % (pnode, msg))
2797

    
2798
  if instance.name in ins_l.payload:
2799
    raise errors.OpExecError("Instance is running, can't shutdown"
2800
                             " block devices.")
2801

    
2802
  _ShutdownInstanceDisks(lu, instance)
2803

    
2804

    
2805
def _ShutdownInstanceDisks(lu, instance, ignore_primary=False):
2806
  """Shutdown block devices of an instance.
2807

2808
  This does the shutdown on all nodes of the instance.
2809

2810
  If the ignore_primary is false, errors on the primary node are
2811
  ignored.
2812

2813
  """
2814
  all_result = True
2815
  for disk in instance.disks:
2816
    for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
2817
      lu.cfg.SetDiskID(top_disk, node)
2818
      result = lu.rpc.call_blockdev_shutdown(node, top_disk)
2819
      msg = result.RemoteFailMsg()
2820
      if msg:
2821
        lu.LogWarning("Could not shutdown block device %s on node %s: %s",
2822
                      disk.iv_name, node, msg)
2823
        if not ignore_primary or node != instance.primary_node:
2824
          all_result = False
2825
  return all_result
2826

    
2827

    
2828
def _CheckNodeFreeMemory(lu, node, reason, requested, hypervisor_name):
2829
  """Checks if a node has enough free memory.
2830

2831
  This function check if a given node has the needed amount of free
2832
  memory. In case the node has less memory or we cannot get the
2833
  information from the node, this function raise an OpPrereqError
2834
  exception.
2835

2836
  @type lu: C{LogicalUnit}
2837
  @param lu: a logical unit from which we get configuration data
2838
  @type node: C{str}
2839
  @param node: the node to check
2840
  @type reason: C{str}
2841
  @param reason: string to use in the error message
2842
  @type requested: C{int}
2843
  @param requested: the amount of memory in MiB to check for
2844
  @type hypervisor_name: C{str}
2845
  @param hypervisor_name: the hypervisor to ask for memory stats
2846
  @raise errors.OpPrereqError: if the node doesn't have enough memory, or
2847
      we cannot check the node
2848

2849
  """
2850
  nodeinfo = lu.rpc.call_node_info([node], lu.cfg.GetVGName(), hypervisor_name)
2851
  msg = nodeinfo[node].RemoteFailMsg()
2852
  if msg:
2853
    raise errors.OpPrereqError("Can't get data from node %s: %s" % (node, msg))
2854
  free_mem = nodeinfo[node].payload.get('memory_free', None)
2855
  if not isinstance(free_mem, int):
2856
    raise errors.OpPrereqError("Can't compute free memory on node %s, result"
2857
                               " was '%s'" % (node, free_mem))
2858
  if requested > free_mem:
2859
    raise errors.OpPrereqError("Not enough memory on node %s for %s:"
2860
                               " needed %s MiB, available %s MiB" %
2861
                               (node, reason, requested, free_mem))
2862

    
2863

    
2864
class LUStartupInstance(LogicalUnit):
2865
  """Starts an instance.
2866

2867
  """
2868
  HPATH = "instance-start"
2869
  HTYPE = constants.HTYPE_INSTANCE
2870
  _OP_REQP = ["instance_name", "force"]
2871
  REQ_BGL = False
2872

    
2873
  def ExpandNames(self):
2874
    self._ExpandAndLockInstance()
2875

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

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

2881
    """
2882
    env = {
2883
      "FORCE": self.op.force,
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
    # extra beparams
2900
    self.beparams = getattr(self.op, "beparams", {})
2901
    if self.beparams:
2902
      if not isinstance(self.beparams, dict):
2903
        raise errors.OpPrereqError("Invalid beparams passed: %s, expected"
2904
                                   " dict" % (type(self.beparams), ))
2905
      # fill the beparams dict
2906
      utils.ForceDictType(self.beparams, constants.BES_PARAMETER_TYPES)
2907
      self.op.beparams = self.beparams
2908

    
2909
    # extra hvparams
2910
    self.hvparams = getattr(self.op, "hvparams", {})
2911
    if self.hvparams:
2912
      if not isinstance(self.hvparams, dict):
2913
        raise errors.OpPrereqError("Invalid hvparams passed: %s, expected"
2914
                                   " dict" % (type(self.hvparams), ))
2915

    
2916
      # check hypervisor parameter syntax (locally)
2917
      cluster = self.cfg.GetClusterInfo()
2918
      utils.ForceDictType(self.hvparams, constants.HVS_PARAMETER_TYPES)
2919
      filled_hvp = objects.FillDict(cluster.hvparams[instance.hypervisor],
2920
                                    instance.hvparams)
2921
      filled_hvp.update(self.hvparams)
2922
      hv_type = hypervisor.GetHypervisor(instance.hypervisor)
2923
      hv_type.CheckParameterSyntax(filled_hvp)
2924
      _CheckHVParams(self, instance.all_nodes, instance.hypervisor, filled_hvp)
2925
      self.op.hvparams = self.hvparams
2926

    
2927
    _CheckNodeOnline(self, instance.primary_node)
2928

    
2929
    bep = self.cfg.GetClusterInfo().FillBE(instance)
2930
    # check bridges existance
2931
    _CheckInstanceBridgesExist(self, instance)
2932

    
2933
    remote_info = self.rpc.call_instance_info(instance.primary_node,
2934
                                              instance.name,
2935
                                              instance.hypervisor)
2936
    msg = remote_info.RemoteFailMsg()
2937
    if msg:
2938
      raise errors.OpPrereqError("Error checking node %s: %s" %
2939
                                 (instance.primary_node, msg))
2940
    if not remote_info.payload: # not running already
2941
      _CheckNodeFreeMemory(self, instance.primary_node,
2942
                           "starting instance %s" % instance.name,
2943
                           bep[constants.BE_MEMORY], instance.hypervisor)
2944

    
2945
  def Exec(self, feedback_fn):
2946
    """Start the instance.
2947

2948
    """
2949
    instance = self.instance
2950
    force = self.op.force
2951

    
2952
    self.cfg.MarkInstanceUp(instance.name)
2953

    
2954
    node_current = instance.primary_node
2955

    
2956
    _StartInstanceDisks(self, instance, force)
2957

    
2958
    result = self.rpc.call_instance_start(node_current, instance,
2959
                                          self.hvparams, self.beparams)
2960
    msg = result.RemoteFailMsg()
2961
    if msg:
2962
      _ShutdownInstanceDisks(self, instance)
2963
      raise errors.OpExecError("Could not start instance: %s" % msg)
2964

    
2965

    
2966
class LURebootInstance(LogicalUnit):
2967
  """Reboot an instance.
2968

2969
  """
2970
  HPATH = "instance-reboot"
2971
  HTYPE = constants.HTYPE_INSTANCE
2972
  _OP_REQP = ["instance_name", "ignore_secondaries", "reboot_type"]
2973
  REQ_BGL = False
2974

    
2975
  def ExpandNames(self):
2976
    if self.op.reboot_type not in [constants.INSTANCE_REBOOT_SOFT,
2977
                                   constants.INSTANCE_REBOOT_HARD,
2978
                                   constants.INSTANCE_REBOOT_FULL]:
2979
      raise errors.ParameterError("reboot type not in [%s, %s, %s]" %
2980
                                  (constants.INSTANCE_REBOOT_SOFT,
2981
                                   constants.INSTANCE_REBOOT_HARD,
2982
                                   constants.INSTANCE_REBOOT_FULL))
2983
    self._ExpandAndLockInstance()
2984

    
2985
  def BuildHooksEnv(self):
2986
    """Build hooks env.
2987

2988
    This runs on master, primary and secondary nodes of the instance.
2989

2990
    """
2991
    env = {
2992
      "IGNORE_SECONDARIES": self.op.ignore_secondaries,
2993
      "REBOOT_TYPE": self.op.reboot_type,
2994
      }
2995
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
2996
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2997
    return env, nl, nl
2998

    
2999
  def CheckPrereq(self):
3000
    """Check prerequisites.
3001

3002
    This checks that the instance is in the cluster.
3003

3004
    """
3005
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3006
    assert self.instance is not None, \
3007
      "Cannot retrieve locked instance %s" % self.op.instance_name
3008

    
3009
    _CheckNodeOnline(self, instance.primary_node)
3010

    
3011
    # check bridges existance
3012
    _CheckInstanceBridgesExist(self, instance)
3013

    
3014
  def Exec(self, feedback_fn):
3015
    """Reboot the instance.
3016

3017
    """
3018
    instance = self.instance
3019
    ignore_secondaries = self.op.ignore_secondaries
3020
    reboot_type = self.op.reboot_type
3021

    
3022
    node_current = instance.primary_node
3023

    
3024
    if reboot_type in [constants.INSTANCE_REBOOT_SOFT,
3025
                       constants.INSTANCE_REBOOT_HARD]:
3026
      for disk in instance.disks:
3027
        self.cfg.SetDiskID(disk, node_current)
3028
      result = self.rpc.call_instance_reboot(node_current, instance,
3029
                                             reboot_type)
3030
      msg = result.RemoteFailMsg()
3031
      if msg:
3032
        raise errors.OpExecError("Could not reboot instance: %s" % msg)
3033
    else:
3034
      result = self.rpc.call_instance_shutdown(node_current, instance)
3035
      msg = result.RemoteFailMsg()
3036
      if msg:
3037
        raise errors.OpExecError("Could not shutdown instance for"
3038
                                 " full reboot: %s" % msg)
3039
      _ShutdownInstanceDisks(self, instance)
3040
      _StartInstanceDisks(self, instance, ignore_secondaries)
3041
      result = self.rpc.call_instance_start(node_current, instance, None, None)
3042
      msg = result.RemoteFailMsg()
3043
      if msg:
3044
        _ShutdownInstanceDisks(self, instance)
3045
        raise errors.OpExecError("Could not start instance for"
3046
                                 " full reboot: %s" % msg)
3047

    
3048
    self.cfg.MarkInstanceUp(instance.name)
3049

    
3050

    
3051
class LUShutdownInstance(LogicalUnit):
3052
  """Shutdown an instance.
3053

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

    
3060
  def ExpandNames(self):
3061
    self._ExpandAndLockInstance()
3062

    
3063
  def BuildHooksEnv(self):
3064
    """Build hooks env.
3065

3066
    This runs on master, primary and secondary nodes of the instance.
3067

3068
    """
3069
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3070
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3071
    return env, nl, nl
3072

    
3073
  def CheckPrereq(self):
3074
    """Check prerequisites.
3075

3076
    This checks that the instance is in the cluster.
3077

3078
    """
3079
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3080
    assert self.instance is not None, \
3081
      "Cannot retrieve locked instance %s" % self.op.instance_name
3082
    _CheckNodeOnline(self, self.instance.primary_node)
3083

    
3084
  def Exec(self, feedback_fn):
3085
    """Shutdown the instance.
3086

3087
    """
3088
    instance = self.instance
3089
    node_current = instance.primary_node
3090
    self.cfg.MarkInstanceDown(instance.name)
3091
    result = self.rpc.call_instance_shutdown(node_current, instance)
3092
    msg = result.RemoteFailMsg()
3093
    if msg:
3094
      self.proc.LogWarning("Could not shutdown instance: %s" % msg)
3095

    
3096
    _ShutdownInstanceDisks(self, instance)
3097

    
3098

    
3099
class LUReinstallInstance(LogicalUnit):
3100
  """Reinstall an instance.
3101

3102
  """
3103
  HPATH = "instance-reinstall"
3104
  HTYPE = constants.HTYPE_INSTANCE
3105
  _OP_REQP = ["instance_name"]
3106
  REQ_BGL = False
3107

    
3108
  def ExpandNames(self):
3109
    self._ExpandAndLockInstance()
3110

    
3111
  def BuildHooksEnv(self):
3112
    """Build hooks env.
3113

3114
    This runs on master, primary and secondary nodes of the instance.
3115

3116
    """
3117
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3118
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3119
    return env, nl, nl
3120

    
3121
  def CheckPrereq(self):
3122
    """Check prerequisites.
3123

3124
    This checks that the instance is in the cluster and is not running.
3125

3126
    """
3127
    instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3128
    assert instance is not None, \
3129
      "Cannot retrieve locked instance %s" % self.op.instance_name
3130
    _CheckNodeOnline(self, instance.primary_node)
3131

    
3132
    if instance.disk_template == constants.DT_DISKLESS:
3133
      raise errors.OpPrereqError("Instance '%s' has no disks" %
3134
                                 self.op.instance_name)
3135
    if instance.admin_up:
3136
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
3137
                                 self.op.instance_name)
3138
    remote_info = self.rpc.call_instance_info(instance.primary_node,
3139
                                              instance.name,
3140
                                              instance.hypervisor)
3141
    msg = remote_info.RemoteFailMsg()
3142
    if msg:
3143
      raise errors.OpPrereqError("Error checking node %s: %s" %
3144
                                 (instance.primary_node, msg))
3145
    if remote_info.payload:
3146
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
3147
                                 (self.op.instance_name,
3148
                                  instance.primary_node))
3149

    
3150
    self.op.os_type = getattr(self.op, "os_type", None)
3151
    if self.op.os_type is not None:
3152
      # OS verification
3153
      pnode = self.cfg.GetNodeInfo(
3154
        self.cfg.ExpandNodeName(instance.primary_node))
3155
      if pnode is None:
3156
        raise errors.OpPrereqError("Primary node '%s' is unknown" %
3157
                                   self.op.pnode)
3158
      result = self.rpc.call_os_get(pnode.name, self.op.os_type)
3159
      result.Raise()
3160
      if not isinstance(result.data, objects.OS):
3161
        raise errors.OpPrereqError("OS '%s' not in supported OS list for"
3162
                                   " primary node"  % self.op.os_type)
3163

    
3164
    self.instance = instance
3165

    
3166
  def Exec(self, feedback_fn):
3167
    """Reinstall the instance.
3168

3169
    """
3170
    inst = self.instance
3171

    
3172
    if self.op.os_type is not None:
3173
      feedback_fn("Changing OS to '%s'..." % self.op.os_type)
3174
      inst.os = self.op.os_type
3175
      self.cfg.Update(inst)
3176

    
3177
    _StartInstanceDisks(self, inst, None)
3178
    try:
3179
      feedback_fn("Running the instance OS create scripts...")
3180
      result = self.rpc.call_instance_os_add(inst.primary_node, inst, True)
3181
      msg = result.RemoteFailMsg()
3182
      if msg:
3183
        raise errors.OpExecError("Could not install OS for instance %s"
3184
                                 " on node %s: %s" %
3185
                                 (inst.name, inst.primary_node, msg))
3186
    finally:
3187
      _ShutdownInstanceDisks(self, inst)
3188

    
3189

    
3190
class LURenameInstance(LogicalUnit):
3191
  """Rename an instance.
3192

3193
  """
3194
  HPATH = "instance-rename"
3195
  HTYPE = constants.HTYPE_INSTANCE
3196
  _OP_REQP = ["instance_name", "new_name"]
3197

    
3198
  def BuildHooksEnv(self):
3199
    """Build hooks env.
3200

3201
    This runs on master, primary and secondary nodes of the instance.
3202

3203
    """
3204
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3205
    env["INSTANCE_NEW_NAME"] = self.op.new_name
3206
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
3207
    return env, nl, nl
3208

    
3209
  def CheckPrereq(self):
3210
    """Check prerequisites.
3211

3212
    This checks that the instance is in the cluster and is not running.
3213

3214
    """
3215
    instance = self.cfg.GetInstanceInfo(
3216
      self.cfg.ExpandInstanceName(self.op.instance_name))
3217
    if instance is None:
3218
      raise errors.OpPrereqError("Instance '%s' not known" %
3219
                                 self.op.instance_name)
3220
    _CheckNodeOnline(self, instance.primary_node)
3221

    
3222
    if instance.admin_up:
3223
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
3224
                                 self.op.instance_name)
3225
    remote_info = self.rpc.call_instance_info(instance.primary_node,
3226
                                              instance.name,
3227
                                              instance.hypervisor)
3228
    msg = remote_info.RemoteFailMsg()
3229
    if msg:
3230
      raise errors.OpPrereqError("Error checking node %s: %s" %
3231
                                 (instance.primary_node, msg))
3232
    if remote_info.payload:
3233
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
3234
                                 (self.op.instance_name,
3235
                                  instance.primary_node))
3236
    self.instance = instance
3237

    
3238
    # new name verification
3239
    name_info = utils.HostInfo(self.op.new_name)
3240

    
3241
    self.op.new_name = new_name = name_info.name
3242
    instance_list = self.cfg.GetInstanceList()
3243
    if new_name in instance_list:
3244
      raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
3245
                                 new_name)
3246

    
3247
    if not getattr(self.op, "ignore_ip", False):
3248
      if utils.TcpPing(name_info.ip, constants.DEFAULT_NODED_PORT):
3249
        raise errors.OpPrereqError("IP %s of instance %s already in use" %
3250
                                   (name_info.ip, new_name))
3251

    
3252

    
3253
  def Exec(self, feedback_fn):
3254
    """Reinstall the instance.
3255

3256
    """
3257
    inst = self.instance
3258
    old_name = inst.name
3259

    
3260
    if inst.disk_template == constants.DT_FILE:
3261
      old_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
3262

    
3263
    self.cfg.RenameInstance(inst.name, self.op.new_name)
3264
    # Change the instance lock. This is definitely safe while we hold the BGL
3265
    self.context.glm.remove(locking.LEVEL_INSTANCE, old_name)
3266
    self.context.glm.add(locking.LEVEL_INSTANCE, self.op.new_name)
3267

    
3268
    # re-read the instance from the configuration after rename
3269
    inst = self.cfg.GetInstanceInfo(self.op.new_name)
3270

    
3271
    if inst.disk_template == constants.DT_FILE:
3272
      new_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
3273
      result = self.rpc.call_file_storage_dir_rename(inst.primary_node,
3274
                                                     old_file_storage_dir,
3275
                                                     new_file_storage_dir)
3276
      msg = result.RemoteFailMsg()
3277
      if msg:
3278
        raise errors.OpExecError("Could not rename on node %s"
3279
                                 " directory '%s' to '%s' (but the instance"
3280
                                 " has been renamed in Ganeti): %s" %
3281
                                 (inst.primary_node, old_file_storage_dir,
3282
                                  new_file_storage_dir, msg))
3283

    
3284
    _StartInstanceDisks(self, inst, None)
3285
    try:
3286
      result = self.rpc.call_instance_run_rename(inst.primary_node, inst,
3287
                                                 old_name)
3288
      msg = result.RemoteFailMsg()
3289
      if msg:
3290
        msg = ("Could not run OS rename script for instance %s on node %s"
3291
               " (but the instance has been renamed in Ganeti): %s" %
3292
               (inst.name, inst.primary_node, msg))
3293
        self.proc.LogWarning(msg)
3294
    finally:
3295
      _ShutdownInstanceDisks(self, inst)
3296

    
3297

    
3298
class LURemoveInstance(LogicalUnit):
3299
  """Remove an instance.
3300

3301
  """
3302
  HPATH = "instance-remove"
3303
  HTYPE = constants.HTYPE_INSTANCE
3304
  _OP_REQP = ["instance_name", "ignore_failures"]
3305
  REQ_BGL = False
3306

    
3307
  def ExpandNames(self):
3308
    self._ExpandAndLockInstance()
3309
    self.needed_locks[locking.LEVEL_NODE] = []
3310
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3311

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

    
3316
  def BuildHooksEnv(self):
3317
    """Build hooks env.
3318

3319
    This runs on master, primary and secondary nodes of the instance.
3320

3321
    """
3322
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3323
    nl = [self.cfg.GetMasterNode()]
3324
    return env, nl, nl
3325

    
3326
  def CheckPrereq(self):
3327
    """Check prerequisites.
3328

3329
    This checks that the instance is in the cluster.
3330

3331
    """
3332
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3333
    assert self.instance is not None, \
3334
      "Cannot retrieve locked instance %s" % self.op.instance_name
3335

    
3336
  def Exec(self, feedback_fn):
3337
    """Remove the instance.
3338

3339
    """
3340
    instance = self.instance
3341
    logging.info("Shutting down instance %s on node %s",
3342
                 instance.name, instance.primary_node)
3343

    
3344
    result = self.rpc.call_instance_shutdown(instance.primary_node, instance)
3345
    msg = result.RemoteFailMsg()
3346
    if msg:
3347
      if self.op.ignore_failures:
3348
        feedback_fn("Warning: can't shutdown instance: %s" % msg)
3349
      else:
3350
        raise errors.OpExecError("Could not shutdown instance %s on"
3351
                                 " node %s: %s" %
3352
                                 (instance.name, instance.primary_node, msg))
3353

    
3354
    logging.info("Removing block devices for instance %s", instance.name)
3355

    
3356
    if not _RemoveDisks(self, instance):
3357
      if self.op.ignore_failures:
3358
        feedback_fn("Warning: can't remove instance's disks")
3359
      else:
3360
        raise errors.OpExecError("Can't remove instance's disks")
3361

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

    
3364
    self.cfg.RemoveInstance(instance.name)
3365
    self.remove_locks[locking.LEVEL_INSTANCE] = instance.name
3366

    
3367

    
3368
class LUQueryInstances(NoHooksLU):
3369
  """Logical unit for querying instances.
3370

3371
  """
3372
  _OP_REQP = ["output_fields", "names", "use_locking"]
3373
  REQ_BGL = False
3374
  _FIELDS_STATIC = utils.FieldSet(*["name", "os", "pnode", "snodes",
3375
                                    "admin_state",
3376
                                    "disk_template", "ip", "mac", "bridge",
3377
                                    "sda_size", "sdb_size", "vcpus", "tags",
3378
                                    "network_port", "beparams",
3379
                                    r"(disk)\.(size)/([0-9]+)",
3380
                                    r"(disk)\.(sizes)", "disk_usage",
3381
                                    r"(nic)\.(mac|ip|bridge)/([0-9]+)",
3382
                                    r"(nic)\.(macs|ips|bridges)",
3383
                                    r"(disk|nic)\.(count)",
3384
                                    "serial_no", "hypervisor", "hvparams",] +
3385
                                  ["hv/%s" % name
3386
                                   for name in constants.HVS_PARAMETERS] +
3387
                                  ["be/%s" % name
3388
                                   for name in constants.BES_PARAMETERS])
3389
  _FIELDS_DYNAMIC = utils.FieldSet("oper_state", "oper_ram", "status")
3390

    
3391

    
3392
  def ExpandNames(self):
3393
    _CheckOutputFields(static=self._FIELDS_STATIC,
3394
                       dynamic=self._FIELDS_DYNAMIC,
3395
                       selected=self.op.output_fields)
3396

    
3397
    self.needed_locks = {}
3398
    self.share_locks[locking.LEVEL_INSTANCE] = 1
3399
    self.share_locks[locking.LEVEL_NODE] = 1
3400

    
3401
    if self.op.names:
3402
      self.wanted = _GetWantedInstances(self, self.op.names)
3403
    else:
3404
      self.wanted = locking.ALL_SET
3405

    
3406
    self.do_node_query = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
3407
    self.do_locking = self.do_node_query and self.op.use_locking
3408
    if self.do_locking:
3409
      self.needed_locks[locking.LEVEL_INSTANCE] = self.wanted
3410
      self.needed_locks[locking.LEVEL_NODE] = []
3411
      self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3412

    
3413
  def DeclareLocks(self, level):
3414
    if level == locking.LEVEL_NODE and self.do_locking:
3415
      self._LockInstancesNodes()
3416

    
3417
  def CheckPrereq(self):
3418
    """Check prerequisites.
3419

3420
    """
3421
    pass
3422

    
3423
  def Exec(self, feedback_fn):
3424
    """Computes the list of nodes and their attributes.
3425

3426
    """
3427
    all_info = self.cfg.GetAllInstancesInfo()
3428
    if self.wanted == locking.ALL_SET:
3429
      # caller didn't specify instance names, so ordering is not important
3430
      if self.do_locking:
3431
        instance_names = self.acquired_locks[locking.LEVEL_INSTANCE]
3432
      else:
3433
        instance_names = all_info.keys()
3434
      instance_names = utils.NiceSort(instance_names)
3435
    else:
3436
      # caller did specify names, so we must keep the ordering
3437
      if self.do_locking:
3438
        tgt_set = self.acquired_locks[locking.LEVEL_INSTANCE]
3439
      else:
3440
        tgt_set = all_info.keys()
3441
      missing = set(self.wanted).difference(tgt_set)
3442
      if missing:
3443
        raise errors.OpExecError("Some instances were removed before"
3444
                                 " retrieving their data: %s" % missing)
3445
      instance_names = self.wanted
3446

    
3447
    instance_list = [all_info[iname] for iname in instance_names]
3448

    
3449
    # begin data gathering
3450

    
3451
    nodes = frozenset([inst.primary_node for inst in instance_list])
3452
    hv_list = list(set([inst.hypervisor for inst in instance_list]))
3453

    
3454
    bad_nodes = []
3455
    off_nodes = []
3456
    if self.do_node_query:
3457
      live_data = {}
3458
      node_data = self.rpc.call_all_instances_info(nodes, hv_list)
3459
      for name in nodes:
3460
        result = node_data[name]
3461
        if result.offline:
3462
          # offline nodes will be in both lists
3463
          off_nodes.append(name)
3464
        if result.failed or result.RemoteFailMsg():
3465
          bad_nodes.append(name)
3466
        else:
3467
          if result.payload:
3468
            live_data.update(result.payload)
3469
          # else no instance is alive
3470
    else:
3471
      live_data = dict([(name, {}) for name in instance_names])
3472

    
3473
    # end data gathering
3474

    
3475
    HVPREFIX = "hv/"
3476
    BEPREFIX = "be/"
3477
    output = []
3478
    for instance in instance_list:
3479
      iout = []
3480
      i_hv = self.cfg.GetClusterInfo().FillHV(instance)
3481
      i_be = self.cfg.GetClusterInfo().FillBE(instance)
3482
      for field in self.op.output_fields:
3483
        st_match = self._FIELDS_STATIC.Matches(field)
3484
        if field == "name":
3485
          val = instance.name
3486
        elif field == "os":
3487
          val = instance.os
3488
        elif field == "pnode":
3489
          val = instance.primary_node
3490
        elif field == "snodes":
3491
          val = list(instance.secondary_nodes)
3492
        elif field == "admin_state":
3493
          val = instance.admin_up
3494
        elif field == "oper_state":
3495
          if instance.primary_node in bad_nodes:
3496
            val = None
3497
          else:
3498
            val = bool(live_data.get(instance.name))
3499
        elif field == "status":
3500
          if instance.primary_node in off_nodes:
3501
            val = "ERROR_nodeoffline"
3502
          elif instance.primary_node in bad_nodes:
3503
            val = "ERROR_nodedown"
3504
          else:
3505
            running = bool(live_data.get(instance.name))
3506
            if running:
3507
              if instance.admin_up:
3508
                val = "running"
3509
              else:
3510
                val = "ERROR_up"
3511
            else:
3512
              if instance.admin_up:
3513
                val = "ERROR_down"
3514
              else:
3515
                val = "ADMIN_down"
3516
        elif field == "oper_ram":
3517
          if instance.primary_node in bad_nodes:
3518
            val = None
3519
          elif instance.name in live_data:
3520
            val = live_data[instance.name].get("memory", "?")
3521
          else:
3522
            val = "-"
3523
        elif field == "disk_template":
3524
          val = instance.disk_template
3525
        elif field == "ip":
3526
          val = instance.nics[0].ip
3527
        elif field == "bridge":
3528
          val = instance.nics[0].bridge
3529
        elif field == "mac":
3530
          val = instance.nics[0].mac
3531
        elif field == "sda_size" or field == "sdb_size":
3532
          idx = ord(field[2]) - ord('a')
3533
          try:
3534
            val = instance.FindDisk(idx).size
3535
          except errors.OpPrereqError:
3536
            val = None
3537
        elif field == "disk_usage": # total disk usage per node
3538
          disk_sizes = [{'size': disk.size} for disk in instance.disks]
3539
          val = _ComputeDiskSize(instance.disk_template, disk_sizes)
3540
        elif field == "tags":
3541
          val = list(instance.GetTags())
3542
        elif field == "serial_no":
3543
          val = instance.serial_no
3544
        elif field == "network_port":
3545
          val = instance.network_port
3546
        elif field == "hypervisor":
3547
          val = instance.hypervisor
3548
        elif field == "hvparams":
3549
          val = i_hv
3550
        elif (field.startswith(HVPREFIX) and
3551
              field[len(HVPREFIX):] in constants.HVS_PARAMETERS):
3552
          val = i_hv.get(field[len(HVPREFIX):], None)
3553
        elif field == "beparams":
3554
          val = i_be
3555
        elif (field.startswith(BEPREFIX) and
3556
              field[len(BEPREFIX):] in constants.BES_PARAMETERS):
3557
          val = i_be.get(field[len(BEPREFIX):], None)
3558
        elif st_match and st_match.groups():
3559
          # matches a variable list
3560
          st_groups = st_match.groups()
3561
          if st_groups and st_groups[0] == "disk":
3562
            if st_groups[1] == "count":
3563
              val = len(instance.disks)
3564
            elif st_groups[1] == "sizes":
3565
              val = [disk.size for disk in instance.disks]
3566
            elif st_groups[1] == "size":
3567
              try:
3568
                val = instance.FindDisk(st_groups[2]).size
3569
              except errors.OpPrereqError:
3570
                val = None
3571
            else:
3572
              assert False, "Unhandled disk parameter"
3573
          elif st_groups[0] == "nic":
3574
            if st_groups[1] == "count":
3575
              val = len(instance.nics)
3576
            elif st_groups[1] == "macs":
3577
              val = [nic.mac for nic in instance.nics]
3578
            elif st_groups[1] == "ips":
3579
              val = [nic.ip for nic in instance.nics]
3580
            elif st_groups[1] == "bridges":
3581
              val = [nic.bridge for nic in instance.nics]
3582
            else:
3583
              # index-based item
3584
              nic_idx = int(st_groups[2])
3585
              if nic_idx >= len(instance.nics):
3586
                val = None
3587
              else:
3588
                if st_groups[1] == "mac":
3589
                  val = instance.nics[nic_idx].mac
3590
                elif st_groups[1] == "ip":
3591
                  val = instance.nics[nic_idx].ip
3592
                elif st_groups[1] == "bridge":
3593
                  val = instance.nics[nic_idx].bridge
3594
                else:
3595
                  assert False, "Unhandled NIC parameter"
3596
          else:
3597
            assert False, "Unhandled variable parameter"
3598
        else:
3599
          raise errors.ParameterError(field)
3600
        iout.append(val)
3601
      output.append(iout)
3602

    
3603
    return output
3604

    
3605

    
3606
class LUFailoverInstance(LogicalUnit):
3607
  """Failover an instance.
3608

3609
  """
3610
  HPATH = "instance-failover"
3611
  HTYPE = constants.HTYPE_INSTANCE
3612
  _OP_REQP = ["instance_name", "ignore_consistency"]
3613
  REQ_BGL = False
3614

    
3615
  def ExpandNames(self):
3616
    self._ExpandAndLockInstance()
3617
    self.needed_locks[locking.LEVEL_NODE] = []
3618
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3619

    
3620
  def DeclareLocks(self, level):
3621
    if level == locking.LEVEL_NODE:
3622
      self._LockInstancesNodes()
3623

    
3624
  def BuildHooksEnv(self):
3625
    """Build hooks env.
3626

3627
    This runs on master, primary and secondary nodes of the instance.
3628

3629
    """
3630
    env = {
3631
      "IGNORE_CONSISTENCY": self.op.ignore_consistency,
3632
      }
3633
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3634
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3635
    return env, nl, nl
3636

    
3637
  def CheckPrereq(self):
3638
    """Check prerequisites.
3639

3640
    This checks that the instance is in the cluster.
3641

3642
    """
3643
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3644
    assert self.instance is not None, \
3645
      "Cannot retrieve locked instance %s" % self.op.instance_name
3646

    
3647
    bep = self.cfg.GetClusterInfo().FillBE(instance)
3648
    if instance.disk_template not in constants.DTS_NET_MIRROR:
3649
      raise errors.OpPrereqError("Instance's disk layout is not"
3650
                                 " network mirrored, cannot failover.")
3651

    
3652
    secondary_nodes = instance.secondary_nodes
3653
    if not secondary_nodes:
3654
      raise errors.ProgrammerError("no secondary node but using "
3655
                                   "a mirrored disk template")
3656

    
3657
    target_node = secondary_nodes[0]
3658
    _CheckNodeOnline(self, target_node)
3659
    _CheckNodeNotDrained(self, target_node)
3660
    # check memory requirements on the secondary node
3661
    _CheckNodeFreeMemory(self, target_node, "failing over instance %s" %
3662
                         instance.name, bep[constants.BE_MEMORY],
3663
                         instance.hypervisor)
3664
    # check bridge existance
3665
    _CheckInstanceBridgesExist(self, instance, node=target_node)
3666

    
3667
  def Exec(self, feedback_fn):
3668
    """Failover an instance.
3669

3670
    The failover is done by shutting it down on its present node and
3671
    starting it on the secondary.
3672

3673
    """
3674
    instance = self.instance
3675

    
3676
    source_node = instance.primary_node
3677
    target_node = instance.secondary_nodes[0]
3678

    
3679
    feedback_fn("* checking disk consistency between source and target")
3680
    for dev in instance.disks:
3681
      # for drbd, these are drbd over lvm
3682
      if not _CheckDiskConsistency(self, dev, target_node, False):
3683
        if instance.admin_up and not self.op.ignore_consistency:
3684
          raise errors.OpExecError("Disk %s is degraded on target node,"
3685
                                   " aborting failover." % dev.iv_name)
3686

    
3687
    feedback_fn("* shutting down instance on source node")
3688
    logging.info("Shutting down instance %s on node %s",
3689
                 instance.name, source_node)
3690

    
3691
    result = self.rpc.call_instance_shutdown(source_node, instance)
3692
    msg = result.RemoteFailMsg()
3693
    if msg:
3694
      if self.op.ignore_consistency:
3695
        self.proc.LogWarning("Could not shutdown instance %s on node %s."
3696
                             " Proceeding anyway. Please make sure node"
3697
                             " %s is down. Error details: %s",
3698
                             instance.name, source_node, source_node, msg)
3699
      else:
3700
        raise errors.OpExecError("Could not shutdown instance %s on"
3701
                                 " node %s: %s" %
3702
                                 (instance.name, source_node, msg))
3703

    
3704
    feedback_fn("* deactivating the instance's disks on source node")
3705
    if not _ShutdownInstanceDisks(self, instance, ignore_primary=True):
3706
      raise errors.OpExecError("Can't shut down the instance's disks.")
3707

    
3708
    instance.primary_node = target_node
3709
    # distribute new instance config to the other nodes
3710
    self.cfg.Update(instance)
3711

    
3712
    # Only start the instance if it's marked as up
3713
    if instance.admin_up:
3714
      feedback_fn("* activating the instance's disks on target node")
3715
      logging.info("Starting instance %s on node %s",
3716
                   instance.name, target_node)
3717

    
3718
      disks_ok, dummy = _AssembleInstanceDisks(self, instance,
3719
                                               ignore_secondaries=True)
3720
      if not disks_ok:
3721
        _ShutdownInstanceDisks(self, instance)
3722
        raise errors.OpExecError("Can't activate the instance's disks")
3723

    
3724
      feedback_fn("* starting the instance on the target node")
3725
      result = self.rpc.call_instance_start(target_node, instance, None, None)
3726
      msg = result.RemoteFailMsg()
3727
      if msg:
3728
        _ShutdownInstanceDisks(self, instance)
3729
        raise errors.OpExecError("Could not start instance %s on node %s: %s" %
3730
                                 (instance.name, target_node, msg))
3731

    
3732

    
3733
class LUMigrateInstance(LogicalUnit):
3734
  """Migrate an instance.
3735

3736
  This is migration without shutting down, compared to the failover,
3737
  which is done with shutdown.
3738

3739
  """
3740
  HPATH = "instance-migrate"
3741
  HTYPE = constants.HTYPE_INSTANCE
3742
  _OP_REQP = ["instance_name", "live", "cleanup"]
3743

    
3744
  REQ_BGL = False
3745

    
3746
  def ExpandNames(self):
3747
    self._ExpandAndLockInstance()
3748
    self.needed_locks[locking.LEVEL_NODE] = []
3749
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3750

    
3751
  def DeclareLocks(self, level):
3752
    if level == locking.LEVEL_NODE:
3753
      self._LockInstancesNodes()
3754

    
3755
  def BuildHooksEnv(self):
3756
    """Build hooks env.
3757

3758
    This runs on master, primary and secondary nodes of the instance.
3759

3760
    """
3761
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3762
    env["MIGRATE_LIVE"] = self.op.live
3763
    env["MIGRATE_CLEANUP"] = self.op.cleanup
3764
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3765
    return env, nl, nl
3766

    
3767
  def CheckPrereq(self):
3768
    """Check prerequisites.
3769

3770
    This checks that the instance is in the cluster.
3771

3772
    """
3773
    instance = self.cfg.GetInstanceInfo(
3774
      self.cfg.ExpandInstanceName(self.op.instance_name))
3775
    if instance is None:
3776
      raise errors.OpPrereqError("Instance '%s' not known" %
3777
                                 self.op.instance_name)
3778

    
3779
    if instance.disk_template != constants.DT_DRBD8:
3780
      raise errors.OpPrereqError("Instance's disk layout is not"
3781
                                 " drbd8, cannot migrate.")
3782

    
3783
    secondary_nodes = instance.secondary_nodes
3784
    if not secondary_nodes:
3785
      raise errors.ConfigurationError("No secondary node but using"
3786
                                      " drbd8 disk template")
3787

    
3788
    i_be = self.cfg.GetClusterInfo().FillBE(instance)
3789

    
3790
    target_node = secondary_nodes[0]
3791
    # check memory requirements on the secondary node
3792
    _CheckNodeFreeMemory(self, target_node, "migrating instance %s" %
3793
                         instance.name, i_be[constants.BE_MEMORY],
3794
                         instance.hypervisor)
3795

    
3796
    # check bridge existance
3797
    _CheckInstanceBridgesExist(self, instance, node=target_node)
3798

    
3799
    if not self.op.cleanup:
3800
      _CheckNodeNotDrained(self, target_node)
3801
      result = self.rpc.call_instance_migratable(instance.primary_node,
3802
                                                 instance)
3803
      msg = result.RemoteFailMsg()
3804
      if msg:
3805
        raise errors.OpPrereqError("Can't migrate: %s - please use failover" %
3806
                                   msg)
3807

    
3808
    self.instance = instance
3809

    
3810
  def _WaitUntilSync(self):
3811
    """Poll with custom rpc for disk sync.
3812

3813
    This uses our own step-based rpc call.
3814

3815
    """
3816
    self.feedback_fn("* wait until resync is done")
3817
    all_done = False
3818
    while not all_done:
3819
      all_done = True
3820
      result = self.rpc.call_drbd_wait_sync(self.all_nodes,
3821
                                            self.nodes_ip,
3822
                                            self.instance.disks)
3823
      min_percent = 100
3824
      for node, nres in result.items():
3825
        msg = nres.RemoteFailMsg()
3826
        if msg:
3827
          raise errors.OpExecError("Cannot resync disks on node %s: %s" %
3828
                                   (node, msg))
3829
        node_done, node_percent = nres.payload
3830
        all_done = all_done and node_done
3831
        if node_percent is not None:
3832
          min_percent = min(min_percent, node_percent)
3833
      if not all_done:
3834
        if min_percent < 100:
3835
          self.feedback_fn("   - progress: %.1f%%" % min_percent)
3836
        time.sleep(2)
3837

    
3838
  def _EnsureSecondary(self, node):
3839
    """Demote a node to secondary.
3840

3841
    """
3842
    self.feedback_fn("* switching node %s to secondary mode" % node)
3843

    
3844
    for dev in self.instance.disks:
3845
      self.cfg.SetDiskID(dev, node)
3846

    
3847
    result = self.rpc.call_blockdev_close(node, self.instance.name,
3848
                                          self.instance.disks)
3849
    msg = result.RemoteFailMsg()
3850
    if msg:
3851
      raise errors.OpExecError("Cannot change disk to secondary on node %s,"
3852
                               " error %s" % (node, msg))
3853

    
3854
  def _GoStandalone(self):
3855
    """Disconnect from the network.
3856

3857
    """
3858
    self.feedback_fn("* changing into standalone mode")
3859
    result = self.rpc.call_drbd_disconnect_net(self.all_nodes, self.nodes_ip,
3860
                                               self.instance.disks)
3861
    for node, nres in result.items():
3862
      msg = nres.RemoteFailMsg()
3863
      if msg:
3864
        raise errors.OpExecError("Cannot disconnect disks node %s,"
3865
                                 " error %s" % (node, msg))
3866

    
3867
  def _GoReconnect(self, multimaster):
3868
    """Reconnect to the network.
3869

3870
    """
3871
    if multimaster:
3872
      msg = "dual-master"
3873
    else:
3874
      msg = "single-master"
3875
    self.feedback_fn("* changing disks into %s mode" % msg)
3876
    result = self.rpc.call_drbd_attach_net(self.all_nodes, self.nodes_ip,
3877
                                           self.instance.disks,
3878
                                           self.instance.name, multimaster)
3879
    for node, nres in result.items():
3880
      msg = nres.RemoteFailMsg()
3881
      if msg:
3882
        raise errors.OpExecError("Cannot change disks config on node %s,"
3883
                                 " error: %s" % (node, msg))
3884

    
3885
  def _ExecCleanup(self):
3886
    """Try to cleanup after a failed migration.
3887

3888
    The cleanup is done by:
3889
      - check that the instance is running only on one node
3890
        (and update the config if needed)
3891
      - change disks on its secondary node to secondary
3892
      - wait until disks are fully synchronized
3893
      - disconnect from the network
3894
      - change disks into single-master mode
3895
      - wait again until disks are fully synchronized
3896

3897
    """
3898
    instance = self.instance
3899
    target_node = self.target_node
3900
    source_node = self.source_node
3901

    
3902
    # check running on only one node
3903
    self.feedback_fn("* checking where the instance actually runs"
3904
                     " (if this hangs, the hypervisor might be in"
3905
                     " a bad state)")
3906
    ins_l = self.rpc.call_instance_list(self.all_nodes, [instance.hypervisor])
3907
    for node, result in ins_l.items():
3908
      msg = result.RemoteFailMsg()
3909
      if msg:
3910
        raise errors.OpExecError("Can't contact node %s: %s" % (node, msg))
3911

    
3912
    runningon_source = instance.name in ins_l[source_node].payload
3913
    runningon_target = instance.name in ins_l[target_node].payload
3914

    
3915
    if runningon_source and runningon_target:
3916
      raise errors.OpExecError("Instance seems to be running on two nodes,"
3917
                               " or the hypervisor is confused. You will have"
3918
                               " to ensure manually that it runs only on one"
3919
                               " and restart this operation.")
3920

    
3921
    if not (runningon_source or runningon_target):
3922
      raise errors.OpExecError("Instance does not seem to be running at all."
3923
                               " In this case, it's safer to repair by"
3924
                               " running 'gnt-instance stop' to ensure disk"
3925
                               " shutdown, and then restarting it.")
3926

    
3927
    if runningon_target:
3928
      # the migration has actually succeeded, we need to update the config
3929
      self.feedback_fn("* instance running on secondary node (%s),"
3930
                       " updating config" % target_node)
3931
      instance.primary_node = target_node
3932
      self.cfg.Update(instance)
3933
      demoted_node = source_node
3934
    else:
3935
      self.feedback_fn("* instance confirmed to be running on its"
3936
                       " primary node (%s)" % source_node)
3937
      demoted_node = target_node
3938

    
3939
    self._EnsureSecondary(demoted_node)
3940
    try:
3941
      self._WaitUntilSync()
3942
    except errors.OpExecError:
3943
      # we ignore here errors, since if the device is standalone, it
3944
      # won't be able to sync
3945
      pass
3946
    self._GoStandalone()
3947
    self._GoReconnect(False)
3948
    self._WaitUntilSync()
3949

    
3950
    self.feedback_fn("* done")
3951

    
3952
  def _RevertDiskStatus(self):
3953
    """Try to revert the disk status after a failed migration.
3954

3955
    """
3956
    target_node = self.target_node
3957
    try:
3958
      self._EnsureSecondary(target_node)
3959
      self._GoStandalone()
3960
      self._GoReconnect(False)
3961
      self._WaitUntilSync()
3962
    except errors.OpExecError, err:
3963
      self.LogWarning("Migration failed and I can't reconnect the"
3964
                      " drives: error '%s'\n"
3965
                      "Please look and recover the instance status" %
3966
                      str(err))
3967

    
3968
  def _AbortMigration(self):
3969
    """Call the hypervisor code to abort a started migration.
3970

3971
    """
3972
    instance = self.instance
3973
    target_node = self.target_node
3974
    migration_info = self.migration_info
3975

    
3976
    abort_result = self.rpc.call_finalize_migration(target_node,
3977
                                                    instance,
3978
                                                    migration_info,
3979
                                                    False)
3980
    abort_msg = abort_result.RemoteFailMsg()
3981
    if abort_msg:
3982
      logging.error("Aborting migration failed on target node %s: %s" %
3983
                    (target_node, abort_msg))
3984
      # Don't raise an exception here, as we stil have to try to revert the
3985
      # disk status, even if this step failed.
3986

    
3987
  def _ExecMigration(self):
3988
    """Migrate an instance.
3989

3990
    The migrate is done by:
3991
      - change the disks into dual-master mode
3992
      - wait until disks are fully synchronized again
3993
      - migrate the instance
3994
      - change disks on the new secondary node (the old primary) to secondary
3995
      - wait until disks are fully synchronized<