Statistics
| Branch: | Tag: | Revision:

root / lib / cmdlib.py @ 07813a9e

History | View | Annotate | Download (239.2 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 sha
29
import time
30
import tempfile
31
import re
32
import platform
33
import logging
34
import copy
35
import random
36

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

    
48

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

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

61
  Note that all commands require root permissions.
62

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

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

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

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

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

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

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

    
109
  ssh = property(fget=__GetSSH)
110

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

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

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

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

126
    """
127
    pass
128

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

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

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

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

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

150
    Examples::
151

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

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

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

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

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

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

189
    """
190

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

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

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

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

205
    """
206
    raise NotImplementedError
207

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

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

215
    """
216
    raise NotImplementedError
217

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

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

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

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

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

237
    """
238
    raise NotImplementedError
239

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

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

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

258
    """
259
    return lu_result
260

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

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

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

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

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

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

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

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

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

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

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

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

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

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

    
326
    del self.recalculate_locks[locking.LEVEL_NODE]
327

    
328

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

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

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

    
339

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

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

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

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

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

    
366
  return utils.NiceSort(wanted)
367

    
368

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

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

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

    
385
  if instances:
386
    wanted = []
387

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

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

    
398

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

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

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

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

    
417

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

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

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

    
431

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

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

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

    
443

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

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

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

    
455

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

460
  This builds the hook environment from individual variables.
461

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

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

    
498
  if nics:
499
    nic_count = len(nics)
500
    for idx, (ip, bridge, mac) in enumerate(nics):
501
      if ip is None:
502
        ip = ""
503
      env["INSTANCE_NIC%d_IP" % idx] = ip
504
      env["INSTANCE_NIC%d_BRIDGE" % idx] = bridge
505
      env["INSTANCE_NIC%d_HWADDR" % idx] = mac
506
  else:
507
    nic_count = 0
508

    
509
  env["INSTANCE_NIC_COUNT"] = nic_count
510

    
511
  return env
512

    
513

    
514
def _BuildInstanceHookEnvByObject(lu, instance, override=None):
515
  """Builds instance related env variables for hooks from an object.
516

517
  @type lu: L{LogicalUnit}
518
  @param lu: the logical unit on whose behalf we execute
519
  @type instance: L{objects.Instance}
520
  @param instance: the instance for which we should build the
521
      environment
522
  @type override: dict
523
  @param override: dictionary with key/values that will override
524
      our values
525
  @rtype: dict
526
  @return: the hook environment dictionary
527

528
  """
529
  bep = lu.cfg.GetClusterInfo().FillBE(instance)
530
  args = {
531
    'name': instance.name,
532
    'primary_node': instance.primary_node,
533
    'secondary_nodes': instance.secondary_nodes,
534
    'os_type': instance.os,
535
    'status': instance.admin_up,
536
    'memory': bep[constants.BE_MEMORY],
537
    'vcpus': bep[constants.BE_VCPUS],
538
    'nics': [(nic.ip, nic.bridge, nic.mac) for nic in instance.nics],
539
  }
540
  if override:
541
    args.update(override)
542
  return _BuildInstanceHookEnv(**args)
543

    
544

    
545
def _AdjustCandidatePool(lu):
546
  """Adjust the candidate pool after node operations.
547

548
  """
549
  mod_list = lu.cfg.MaintainCandidatePool()
550
  if mod_list:
551
    lu.LogInfo("Promoted nodes to master candidate role: %s",
552
               ", ".join(node.name for node in mod_list))
553
    for name in mod_list:
554
      lu.context.ReaddNode(name)
555
  mc_now, mc_max = lu.cfg.GetMasterCandidateStats()
556
  if mc_now > mc_max:
557
    lu.LogInfo("Note: more nodes are candidates (%d) than desired (%d)" %
558
               (mc_now, mc_max))
559

    
560

    
561
def _CheckInstanceBridgesExist(lu, instance):
562
  """Check that the brigdes needed by an instance exist.
563

564
  """
565
  # check bridges existance
566
  brlist = [nic.bridge for nic in instance.nics]
567
  result = lu.rpc.call_bridges_exist(instance.primary_node, brlist)
568
  result.Raise()
569
  if not result.data:
570
    raise errors.OpPrereqError("One or more target bridges %s does not"
571
                               " exist on destination node '%s'" %
572
                               (brlist, instance.primary_node))
573

    
574

    
575
class LUDestroyCluster(NoHooksLU):
576
  """Logical unit for destroying the cluster.
577

578
  """
579
  _OP_REQP = []
580

    
581
  def CheckPrereq(self):
582
    """Check prerequisites.
583

584
    This checks whether the cluster is empty.
585

586
    Any errors are signalled by raising errors.OpPrereqError.
587

588
    """
589
    master = self.cfg.GetMasterNode()
590

    
591
    nodelist = self.cfg.GetNodeList()
592
    if len(nodelist) != 1 or nodelist[0] != master:
593
      raise errors.OpPrereqError("There are still %d node(s) in"
594
                                 " this cluster." % (len(nodelist) - 1))
595
    instancelist = self.cfg.GetInstanceList()
596
    if instancelist:
597
      raise errors.OpPrereqError("There are still %d instance(s) in"
598
                                 " this cluster." % len(instancelist))
599

    
600
  def Exec(self, feedback_fn):
601
    """Destroys the cluster.
602

603
    """
604
    master = self.cfg.GetMasterNode()
605
    result = self.rpc.call_node_stop_master(master, False)
606
    result.Raise()
607
    if not result.data:
608
      raise errors.OpExecError("Could not disable the master role")
609
    priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
610
    utils.CreateBackup(priv_key)
611
    utils.CreateBackup(pub_key)
612
    return master
613

    
614

    
615
class LUVerifyCluster(LogicalUnit):
616
  """Verifies the cluster status.
617

618
  """
619
  HPATH = "cluster-verify"
620
  HTYPE = constants.HTYPE_CLUSTER
621
  _OP_REQP = ["skip_checks"]
622
  REQ_BGL = False
623

    
624
  def ExpandNames(self):
625
    self.needed_locks = {
626
      locking.LEVEL_NODE: locking.ALL_SET,
627
      locking.LEVEL_INSTANCE: locking.ALL_SET,
628
    }
629
    self.share_locks = dict(((i, 1) for i in locking.LEVELS))
630

    
631
  def _VerifyNode(self, nodeinfo, file_list, local_cksum,
632
                  node_result, feedback_fn, master_files,
633
                  drbd_map):
634
    """Run multiple tests against a node.
635

636
    Test list:
637

638
      - compares ganeti version
639
      - checks vg existance and size > 20G
640
      - checks config file checksum
641
      - checks ssh to other nodes
642

643
    @type nodeinfo: L{objects.Node}
644
    @param nodeinfo: the node to check
645
    @param file_list: required list of files
646
    @param local_cksum: dictionary of local files and their checksums
647
    @param node_result: the results from the node
648
    @param feedback_fn: function used to accumulate results
649
    @param master_files: list of files that only masters should have
650
    @param drbd_map: the useddrbd minors for this node, in
651
        form of minor: (instance, must_exist) which correspond to instances
652
        and their running status
653

654
    """
655
    node = nodeinfo.name
656

    
657
    # main result, node_result should be a non-empty dict
658
    if not node_result or not isinstance(node_result, dict):
659
      feedback_fn("  - ERROR: unable to verify node %s." % (node,))
660
      return True
661

    
662
    # compares ganeti version
663
    local_version = constants.PROTOCOL_VERSION
664
    remote_version = node_result.get('version', None)
665
    if not (remote_version and isinstance(remote_version, (list, tuple)) and
666
            len(remote_version) == 2):
667
      feedback_fn("  - ERROR: connection to %s failed" % (node))
668
      return True
669

    
670
    if local_version != remote_version[0]:
671
      feedback_fn("  - ERROR: incompatible protocol versions: master %s,"
672
                  " node %s %s" % (local_version, node, remote_version[0]))
673
      return True
674

    
675
    # node seems compatible, we can actually try to look into its results
676

    
677
    bad = False
678

    
679
    # full package version
680
    if constants.RELEASE_VERSION != remote_version[1]:
681
      feedback_fn("  - WARNING: software version mismatch: master %s,"
682
                  " node %s %s" %
683
                  (constants.RELEASE_VERSION, node, remote_version[1]))
684

    
685
    # checks vg existence and size > 20G
686

    
687
    vglist = node_result.get(constants.NV_VGLIST, None)
688
    if not vglist:
689
      feedback_fn("  - ERROR: unable to check volume groups on node %s." %
690
                      (node,))
691
      bad = True
692
    else:
693
      vgstatus = utils.CheckVolumeGroupSize(vglist, self.cfg.GetVGName(),
694
                                            constants.MIN_VG_SIZE)
695
      if vgstatus:
696
        feedback_fn("  - ERROR: %s on node %s" % (vgstatus, node))
697
        bad = True
698

    
699
    # checks config file checksum
700

    
701
    remote_cksum = node_result.get(constants.NV_FILELIST, None)
702
    if not isinstance(remote_cksum, dict):
703
      bad = True
704
      feedback_fn("  - ERROR: node hasn't returned file checksum data")
705
    else:
706
      for file_name in file_list:
707
        node_is_mc = nodeinfo.master_candidate
708
        must_have_file = file_name not in master_files
709
        if file_name not in remote_cksum:
710
          if node_is_mc or must_have_file:
711
            bad = True
712
            feedback_fn("  - ERROR: file '%s' missing" % file_name)
713
        elif remote_cksum[file_name] != local_cksum[file_name]:
714
          if node_is_mc or must_have_file:
715
            bad = True
716
            feedback_fn("  - ERROR: file '%s' has wrong checksum" % file_name)
717
          else:
718
            # not candidate and this is not a must-have file
719
            bad = True
720
            feedback_fn("  - ERROR: non master-candidate has old/wrong file"
721
                        " '%s'" % file_name)
722
        else:
723
          # all good, except non-master/non-must have combination
724
          if not node_is_mc and not must_have_file:
725
            feedback_fn("  - ERROR: file '%s' should not exist on non master"
726
                        " candidates" % file_name)
727

    
728
    # checks ssh to any
729

    
730
    if constants.NV_NODELIST not in node_result:
731
      bad = True
732
      feedback_fn("  - ERROR: node hasn't returned node ssh connectivity data")
733
    else:
734
      if node_result[constants.NV_NODELIST]:
735
        bad = True
736
        for node in node_result[constants.NV_NODELIST]:
737
          feedback_fn("  - ERROR: ssh communication with node '%s': %s" %
738
                          (node, node_result[constants.NV_NODELIST][node]))
739

    
740
    if constants.NV_NODENETTEST not in node_result:
741
      bad = True
742
      feedback_fn("  - ERROR: node hasn't returned node tcp connectivity data")
743
    else:
744
      if node_result[constants.NV_NODENETTEST]:
745
        bad = True
746
        nlist = utils.NiceSort(node_result[constants.NV_NODENETTEST].keys())
747
        for node in nlist:
748
          feedback_fn("  - ERROR: tcp communication with node '%s': %s" %
749
                          (node, node_result[constants.NV_NODENETTEST][node]))
750

    
751
    hyp_result = node_result.get(constants.NV_HYPERVISOR, None)
752
    if isinstance(hyp_result, dict):
753
      for hv_name, hv_result in hyp_result.iteritems():
754
        if hv_result is not None:
755
          feedback_fn("  - ERROR: hypervisor %s verify failure: '%s'" %
756
                      (hv_name, hv_result))
757

    
758
    # check used drbd list
759
    used_minors = node_result.get(constants.NV_DRBDLIST, [])
760
    if not isinstance(used_minors, (tuple, list)):
761
      feedback_fn("  - ERROR: cannot parse drbd status file: %s" %
762
                  str(used_minors))
763
    else:
764
      for minor, (iname, must_exist) in drbd_map.items():
765
        if minor not in used_minors and must_exist:
766
          feedback_fn("  - ERROR: drbd minor %d of instance %s is not active" %
767
                      (minor, iname))
768
          bad = True
769
      for minor in used_minors:
770
        if minor not in drbd_map:
771
          feedback_fn("  - ERROR: unallocated drbd minor %d is in use" % minor)
772
          bad = True
773

    
774
    return bad
775

    
776
  def _VerifyInstance(self, instance, instanceconfig, node_vol_is,
777
                      node_instance, feedback_fn, n_offline):
778
    """Verify an instance.
779

780
    This function checks to see if the required block devices are
781
    available on the instance's node.
782

783
    """
784
    bad = False
785

    
786
    node_current = instanceconfig.primary_node
787

    
788
    node_vol_should = {}
789
    instanceconfig.MapLVsByNode(node_vol_should)
790

    
791
    for node in node_vol_should:
792
      if node in n_offline:
793
        # ignore missing volumes on offline nodes
794
        continue
795
      for volume in node_vol_should[node]:
796
        if node not in node_vol_is or volume not in node_vol_is[node]:
797
          feedback_fn("  - ERROR: volume %s missing on node %s" %
798
                          (volume, node))
799
          bad = True
800

    
801
    if instanceconfig.admin_up:
802
      if ((node_current not in node_instance or
803
          not instance in node_instance[node_current]) and
804
          node_current not in n_offline):
805
        feedback_fn("  - ERROR: instance %s not running on node %s" %
806
                        (instance, node_current))
807
        bad = True
808

    
809
    for node in node_instance:
810
      if (not node == node_current):
811
        if instance in node_instance[node]:
812
          feedback_fn("  - ERROR: instance %s should not run on node %s" %
813
                          (instance, node))
814
          bad = True
815

    
816
    return bad
817

    
818
  def _VerifyOrphanVolumes(self, node_vol_should, node_vol_is, feedback_fn):
819
    """Verify if there are any unknown volumes in the cluster.
820

821
    The .os, .swap and backup volumes are ignored. All other volumes are
822
    reported as unknown.
823

824
    """
825
    bad = False
826

    
827
    for node in node_vol_is:
828
      for volume in node_vol_is[node]:
829
        if node not in node_vol_should or volume not in node_vol_should[node]:
830
          feedback_fn("  - ERROR: volume %s on node %s should not exist" %
831
                      (volume, node))
832
          bad = True
833
    return bad
834

    
835
  def _VerifyOrphanInstances(self, instancelist, node_instance, feedback_fn):
836
    """Verify the list of running instances.
837

838
    This checks what instances are running but unknown to the cluster.
839

840
    """
841
    bad = False
842
    for node in node_instance:
843
      for runninginstance in node_instance[node]:
844
        if runninginstance not in instancelist:
845
          feedback_fn("  - ERROR: instance %s on node %s should not exist" %
846
                          (runninginstance, node))
847
          bad = True
848
    return bad
849

    
850
  def _VerifyNPlusOneMemory(self, node_info, instance_cfg, feedback_fn):
851
    """Verify N+1 Memory Resilience.
852

853
    Check that if one single node dies we can still start all the instances it
854
    was primary for.
855

856
    """
857
    bad = False
858

    
859
    for node, nodeinfo in node_info.iteritems():
860
      # This code checks that every node which is now listed as secondary has
861
      # enough memory to host all instances it is supposed to should a single
862
      # other node in the cluster fail.
863
      # FIXME: not ready for failover to an arbitrary node
864
      # FIXME: does not support file-backed instances
865
      # WARNING: we currently take into account down instances as well as up
866
      # ones, considering that even if they're down someone might want to start
867
      # them even in the event of a node failure.
868
      for prinode, instances in nodeinfo['sinst-by-pnode'].iteritems():
869
        needed_mem = 0
870
        for instance in instances:
871
          bep = self.cfg.GetClusterInfo().FillBE(instance_cfg[instance])
872
          if bep[constants.BE_AUTO_BALANCE]:
873
            needed_mem += bep[constants.BE_MEMORY]
874
        if nodeinfo['mfree'] < needed_mem:
875
          feedback_fn("  - ERROR: not enough memory on node %s to accomodate"
876
                      " failovers should node %s fail" % (node, prinode))
877
          bad = True
878
    return bad
879

    
880
  def CheckPrereq(self):
881
    """Check prerequisites.
882

883
    Transform the list of checks we're going to skip into a set and check that
884
    all its members are valid.
885

886
    """
887
    self.skip_set = frozenset(self.op.skip_checks)
888
    if not constants.VERIFY_OPTIONAL_CHECKS.issuperset(self.skip_set):
889
      raise errors.OpPrereqError("Invalid checks to be skipped specified")
890

    
891
  def BuildHooksEnv(self):
892
    """Build hooks env.
893

894
    Cluster-Verify hooks just rone in the post phase and their failure makes
895
    the output be logged in the verify output and the verification to fail.
896

897
    """
898
    all_nodes = self.cfg.GetNodeList()
899
    # TODO: populate the environment with useful information for verify hooks
900
    env = {}
901
    return env, [], all_nodes
902

    
903
  def Exec(self, feedback_fn):
904
    """Verify integrity of cluster, performing various test on nodes.
905

906
    """
907
    bad = False
908
    feedback_fn("* Verifying global settings")
909
    for msg in self.cfg.VerifyConfig():
910
      feedback_fn("  - ERROR: %s" % msg)
911

    
912
    vg_name = self.cfg.GetVGName()
913
    hypervisors = self.cfg.GetClusterInfo().enabled_hypervisors
914
    nodelist = utils.NiceSort(self.cfg.GetNodeList())
915
    nodeinfo = [self.cfg.GetNodeInfo(nname) for nname in nodelist]
916
    instancelist = utils.NiceSort(self.cfg.GetInstanceList())
917
    instanceinfo = dict((iname, self.cfg.GetInstanceInfo(iname))
918
                        for iname in instancelist)
919
    i_non_redundant = [] # Non redundant instances
920
    i_non_a_balanced = [] # Non auto-balanced instances
921
    n_offline = [] # List of offline nodes
922
    n_drained = [] # List of nodes being drained
923
    node_volume = {}
924
    node_instance = {}
925
    node_info = {}
926
    instance_cfg = {}
927

    
928
    # FIXME: verify OS list
929
    # do local checksums
930
    master_files = [constants.CLUSTER_CONF_FILE]
931

    
932
    file_names = ssconf.SimpleStore().GetFileList()
933
    file_names.append(constants.SSL_CERT_FILE)
934
    file_names.append(constants.RAPI_CERT_FILE)
935
    file_names.extend(master_files)
936

    
937
    local_checksums = utils.FingerprintFiles(file_names)
938

    
939
    feedback_fn("* Gathering data (%d nodes)" % len(nodelist))
940
    node_verify_param = {
941
      constants.NV_FILELIST: file_names,
942
      constants.NV_NODELIST: [node.name for node in nodeinfo
943
                              if not node.offline],
944
      constants.NV_HYPERVISOR: hypervisors,
945
      constants.NV_NODENETTEST: [(node.name, node.primary_ip,
946
                                  node.secondary_ip) for node in nodeinfo
947
                                 if not node.offline],
948
      constants.NV_LVLIST: vg_name,
949
      constants.NV_INSTANCELIST: hypervisors,
950
      constants.NV_VGLIST: None,
951
      constants.NV_VERSION: None,
952
      constants.NV_HVINFO: self.cfg.GetHypervisorType(),
953
      constants.NV_DRBDLIST: None,
954
      }
955
    all_nvinfo = self.rpc.call_node_verify(nodelist, node_verify_param,
956
                                           self.cfg.GetClusterName())
957

    
958
    cluster = self.cfg.GetClusterInfo()
959
    master_node = self.cfg.GetMasterNode()
960
    all_drbd_map = self.cfg.ComputeDRBDMap()
961

    
962
    for node_i in nodeinfo:
963
      node = node_i.name
964
      nresult = all_nvinfo[node].data
965

    
966
      if node_i.offline:
967
        feedback_fn("* Skipping offline node %s" % (node,))
968
        n_offline.append(node)
969
        continue
970

    
971
      if node == master_node:
972
        ntype = "master"
973
      elif node_i.master_candidate:
974
        ntype = "master candidate"
975
      elif node_i.drained:
976
        ntype = "drained"
977
        n_drained.append(node)
978
      else:
979
        ntype = "regular"
980
      feedback_fn("* Verifying node %s (%s)" % (node, ntype))
981

    
982
      if all_nvinfo[node].failed or not isinstance(nresult, dict):
983
        feedback_fn("  - ERROR: connection to %s failed" % (node,))
984
        bad = True
985
        continue
986

    
987
      node_drbd = {}
988
      for minor, instance in all_drbd_map[node].items():
989
        instance = instanceinfo[instance]
990
        node_drbd[minor] = (instance.name, instance.admin_up)
991
      result = self._VerifyNode(node_i, file_names, local_checksums,
992
                                nresult, feedback_fn, master_files,
993
                                node_drbd)
994
      bad = bad or result
995

    
996
      lvdata = nresult.get(constants.NV_LVLIST, "Missing LV data")
997
      if isinstance(lvdata, basestring):
998
        feedback_fn("  - ERROR: LVM problem on node %s: %s" %
999
                    (node, utils.SafeEncode(lvdata)))
1000
        bad = True
1001
        node_volume[node] = {}
1002
      elif not isinstance(lvdata, dict):
1003
        feedback_fn("  - ERROR: connection to %s failed (lvlist)" % (node,))
1004
        bad = True
1005
        continue
1006
      else:
1007
        node_volume[node] = lvdata
1008

    
1009
      # node_instance
1010
      idata = nresult.get(constants.NV_INSTANCELIST, None)
1011
      if not isinstance(idata, list):
1012
        feedback_fn("  - ERROR: connection to %s failed (instancelist)" %
1013
                    (node,))
1014
        bad = True
1015
        continue
1016

    
1017
      node_instance[node] = idata
1018

    
1019
      # node_info
1020
      nodeinfo = nresult.get(constants.NV_HVINFO, None)
1021
      if not isinstance(nodeinfo, dict):
1022
        feedback_fn("  - ERROR: connection to %s failed (hvinfo)" % (node,))
1023
        bad = True
1024
        continue
1025

    
1026
      try:
1027
        node_info[node] = {
1028
          "mfree": int(nodeinfo['memory_free']),
1029
          "dfree": int(nresult[constants.NV_VGLIST][vg_name]),
1030
          "pinst": [],
1031
          "sinst": [],
1032
          # dictionary holding all instances this node is secondary for,
1033
          # grouped by their primary node. Each key is a cluster node, and each
1034
          # value is a list of instances which have the key as primary and the
1035
          # current node as secondary.  this is handy to calculate N+1 memory
1036
          # availability if you can only failover from a primary to its
1037
          # secondary.
1038
          "sinst-by-pnode": {},
1039
        }
1040
      except ValueError:
1041
        feedback_fn("  - ERROR: invalid value returned from node %s" % (node,))
1042
        bad = True
1043
        continue
1044

    
1045
    node_vol_should = {}
1046

    
1047
    for instance in instancelist:
1048
      feedback_fn("* Verifying instance %s" % instance)
1049
      inst_config = instanceinfo[instance]
1050
      result =  self._VerifyInstance(instance, inst_config, node_volume,
1051
                                     node_instance, feedback_fn, n_offline)
1052
      bad = bad or result
1053
      inst_nodes_offline = []
1054

    
1055
      inst_config.MapLVsByNode(node_vol_should)
1056

    
1057
      instance_cfg[instance] = inst_config
1058

    
1059
      pnode = inst_config.primary_node
1060
      if pnode in node_info:
1061
        node_info[pnode]['pinst'].append(instance)
1062
      elif pnode not in n_offline:
1063
        feedback_fn("  - ERROR: instance %s, connection to primary node"
1064
                    " %s failed" % (instance, pnode))
1065
        bad = True
1066

    
1067
      if pnode in n_offline:
1068
        inst_nodes_offline.append(pnode)
1069

    
1070
      # If the instance is non-redundant we cannot survive losing its primary
1071
      # node, so we are not N+1 compliant. On the other hand we have no disk
1072
      # templates with more than one secondary so that situation is not well
1073
      # supported either.
1074
      # FIXME: does not support file-backed instances
1075
      if len(inst_config.secondary_nodes) == 0:
1076
        i_non_redundant.append(instance)
1077
      elif len(inst_config.secondary_nodes) > 1:
1078
        feedback_fn("  - WARNING: multiple secondaries for instance %s"
1079
                    % instance)
1080

    
1081
      if not cluster.FillBE(inst_config)[constants.BE_AUTO_BALANCE]:
1082
        i_non_a_balanced.append(instance)
1083

    
1084
      for snode in inst_config.secondary_nodes:
1085
        if snode in node_info:
1086
          node_info[snode]['sinst'].append(instance)
1087
          if pnode not in node_info[snode]['sinst-by-pnode']:
1088
            node_info[snode]['sinst-by-pnode'][pnode] = []
1089
          node_info[snode]['sinst-by-pnode'][pnode].append(instance)
1090
        elif snode not in n_offline:
1091
          feedback_fn("  - ERROR: instance %s, connection to secondary node"
1092
                      " %s failed" % (instance, snode))
1093
          bad = True
1094
        if snode in n_offline:
1095
          inst_nodes_offline.append(snode)
1096

    
1097
      if inst_nodes_offline:
1098
        # warn that the instance lives on offline nodes, and set bad=True
1099
        feedback_fn("  - ERROR: instance lives on offline node(s) %s" %
1100
                    ", ".join(inst_nodes_offline))
1101
        bad = True
1102

    
1103
    feedback_fn("* Verifying orphan volumes")
1104
    result = self._VerifyOrphanVolumes(node_vol_should, node_volume,
1105
                                       feedback_fn)
1106
    bad = bad or result
1107

    
1108
    feedback_fn("* Verifying remaining instances")
1109
    result = self._VerifyOrphanInstances(instancelist, node_instance,
1110
                                         feedback_fn)
1111
    bad = bad or result
1112

    
1113
    if constants.VERIFY_NPLUSONE_MEM not in self.skip_set:
1114
      feedback_fn("* Verifying N+1 Memory redundancy")
1115
      result = self._VerifyNPlusOneMemory(node_info, instance_cfg, feedback_fn)
1116
      bad = bad or result
1117

    
1118
    feedback_fn("* Other Notes")
1119
    if i_non_redundant:
1120
      feedback_fn("  - NOTICE: %d non-redundant instance(s) found."
1121
                  % len(i_non_redundant))
1122

    
1123
    if i_non_a_balanced:
1124
      feedback_fn("  - NOTICE: %d non-auto-balanced instance(s) found."
1125
                  % len(i_non_a_balanced))
1126

    
1127
    if n_offline:
1128
      feedback_fn("  - NOTICE: %d offline node(s) found." % len(n_offline))
1129

    
1130
    if n_drained:
1131
      feedback_fn("  - NOTICE: %d drained node(s) found." % len(n_drained))
1132

    
1133
    return not bad
1134

    
1135
  def HooksCallBack(self, phase, hooks_results, feedback_fn, lu_result):
1136
    """Analize the post-hooks' result
1137

1138
    This method analyses the hook result, handles it, and sends some
1139
    nicely-formatted feedback back to the user.
1140

1141
    @param phase: one of L{constants.HOOKS_PHASE_POST} or
1142
        L{constants.HOOKS_PHASE_PRE}; it denotes the hooks phase
1143
    @param hooks_results: the results of the multi-node hooks rpc call
1144
    @param feedback_fn: function used send feedback back to the caller
1145
    @param lu_result: previous Exec result
1146
    @return: the new Exec result, based on the previous result
1147
        and hook results
1148

1149
    """
1150
    # We only really run POST phase hooks, and are only interested in
1151
    # their results
1152
    if phase == constants.HOOKS_PHASE_POST:
1153
      # Used to change hooks' output to proper indentation
1154
      indent_re = re.compile('^', re.M)
1155
      feedback_fn("* Hooks Results")
1156
      if not hooks_results:
1157
        feedback_fn("  - ERROR: general communication failure")
1158
        lu_result = 1
1159
      else:
1160
        for node_name in hooks_results:
1161
          show_node_header = True
1162
          res = hooks_results[node_name]
1163
          if res.failed or res.data is False or not isinstance(res.data, list):
1164
            if res.offline:
1165
              # no need to warn or set fail return value
1166
              continue
1167
            feedback_fn("    Communication failure in hooks execution")
1168
            lu_result = 1
1169
            continue
1170
          for script, hkr, output in res.data:
1171
            if hkr == constants.HKR_FAIL:
1172
              # The node header is only shown once, if there are
1173
              # failing hooks on that node
1174
              if show_node_header:
1175
                feedback_fn("  Node %s:" % node_name)
1176
                show_node_header = False
1177
              feedback_fn("    ERROR: Script %s failed, output:" % script)
1178
              output = indent_re.sub('      ', output)
1179
              feedback_fn("%s" % output)
1180
              lu_result = 1
1181

    
1182
      return lu_result
1183

    
1184

    
1185
class LUVerifyDisks(NoHooksLU):
1186
  """Verifies the cluster disks status.
1187

1188
  """
1189
  _OP_REQP = []
1190
  REQ_BGL = False
1191

    
1192
  def ExpandNames(self):
1193
    self.needed_locks = {
1194
      locking.LEVEL_NODE: locking.ALL_SET,
1195
      locking.LEVEL_INSTANCE: locking.ALL_SET,
1196
    }
1197
    self.share_locks = dict(((i, 1) for i in locking.LEVELS))
1198

    
1199
  def CheckPrereq(self):
1200
    """Check prerequisites.
1201

1202
    This has no prerequisites.
1203

1204
    """
1205
    pass
1206

    
1207
  def Exec(self, feedback_fn):
1208
    """Verify integrity of cluster disks.
1209

1210
    """
1211
    result = res_nodes, res_nlvm, res_instances, res_missing = [], {}, [], {}
1212

    
1213
    vg_name = self.cfg.GetVGName()
1214
    nodes = utils.NiceSort(self.cfg.GetNodeList())
1215
    instances = [self.cfg.GetInstanceInfo(name)
1216
                 for name in self.cfg.GetInstanceList()]
1217

    
1218
    nv_dict = {}
1219
    for inst in instances:
1220
      inst_lvs = {}
1221
      if (not inst.admin_up or
1222
          inst.disk_template not in constants.DTS_NET_MIRROR):
1223
        continue
1224
      inst.MapLVsByNode(inst_lvs)
1225
      # transform { iname: {node: [vol,],},} to {(node, vol): iname}
1226
      for node, vol_list in inst_lvs.iteritems():
1227
        for vol in vol_list:
1228
          nv_dict[(node, vol)] = inst
1229

    
1230
    if not nv_dict:
1231
      return result
1232

    
1233
    node_lvs = self.rpc.call_volume_list(nodes, vg_name)
1234

    
1235
    to_act = set()
1236
    for node in nodes:
1237
      # node_volume
1238
      lvs = node_lvs[node]
1239
      if lvs.failed:
1240
        if not lvs.offline:
1241
          self.LogWarning("Connection to node %s failed: %s" %
1242
                          (node, lvs.data))
1243
        continue
1244
      lvs = lvs.data
1245
      if isinstance(lvs, basestring):
1246
        logging.warning("Error enumerating LVs on node %s: %s", node, lvs)
1247
        res_nlvm[node] = lvs
1248
      elif not isinstance(lvs, dict):
1249
        logging.warning("Connection to node %s failed or invalid data"
1250
                        " returned", node)
1251
        res_nodes.append(node)
1252
        continue
1253

    
1254
      for lv_name, (_, lv_inactive, lv_online) in lvs.iteritems():
1255
        inst = nv_dict.pop((node, lv_name), None)
1256
        if (not lv_online and inst is not None
1257
            and inst.name not in res_instances):
1258
          res_instances.append(inst.name)
1259

    
1260
    # any leftover items in nv_dict are missing LVs, let's arrange the
1261
    # data better
1262
    for key, inst in nv_dict.iteritems():
1263
      if inst.name not in res_missing:
1264
        res_missing[inst.name] = []
1265
      res_missing[inst.name].append(key)
1266

    
1267
    return result
1268

    
1269

    
1270
class LURenameCluster(LogicalUnit):
1271
  """Rename the cluster.
1272

1273
  """
1274
  HPATH = "cluster-rename"
1275
  HTYPE = constants.HTYPE_CLUSTER
1276
  _OP_REQP = ["name"]
1277

    
1278
  def BuildHooksEnv(self):
1279
    """Build hooks env.
1280

1281
    """
1282
    env = {
1283
      "OP_TARGET": self.cfg.GetClusterName(),
1284
      "NEW_NAME": self.op.name,
1285
      }
1286
    mn = self.cfg.GetMasterNode()
1287
    return env, [mn], [mn]
1288

    
1289
  def CheckPrereq(self):
1290
    """Verify that the passed name is a valid one.
1291

1292
    """
1293
    hostname = utils.HostInfo(self.op.name)
1294

    
1295
    new_name = hostname.name
1296
    self.ip = new_ip = hostname.ip
1297
    old_name = self.cfg.GetClusterName()
1298
    old_ip = self.cfg.GetMasterIP()
1299
    if new_name == old_name and new_ip == old_ip:
1300
      raise errors.OpPrereqError("Neither the name nor the IP address of the"
1301
                                 " cluster has changed")
1302
    if new_ip != old_ip:
1303
      if utils.TcpPing(new_ip, constants.DEFAULT_NODED_PORT):
1304
        raise errors.OpPrereqError("The given cluster IP address (%s) is"
1305
                                   " reachable on the network. Aborting." %
1306
                                   new_ip)
1307

    
1308
    self.op.name = new_name
1309

    
1310
  def Exec(self, feedback_fn):
1311
    """Rename the cluster.
1312

1313
    """
1314
    clustername = self.op.name
1315
    ip = self.ip
1316

    
1317
    # shutdown the master IP
1318
    master = self.cfg.GetMasterNode()
1319
    result = self.rpc.call_node_stop_master(master, False)
1320
    if result.failed or not result.data:
1321
      raise errors.OpExecError("Could not disable the master role")
1322

    
1323
    try:
1324
      cluster = self.cfg.GetClusterInfo()
1325
      cluster.cluster_name = clustername
1326
      cluster.master_ip = ip
1327
      self.cfg.Update(cluster)
1328

    
1329
      # update the known hosts file
1330
      ssh.WriteKnownHostsFile(self.cfg, constants.SSH_KNOWN_HOSTS_FILE)
1331
      node_list = self.cfg.GetNodeList()
1332
      try:
1333
        node_list.remove(master)
1334
      except ValueError:
1335
        pass
1336
      result = self.rpc.call_upload_file(node_list,
1337
                                         constants.SSH_KNOWN_HOSTS_FILE)
1338
      for to_node, to_result in result.iteritems():
1339
        if to_result.failed or not to_result.data:
1340
          logging.error("Copy of file %s to node %s failed",
1341
                        constants.SSH_KNOWN_HOSTS_FILE, to_node)
1342

    
1343
    finally:
1344
      result = self.rpc.call_node_start_master(master, False)
1345
      if result.failed or not result.data:
1346
        self.LogWarning("Could not re-enable the master role on"
1347
                        " the master, please restart manually.")
1348

    
1349

    
1350
def _RecursiveCheckIfLVMBased(disk):
1351
  """Check if the given disk or its children are lvm-based.
1352

1353
  @type disk: L{objects.Disk}
1354
  @param disk: the disk to check
1355
  @rtype: booleean
1356
  @return: boolean indicating whether a LD_LV dev_type was found or not
1357

1358
  """
1359
  if disk.children:
1360
    for chdisk in disk.children:
1361
      if _RecursiveCheckIfLVMBased(chdisk):
1362
        return True
1363
  return disk.dev_type == constants.LD_LV
1364

    
1365

    
1366
class LUSetClusterParams(LogicalUnit):
1367
  """Change the parameters of the cluster.
1368

1369
  """
1370
  HPATH = "cluster-modify"
1371
  HTYPE = constants.HTYPE_CLUSTER
1372
  _OP_REQP = []
1373
  REQ_BGL = False
1374

    
1375
  def CheckParameters(self):
1376
    """Check parameters
1377

1378
    """
1379
    if not hasattr(self.op, "candidate_pool_size"):
1380
      self.op.candidate_pool_size = None
1381
    if self.op.candidate_pool_size is not None:
1382
      try:
1383
        self.op.candidate_pool_size = int(self.op.candidate_pool_size)
1384
      except ValueError, err:
1385
        raise errors.OpPrereqError("Invalid candidate_pool_size value: %s" %
1386
                                   str(err))
1387
      if self.op.candidate_pool_size < 1:
1388
        raise errors.OpPrereqError("At least one master candidate needed")
1389

    
1390
  def ExpandNames(self):
1391
    # FIXME: in the future maybe other cluster params won't require checking on
1392
    # all nodes to be modified.
1393
    self.needed_locks = {
1394
      locking.LEVEL_NODE: locking.ALL_SET,
1395
    }
1396
    self.share_locks[locking.LEVEL_NODE] = 1
1397

    
1398
  def BuildHooksEnv(self):
1399
    """Build hooks env.
1400

1401
    """
1402
    env = {
1403
      "OP_TARGET": self.cfg.GetClusterName(),
1404
      "NEW_VG_NAME": self.op.vg_name,
1405
      }
1406
    mn = self.cfg.GetMasterNode()
1407
    return env, [mn], [mn]
1408

    
1409
  def CheckPrereq(self):
1410
    """Check prerequisites.
1411

1412
    This checks whether the given params don't conflict and
1413
    if the given volume group is valid.
1414

1415
    """
1416
    if self.op.vg_name is not None and not self.op.vg_name:
1417
      instances = self.cfg.GetAllInstancesInfo().values()
1418
      for inst in instances:
1419
        for disk in inst.disks:
1420
          if _RecursiveCheckIfLVMBased(disk):
1421
            raise errors.OpPrereqError("Cannot disable lvm storage while"
1422
                                       " lvm-based instances exist")
1423

    
1424
    node_list = self.acquired_locks[locking.LEVEL_NODE]
1425

    
1426
    # if vg_name not None, checks given volume group on all nodes
1427
    if self.op.vg_name:
1428
      vglist = self.rpc.call_vg_list(node_list)
1429
      for node in node_list:
1430
        if vglist[node].failed:
1431
          # ignoring down node
1432
          self.LogWarning("Node %s unreachable/error, ignoring" % node)
1433
          continue
1434
        vgstatus = utils.CheckVolumeGroupSize(vglist[node].data,
1435
                                              self.op.vg_name,
1436
                                              constants.MIN_VG_SIZE)
1437
        if vgstatus:
1438
          raise errors.OpPrereqError("Error on node '%s': %s" %
1439
                                     (node, vgstatus))
1440

    
1441
    self.cluster = cluster = self.cfg.GetClusterInfo()
1442
    # validate beparams changes
1443
    if self.op.beparams:
1444
      utils.ForceDictType(self.op.beparams, constants.BES_PARAMETER_TYPES)
1445
      self.new_beparams = cluster.FillDict(
1446
        cluster.beparams[constants.BEGR_DEFAULT], self.op.beparams)
1447

    
1448
    # hypervisor list/parameters
1449
    self.new_hvparams = cluster.FillDict(cluster.hvparams, {})
1450
    if self.op.hvparams:
1451
      if not isinstance(self.op.hvparams, dict):
1452
        raise errors.OpPrereqError("Invalid 'hvparams' parameter on input")
1453
      for hv_name, hv_dict in self.op.hvparams.items():
1454
        if hv_name not in self.new_hvparams:
1455
          self.new_hvparams[hv_name] = hv_dict
1456
        else:
1457
          self.new_hvparams[hv_name].update(hv_dict)
1458

    
1459
    if self.op.enabled_hypervisors is not None:
1460
      self.hv_list = self.op.enabled_hypervisors
1461
    else:
1462
      self.hv_list = cluster.enabled_hypervisors
1463

    
1464
    if self.op.hvparams or self.op.enabled_hypervisors is not None:
1465
      # either the enabled list has changed, or the parameters have, validate
1466
      for hv_name, hv_params in self.new_hvparams.items():
1467
        if ((self.op.hvparams and hv_name in self.op.hvparams) or
1468
            (self.op.enabled_hypervisors and
1469
             hv_name in self.op.enabled_hypervisors)):
1470
          # either this is a new hypervisor, or its parameters have changed
1471
          hv_class = hypervisor.GetHypervisor(hv_name)
1472
          utils.ForceDictType(hv_params, constants.HVS_PARAMETER_TYPES)
1473
          hv_class.CheckParameterSyntax(hv_params)
1474
          _CheckHVParams(self, node_list, hv_name, hv_params)
1475

    
1476
  def Exec(self, feedback_fn):
1477
    """Change the parameters of the cluster.
1478

1479
    """
1480
    if self.op.vg_name is not None:
1481
      if self.op.vg_name != self.cfg.GetVGName():
1482
        self.cfg.SetVGName(self.op.vg_name)
1483
      else:
1484
        feedback_fn("Cluster LVM configuration already in desired"
1485
                    " state, not changing")
1486
    if self.op.hvparams:
1487
      self.cluster.hvparams = self.new_hvparams
1488
    if self.op.enabled_hypervisors is not None:
1489
      self.cluster.enabled_hypervisors = self.op.enabled_hypervisors
1490
    if self.op.beparams:
1491
      self.cluster.beparams[constants.BEGR_DEFAULT] = self.new_beparams
1492
    if self.op.candidate_pool_size is not None:
1493
      self.cluster.candidate_pool_size = self.op.candidate_pool_size
1494

    
1495
    self.cfg.Update(self.cluster)
1496

    
1497
    # we want to update nodes after the cluster so that if any errors
1498
    # happen, we have recorded and saved the cluster info
1499
    if self.op.candidate_pool_size is not None:
1500
      _AdjustCandidatePool(self)
1501

    
1502

    
1503
class LURedistributeConfig(NoHooksLU):
1504
  """Force the redistribution of cluster configuration.
1505

1506
  This is a very simple LU.
1507

1508
  """
1509
  _OP_REQP = []
1510
  REQ_BGL = False
1511

    
1512
  def ExpandNames(self):
1513
    self.needed_locks = {
1514
      locking.LEVEL_NODE: locking.ALL_SET,
1515
    }
1516
    self.share_locks[locking.LEVEL_NODE] = 1
1517

    
1518
  def CheckPrereq(self):
1519
    """Check prerequisites.
1520

1521
    """
1522

    
1523
  def Exec(self, feedback_fn):
1524
    """Redistribute the configuration.
1525

1526
    """
1527
    self.cfg.Update(self.cfg.GetClusterInfo())
1528

    
1529

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

1533
  """
1534
  if not instance.disks:
1535
    return True
1536

    
1537
  if not oneshot:
1538
    lu.proc.LogInfo("Waiting for instance %s to sync disks." % instance.name)
1539

    
1540
  node = instance.primary_node
1541

    
1542
  for dev in instance.disks:
1543
    lu.cfg.SetDiskID(dev, node)
1544

    
1545
  retries = 0
1546
  while True:
1547
    max_time = 0
1548
    done = True
1549
    cumul_degraded = False
1550
    rstats = lu.rpc.call_blockdev_getmirrorstatus(node, instance.disks)
1551
    if rstats.failed or not rstats.data:
1552
      lu.LogWarning("Can't get any data from node %s", node)
1553
      retries += 1
1554
      if retries >= 10:
1555
        raise errors.RemoteError("Can't contact node %s for mirror data,"
1556
                                 " aborting." % node)
1557
      time.sleep(6)
1558
      continue
1559
    rstats = rstats.data
1560
    retries = 0
1561
    for i, mstat in enumerate(rstats):
1562
      if mstat is None:
1563
        lu.LogWarning("Can't compute data for node %s/%s",
1564
                           node, instance.disks[i].iv_name)
1565
        continue
1566
      # we ignore the ldisk parameter
1567
      perc_done, est_time, is_degraded, _ = mstat
1568
      cumul_degraded = cumul_degraded or (is_degraded and perc_done is None)
1569
      if perc_done is not None:
1570
        done = False
1571
        if est_time is not None:
1572
          rem_time = "%d estimated seconds remaining" % est_time
1573
          max_time = est_time
1574
        else:
1575
          rem_time = "no time estimate"
1576
        lu.proc.LogInfo("- device %s: %5.2f%% done, %s" %
1577
                        (instance.disks[i].iv_name, perc_done, rem_time))
1578
    if done or oneshot:
1579
      break
1580

    
1581
    time.sleep(min(60, max_time))
1582

    
1583
  if done:
1584
    lu.proc.LogInfo("Instance %s's disks are in sync." % instance.name)
1585
  return not cumul_degraded
1586

    
1587

    
1588
def _CheckDiskConsistency(lu, dev, node, on_primary, ldisk=False):
1589
  """Check that mirrors are not degraded.
1590

1591
  The ldisk parameter, if True, will change the test from the
1592
  is_degraded attribute (which represents overall non-ok status for
1593
  the device(s)) to the ldisk (representing the local storage status).
1594

1595
  """
1596
  lu.cfg.SetDiskID(dev, node)
1597
  if ldisk:
1598
    idx = 6
1599
  else:
1600
    idx = 5
1601

    
1602
  result = True
1603
  if on_primary or dev.AssembleOnSecondary():
1604
    rstats = lu.rpc.call_blockdev_find(node, dev)
1605
    msg = rstats.RemoteFailMsg()
1606
    if msg:
1607
      lu.LogWarning("Can't find disk on node %s: %s", node, msg)
1608
      result = False
1609
    elif not rstats.payload:
1610
      lu.LogWarning("Can't find disk on node %s", node)
1611
      result = False
1612
    else:
1613
      result = result and (not rstats.payload[idx])
1614
  if dev.children:
1615
    for child in dev.children:
1616
      result = result and _CheckDiskConsistency(lu, child, node, on_primary)
1617

    
1618
  return result
1619

    
1620

    
1621
class LUDiagnoseOS(NoHooksLU):
1622
  """Logical unit for OS diagnose/query.
1623

1624
  """
1625
  _OP_REQP = ["output_fields", "names"]
1626
  REQ_BGL = False
1627
  _FIELDS_STATIC = utils.FieldSet()
1628
  _FIELDS_DYNAMIC = utils.FieldSet("name", "valid", "node_status")
1629

    
1630
  def ExpandNames(self):
1631
    if self.op.names:
1632
      raise errors.OpPrereqError("Selective OS query not supported")
1633

    
1634
    _CheckOutputFields(static=self._FIELDS_STATIC,
1635
                       dynamic=self._FIELDS_DYNAMIC,
1636
                       selected=self.op.output_fields)
1637

    
1638
    # Lock all nodes, in shared mode
1639
    self.needed_locks = {}
1640
    self.share_locks[locking.LEVEL_NODE] = 1
1641
    self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
1642

    
1643
  def CheckPrereq(self):
1644
    """Check prerequisites.
1645

1646
    """
1647

    
1648
  @staticmethod
1649
  def _DiagnoseByOS(node_list, rlist):
1650
    """Remaps a per-node return list into an a per-os per-node dictionary
1651

1652
    @param node_list: a list with the names of all nodes
1653
    @param rlist: a map with node names as keys and OS objects as values
1654

1655
    @rtype: dict
1656
    @returns: a dictionary with osnames as keys and as value another map, with
1657
        nodes as keys and list of OS objects as values, eg::
1658

1659
          {"debian-etch": {"node1": [<object>,...],
1660
                           "node2": [<object>,]}
1661
          }
1662

1663
    """
1664
    all_os = {}
1665
    for node_name, nr in rlist.iteritems():
1666
      if nr.failed or not nr.data:
1667
        continue
1668
      for os_obj in nr.data:
1669
        if os_obj.name not in all_os:
1670
          # build a list of nodes for this os containing empty lists
1671
          # for each node in node_list
1672
          all_os[os_obj.name] = {}
1673
          for nname in node_list:
1674
            all_os[os_obj.name][nname] = []
1675
        all_os[os_obj.name][node_name].append(os_obj)
1676
    return all_os
1677

    
1678
  def Exec(self, feedback_fn):
1679
    """Compute the list of OSes.
1680

1681
    """
1682
    node_list = self.acquired_locks[locking.LEVEL_NODE]
1683
    valid_nodes = [node for node in self.cfg.GetOnlineNodeList()
1684
                   if node in node_list]
1685
    node_data = self.rpc.call_os_diagnose(valid_nodes)
1686
    if node_data == False:
1687
      raise errors.OpExecError("Can't gather the list of OSes")
1688
    pol = self._DiagnoseByOS(valid_nodes, node_data)
1689
    output = []
1690
    for os_name, os_data in pol.iteritems():
1691
      row = []
1692
      for field in self.op.output_fields:
1693
        if field == "name":
1694
          val = os_name
1695
        elif field == "valid":
1696
          val = utils.all([osl and osl[0] for osl in os_data.values()])
1697
        elif field == "node_status":
1698
          val = {}
1699
          for node_name, nos_list in os_data.iteritems():
1700
            val[node_name] = [(v.status, v.path) for v in nos_list]
1701
        else:
1702
          raise errors.ParameterError(field)
1703
        row.append(val)
1704
      output.append(row)
1705

    
1706
    return output
1707

    
1708

    
1709
class LURemoveNode(LogicalUnit):
1710
  """Logical unit for removing a node.
1711

1712
  """
1713
  HPATH = "node-remove"
1714
  HTYPE = constants.HTYPE_NODE
1715
  _OP_REQP = ["node_name"]
1716

    
1717
  def BuildHooksEnv(self):
1718
    """Build hooks env.
1719

1720
    This doesn't run on the target node in the pre phase as a failed
1721
    node would then be impossible to remove.
1722

1723
    """
1724
    env = {
1725
      "OP_TARGET": self.op.node_name,
1726
      "NODE_NAME": self.op.node_name,
1727
      }
1728
    all_nodes = self.cfg.GetNodeList()
1729
    all_nodes.remove(self.op.node_name)
1730
    return env, all_nodes, all_nodes
1731

    
1732
  def CheckPrereq(self):
1733
    """Check prerequisites.
1734

1735
    This checks:
1736
     - the node exists in the configuration
1737
     - it does not have primary or secondary instances
1738
     - it's not the master
1739

1740
    Any errors are signalled by raising errors.OpPrereqError.
1741

1742
    """
1743
    node = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.node_name))
1744
    if node is None:
1745
      raise errors.OpPrereqError, ("Node '%s' is unknown." % self.op.node_name)
1746

    
1747
    instance_list = self.cfg.GetInstanceList()
1748

    
1749
    masternode = self.cfg.GetMasterNode()
1750
    if node.name == masternode:
1751
      raise errors.OpPrereqError("Node is the master node,"
1752
                                 " you need to failover first.")
1753

    
1754
    for instance_name in instance_list:
1755
      instance = self.cfg.GetInstanceInfo(instance_name)
1756
      if node.name in instance.all_nodes:
1757
        raise errors.OpPrereqError("Instance %s is still running on the node,"
1758
                                   " please remove first." % instance_name)
1759
    self.op.node_name = node.name
1760
    self.node = node
1761

    
1762
  def Exec(self, feedback_fn):
1763
    """Removes the node from the cluster.
1764

1765
    """
1766
    node = self.node
1767
    logging.info("Stopping the node daemon and removing configs from node %s",
1768
                 node.name)
1769

    
1770
    self.context.RemoveNode(node.name)
1771

    
1772
    self.rpc.call_node_leave_cluster(node.name)
1773

    
1774
    # Promote nodes to master candidate as needed
1775
    _AdjustCandidatePool(self)
1776

    
1777

    
1778
class LUQueryNodes(NoHooksLU):
1779
  """Logical unit for querying nodes.
1780

1781
  """
1782
  _OP_REQP = ["output_fields", "names", "use_locking"]
1783
  REQ_BGL = False
1784
  _FIELDS_DYNAMIC = utils.FieldSet(
1785
    "dtotal", "dfree",
1786
    "mtotal", "mnode", "mfree",
1787
    "bootid",
1788
    "ctotal", "cnodes", "csockets",
1789
    )
1790

    
1791
  _FIELDS_STATIC = utils.FieldSet(
1792
    "name", "pinst_cnt", "sinst_cnt",
1793
    "pinst_list", "sinst_list",
1794
    "pip", "sip", "tags",
1795
    "serial_no",
1796
    "master_candidate",
1797
    "master",
1798
    "offline",
1799
    "drained",
1800
    )
1801

    
1802
  def ExpandNames(self):
1803
    _CheckOutputFields(static=self._FIELDS_STATIC,
1804
                       dynamic=self._FIELDS_DYNAMIC,
1805
                       selected=self.op.output_fields)
1806

    
1807
    self.needed_locks = {}
1808
    self.share_locks[locking.LEVEL_NODE] = 1
1809

    
1810
    if self.op.names:
1811
      self.wanted = _GetWantedNodes(self, self.op.names)
1812
    else:
1813
      self.wanted = locking.ALL_SET
1814

    
1815
    self.do_node_query = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
1816
    self.do_locking = self.do_node_query and self.op.use_locking
1817
    if self.do_locking:
1818
      # if we don't request only static fields, we need to lock the nodes
1819
      self.needed_locks[locking.LEVEL_NODE] = self.wanted
1820

    
1821

    
1822
  def CheckPrereq(self):
1823
    """Check prerequisites.
1824

1825
    """
1826
    # The validation of the node list is done in the _GetWantedNodes,
1827
    # if non empty, and if empty, there's no validation to do
1828
    pass
1829

    
1830
  def Exec(self, feedback_fn):
1831
    """Computes the list of nodes and their attributes.
1832

1833
    """
1834
    all_info = self.cfg.GetAllNodesInfo()
1835
    if self.do_locking:
1836
      nodenames = self.acquired_locks[locking.LEVEL_NODE]
1837
    elif self.wanted != locking.ALL_SET:
1838
      nodenames = self.wanted
1839
      missing = set(nodenames).difference(all_info.keys())
1840
      if missing:
1841
        raise errors.OpExecError(
1842
          "Some nodes were removed before retrieving their data: %s" % missing)
1843
    else:
1844
      nodenames = all_info.keys()
1845

    
1846
    nodenames = utils.NiceSort(nodenames)
1847
    nodelist = [all_info[name] for name in nodenames]
1848

    
1849
    # begin data gathering
1850

    
1851
    if self.do_node_query:
1852
      live_data = {}
1853
      node_data = self.rpc.call_node_info(nodenames, self.cfg.GetVGName(),
1854
                                          self.cfg.GetHypervisorType())
1855
      for name in nodenames:
1856
        nodeinfo = node_data[name]
1857
        if not nodeinfo.failed and nodeinfo.data:
1858
          nodeinfo = nodeinfo.data
1859
          fn = utils.TryConvert
1860
          live_data[name] = {
1861
            "mtotal": fn(int, nodeinfo.get('memory_total', None)),
1862
            "mnode": fn(int, nodeinfo.get('memory_dom0', None)),
1863
            "mfree": fn(int, nodeinfo.get('memory_free', None)),
1864
            "dtotal": fn(int, nodeinfo.get('vg_size', None)),
1865
            "dfree": fn(int, nodeinfo.get('vg_free', None)),
1866
            "ctotal": fn(int, nodeinfo.get('cpu_total', None)),
1867
            "bootid": nodeinfo.get('bootid', None),
1868
            "cnodes": fn(int, nodeinfo.get('cpu_nodes', None)),
1869
            "csockets": fn(int, nodeinfo.get('cpu_sockets', None)),
1870
            }
1871
        else:
1872
          live_data[name] = {}
1873
    else:
1874
      live_data = dict.fromkeys(nodenames, {})
1875

    
1876
    node_to_primary = dict([(name, set()) for name in nodenames])
1877
    node_to_secondary = dict([(name, set()) for name in nodenames])
1878

    
1879
    inst_fields = frozenset(("pinst_cnt", "pinst_list",
1880
                             "sinst_cnt", "sinst_list"))
1881
    if inst_fields & frozenset(self.op.output_fields):
1882
      instancelist = self.cfg.GetInstanceList()
1883

    
1884
      for instance_name in instancelist:
1885
        inst = self.cfg.GetInstanceInfo(instance_name)
1886
        if inst.primary_node in node_to_primary:
1887
          node_to_primary[inst.primary_node].add(inst.name)
1888
        for secnode in inst.secondary_nodes:
1889
          if secnode in node_to_secondary:
1890
            node_to_secondary[secnode].add(inst.name)
1891

    
1892
    master_node = self.cfg.GetMasterNode()
1893

    
1894
    # end data gathering
1895

    
1896
    output = []
1897
    for node in nodelist:
1898
      node_output = []
1899
      for field in self.op.output_fields:
1900
        if field == "name":
1901
          val = node.name
1902
        elif field == "pinst_list":
1903
          val = list(node_to_primary[node.name])
1904
        elif field == "sinst_list":
1905
          val = list(node_to_secondary[node.name])
1906
        elif field == "pinst_cnt":
1907
          val = len(node_to_primary[node.name])
1908
        elif field == "sinst_cnt":
1909
          val = len(node_to_secondary[node.name])
1910
        elif field == "pip":
1911
          val = node.primary_ip
1912
        elif field == "sip":
1913
          val = node.secondary_ip
1914
        elif field == "tags":
1915
          val = list(node.GetTags())
1916
        elif field == "serial_no":
1917
          val = node.serial_no
1918
        elif field == "master_candidate":
1919
          val = node.master_candidate
1920
        elif field == "master":
1921
          val = node.name == master_node
1922
        elif field == "offline":
1923
          val = node.offline
1924
        elif field == "drained":
1925
          val = node.drained
1926
        elif self._FIELDS_DYNAMIC.Matches(field):
1927
          val = live_data[node.name].get(field, None)
1928
        else:
1929
          raise errors.ParameterError(field)
1930
        node_output.append(val)
1931
      output.append(node_output)
1932

    
1933
    return output
1934

    
1935

    
1936
class LUQueryNodeVolumes(NoHooksLU):
1937
  """Logical unit for getting volumes on node(s).
1938

1939
  """
1940
  _OP_REQP = ["nodes", "output_fields"]
1941
  REQ_BGL = False
1942
  _FIELDS_DYNAMIC = utils.FieldSet("phys", "vg", "name", "size", "instance")
1943
  _FIELDS_STATIC = utils.FieldSet("node")
1944

    
1945
  def ExpandNames(self):
1946
    _CheckOutputFields(static=self._FIELDS_STATIC,
1947
                       dynamic=self._FIELDS_DYNAMIC,
1948
                       selected=self.op.output_fields)
1949

    
1950
    self.needed_locks = {}
1951
    self.share_locks[locking.LEVEL_NODE] = 1
1952
    if not self.op.nodes:
1953
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
1954
    else:
1955
      self.needed_locks[locking.LEVEL_NODE] = \
1956
        _GetWantedNodes(self, self.op.nodes)
1957

    
1958
  def CheckPrereq(self):
1959
    """Check prerequisites.
1960

1961
    This checks that the fields required are valid output fields.
1962

1963
    """
1964
    self.nodes = self.acquired_locks[locking.LEVEL_NODE]
1965

    
1966
  def Exec(self, feedback_fn):
1967
    """Computes the list of nodes and their attributes.
1968

1969
    """
1970
    nodenames = self.nodes
1971
    volumes = self.rpc.call_node_volumes(nodenames)
1972

    
1973
    ilist = [self.cfg.GetInstanceInfo(iname) for iname
1974
             in self.cfg.GetInstanceList()]
1975

    
1976
    lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
1977

    
1978
    output = []
1979
    for node in nodenames:
1980
      if node not in volumes or volumes[node].failed or not volumes[node].data:
1981
        continue
1982

    
1983
      node_vols = volumes[node].data[:]
1984
      node_vols.sort(key=lambda vol: vol['dev'])
1985

    
1986
      for vol in node_vols:
1987
        node_output = []
1988
        for field in self.op.output_fields:
1989
          if field == "node":
1990
            val = node
1991
          elif field == "phys":
1992
            val = vol['dev']
1993
          elif field == "vg":
1994
            val = vol['vg']
1995
          elif field == "name":
1996
            val = vol['name']
1997
          elif field == "size":
1998
            val = int(float(vol['size']))
1999
          elif field == "instance":
2000
            for inst in ilist:
2001
              if node not in lv_by_node[inst]:
2002
                continue
2003
              if vol['name'] in lv_by_node[inst][node]:
2004
                val = inst.name
2005
                break
2006
            else:
2007
              val = '-'
2008
          else:
2009
            raise errors.ParameterError(field)
2010
          node_output.append(str(val))
2011

    
2012
        output.append(node_output)
2013

    
2014
    return output
2015

    
2016

    
2017
class LUAddNode(LogicalUnit):
2018
  """Logical unit for adding node to the cluster.
2019

2020
  """
2021
  HPATH = "node-add"
2022
  HTYPE = constants.HTYPE_NODE
2023
  _OP_REQP = ["node_name"]
2024

    
2025
  def BuildHooksEnv(self):
2026
    """Build hooks env.
2027

2028
    This will run on all nodes before, and on all nodes + the new node after.
2029

2030
    """
2031
    env = {
2032
      "OP_TARGET": self.op.node_name,
2033
      "NODE_NAME": self.op.node_name,
2034
      "NODE_PIP": self.op.primary_ip,
2035
      "NODE_SIP": self.op.secondary_ip,
2036
      }
2037
    nodes_0 = self.cfg.GetNodeList()
2038
    nodes_1 = nodes_0 + [self.op.node_name, ]
2039
    return env, nodes_0, nodes_1
2040

    
2041
  def CheckPrereq(self):
2042
    """Check prerequisites.
2043

2044
    This checks:
2045
     - the new node is not already in the config
2046
     - it is resolvable
2047
     - its parameters (single/dual homed) matches the cluster
2048

2049
    Any errors are signalled by raising errors.OpPrereqError.
2050

2051
    """
2052
    node_name = self.op.node_name
2053
    cfg = self.cfg
2054

    
2055
    dns_data = utils.HostInfo(node_name)
2056

    
2057
    node = dns_data.name
2058
    primary_ip = self.op.primary_ip = dns_data.ip
2059
    secondary_ip = getattr(self.op, "secondary_ip", None)
2060
    if secondary_ip is None:
2061
      secondary_ip = primary_ip
2062
    if not utils.IsValidIP(secondary_ip):
2063
      raise errors.OpPrereqError("Invalid secondary IP given")
2064
    self.op.secondary_ip = secondary_ip
2065

    
2066
    node_list = cfg.GetNodeList()
2067
    if not self.op.readd and node in node_list:
2068
      raise errors.OpPrereqError("Node %s is already in the configuration" %
2069
                                 node)
2070
    elif self.op.readd and node not in node_list:
2071
      raise errors.OpPrereqError("Node %s is not in the configuration" % node)
2072

    
2073
    for existing_node_name in node_list:
2074
      existing_node = cfg.GetNodeInfo(existing_node_name)
2075

    
2076
      if self.op.readd and node == existing_node_name:
2077
        if (existing_node.primary_ip != primary_ip or
2078
            existing_node.secondary_ip != secondary_ip):
2079
          raise errors.OpPrereqError("Readded node doesn't have the same IP"
2080
                                     " address configuration as before")
2081
        continue
2082

    
2083
      if (existing_node.primary_ip == primary_ip or
2084
          existing_node.secondary_ip == primary_ip or
2085
          existing_node.primary_ip == secondary_ip or
2086
          existing_node.secondary_ip == secondary_ip):
2087
        raise errors.OpPrereqError("New node ip address(es) conflict with"
2088
                                   " existing node %s" % existing_node.name)
2089

    
2090
    # check that the type of the node (single versus dual homed) is the
2091
    # same as for the master
2092
    myself = cfg.GetNodeInfo(self.cfg.GetMasterNode())
2093
    master_singlehomed = myself.secondary_ip == myself.primary_ip
2094
    newbie_singlehomed = secondary_ip == primary_ip
2095
    if master_singlehomed != newbie_singlehomed:
2096
      if master_singlehomed:
2097
        raise errors.OpPrereqError("The master has no private ip but the"
2098
                                   " new node has one")
2099
      else:
2100
        raise errors.OpPrereqError("The master has a private ip but the"
2101
                                   " new node doesn't have one")
2102

    
2103
    # checks reachablity
2104
    if not utils.TcpPing(primary_ip, constants.DEFAULT_NODED_PORT):
2105
      raise errors.OpPrereqError("Node not reachable by ping")
2106

    
2107
    if not newbie_singlehomed:
2108
      # check reachability from my secondary ip to newbie's secondary ip
2109
      if not utils.TcpPing(secondary_ip, constants.DEFAULT_NODED_PORT,
2110
                           source=myself.secondary_ip):
2111
        raise errors.OpPrereqError("Node secondary ip not reachable by TCP"
2112
                                   " based ping to noded port")
2113

    
2114
    cp_size = self.cfg.GetClusterInfo().candidate_pool_size
2115
    mc_now, _ = self.cfg.GetMasterCandidateStats()
2116
    master_candidate = mc_now < cp_size
2117

    
2118
    self.new_node = objects.Node(name=node,
2119
                                 primary_ip=primary_ip,
2120
                                 secondary_ip=secondary_ip,
2121
                                 master_candidate=master_candidate,
2122
                                 offline=False, drained=False)
2123

    
2124
  def Exec(self, feedback_fn):
2125
    """Adds the new node to the cluster.
2126

2127
    """
2128
    new_node = self.new_node
2129
    node = new_node.name
2130

    
2131
    # check connectivity
2132
    result = self.rpc.call_version([node])[node]
2133
    result.Raise()
2134
    if result.data:
2135
      if constants.PROTOCOL_VERSION == result.data:
2136
        logging.info("Communication to node %s fine, sw version %s match",
2137
                     node, result.data)
2138
      else:
2139
        raise errors.OpExecError("Version mismatch master version %s,"
2140
                                 " node version %s" %
2141
                                 (constants.PROTOCOL_VERSION, result.data))
2142
    else:
2143
      raise errors.OpExecError("Cannot get version from the new node")
2144

    
2145
    # setup ssh on node
2146
    logging.info("Copy ssh key to node %s", node)
2147
    priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
2148
    keyarray = []
2149
    keyfiles = [constants.SSH_HOST_DSA_PRIV, constants.SSH_HOST_DSA_PUB,
2150
                constants.SSH_HOST_RSA_PRIV, constants.SSH_HOST_RSA_PUB,
2151
                priv_key, pub_key]
2152

    
2153
    for i in keyfiles:
2154
      f = open(i, 'r')
2155
      try:
2156
        keyarray.append(f.read())
2157
      finally:
2158
        f.close()
2159

    
2160
    result = self.rpc.call_node_add(node, keyarray[0], keyarray[1],
2161
                                    keyarray[2],
2162
                                    keyarray[3], keyarray[4], keyarray[5])
2163

    
2164
    msg = result.RemoteFailMsg()
2165
    if msg:
2166
      raise errors.OpExecError("Cannot transfer ssh keys to the"
2167
                               " new node: %s" % msg)
2168

    
2169
    # Add node to our /etc/hosts, and add key to known_hosts
2170
    utils.AddHostToEtcHosts(new_node.name)
2171

    
2172
    if new_node.secondary_ip != new_node.primary_ip:
2173
      result = self.rpc.call_node_has_ip_address(new_node.name,
2174
                                                 new_node.secondary_ip)
2175
      if result.failed or not result.data:
2176
        raise errors.OpExecError("Node claims it doesn't have the secondary ip"
2177
                                 " you gave (%s). Please fix and re-run this"
2178
                                 " command." % new_node.secondary_ip)
2179

    
2180
    node_verify_list = [self.cfg.GetMasterNode()]
2181
    node_verify_param = {
2182
      'nodelist': [node],
2183
      # TODO: do a node-net-test as well?
2184
    }
2185

    
2186
    result = self.rpc.call_node_verify(node_verify_list, node_verify_param,
2187
                                       self.cfg.GetClusterName())
2188
    for verifier in node_verify_list:
2189
      if result[verifier].failed or not result[verifier].data:
2190
        raise errors.OpExecError("Cannot communicate with %s's node daemon"
2191
                                 " for remote verification" % verifier)
2192
      if result[verifier].data['nodelist']:
2193
        for failed in result[verifier].data['nodelist']:
2194
          feedback_fn("ssh/hostname verification failed %s -> %s" %
2195
                      (verifier, result[verifier].data['nodelist'][failed]))
2196
        raise errors.OpExecError("ssh/hostname verification failed.")
2197

    
2198
    # Distribute updated /etc/hosts and known_hosts to all nodes,
2199
    # including the node just added
2200
    myself = self.cfg.GetNodeInfo(self.cfg.GetMasterNode())
2201
    dist_nodes = self.cfg.GetNodeList()
2202
    if not self.op.readd:
2203
      dist_nodes.append(node)
2204
    if myself.name in dist_nodes:
2205
      dist_nodes.remove(myself.name)
2206

    
2207
    logging.debug("Copying hosts and known_hosts to all nodes")
2208
    for fname in (constants.ETC_HOSTS, constants.SSH_KNOWN_HOSTS_FILE):
2209
      result = self.rpc.call_upload_file(dist_nodes, fname)
2210
      for to_node, to_result in result.iteritems():
2211
        if to_result.failed or not to_result.data:
2212
          logging.error("Copy of file %s to node %s failed", fname, to_node)
2213

    
2214
    to_copy = []
2215
    enabled_hypervisors = self.cfg.GetClusterInfo().enabled_hypervisors
2216
    if constants.HTS_COPY_VNC_PASSWORD.intersection(enabled_hypervisors):
2217
      to_copy.append(constants.VNC_PASSWORD_FILE)
2218

    
2219
    for fname in to_copy:
2220
      result = self.rpc.call_upload_file([node], fname)
2221
      if result[node].failed or not result[node]:
2222
        logging.error("Could not copy file %s to node %s", fname, node)
2223

    
2224
    if self.op.readd:
2225
      self.context.ReaddNode(new_node)
2226
    else:
2227
      self.context.AddNode(new_node)
2228

    
2229

    
2230
class LUSetNodeParams(LogicalUnit):
2231
  """Modifies the parameters of a node.
2232

2233
  """
2234
  HPATH = "node-modify"
2235
  HTYPE = constants.HTYPE_NODE
2236
  _OP_REQP = ["node_name"]
2237
  REQ_BGL = False
2238

    
2239
  def CheckArguments(self):
2240
    node_name = self.cfg.ExpandNodeName(self.op.node_name)
2241
    if node_name is None:
2242
      raise errors.OpPrereqError("Invalid node name '%s'" % self.op.node_name)
2243
    self.op.node_name = node_name
2244
    _CheckBooleanOpField(self.op, 'master_candidate')
2245
    _CheckBooleanOpField(self.op, 'offline')
2246
    _CheckBooleanOpField(self.op, 'drained')
2247
    all_mods = [self.op.offline, self.op.master_candidate, self.op.drained]
2248
    if all_mods.count(None) == 3:
2249
      raise errors.OpPrereqError("Please pass at least one modification")
2250
    if all_mods.count(True) > 1:
2251
      raise errors.OpPrereqError("Can't set the node into more than one"
2252
                                 " state at the same time")
2253

    
2254
  def ExpandNames(self):
2255
    self.needed_locks = {locking.LEVEL_NODE: self.op.node_name}
2256

    
2257
  def BuildHooksEnv(self):
2258
    """Build hooks env.
2259

2260
    This runs on the master node.
2261

2262
    """
2263
    env = {
2264
      "OP_TARGET": self.op.node_name,
2265
      "MASTER_CANDIDATE": str(self.op.master_candidate),
2266
      "OFFLINE": str(self.op.offline),
2267
      "DRAINED": str(self.op.drained),
2268
      }
2269
    nl = [self.cfg.GetMasterNode(),
2270
          self.op.node_name]
2271
    return env, nl, nl
2272

    
2273
  def CheckPrereq(self):
2274
    """Check prerequisites.
2275

2276
    This only checks the instance list against the existing names.
2277

2278
    """
2279
    node = self.node = self.cfg.GetNodeInfo(self.op.node_name)
2280

    
2281
    if ((self.op.master_candidate == False or self.op.offline == True or
2282
         self.op.drained == True) and node.master_candidate):
2283
      # we will demote the node from master_candidate
2284
      if self.op.node_name == self.cfg.GetMasterNode():
2285
        raise errors.OpPrereqError("The master node has to be a"
2286
                                   " master candidate, online and not drained")
2287
      cp_size = self.cfg.GetClusterInfo().candidate_pool_size
2288
      num_candidates, _ = self.cfg.GetMasterCandidateStats()
2289
      if num_candidates <= cp_size:
2290
        msg = ("Not enough master candidates (desired"
2291
               " %d, new value will be %d)" % (cp_size, num_candidates-1))
2292
        if self.op.force:
2293
          self.LogWarning(msg)
2294
        else:
2295
          raise errors.OpPrereqError(msg)
2296

    
2297
    if (self.op.master_candidate == True and
2298
        ((node.offline and not self.op.offline == False) or
2299
         (node.drained and not self.op.drained == False))):
2300
      raise errors.OpPrereqError("Node '%s' is offline or drained, can't set"
2301
                                 " to master_candidate")
2302

    
2303
    return
2304

    
2305
  def Exec(self, feedback_fn):
2306
    """Modifies a node.
2307

2308
    """
2309
    node = self.node
2310

    
2311
    result = []
2312
    changed_mc = False
2313

    
2314
    if self.op.offline is not None:
2315
      node.offline = self.op.offline
2316
      result.append(("offline", str(self.op.offline)))
2317
      if self.op.offline == True:
2318
        if node.master_candidate:
2319
          node.master_candidate = False
2320
          changed_mc = True
2321
          result.append(("master_candidate", "auto-demotion due to offline"))
2322
        if node.drained:
2323
          node.drained = False
2324
          result.append(("drained", "clear drained status due to offline"))
2325

    
2326
    if self.op.master_candidate is not None:
2327
      node.master_candidate = self.op.master_candidate
2328
      changed_mc = True
2329
      result.append(("master_candidate", str(self.op.master_candidate)))
2330
      if self.op.master_candidate == False:
2331
        rrc = self.rpc.call_node_demote_from_mc(node.name)
2332
        msg = rrc.RemoteFailMsg()
2333
        if msg:
2334
          self.LogWarning("Node failed to demote itself: %s" % msg)
2335

    
2336
    if self.op.drained is not None:
2337
      node.drained = self.op.drained
2338
      result.append(("drained", str(self.op.drained)))
2339
      if self.op.drained == True:
2340
        if node.master_candidate:
2341
          node.master_candidate = False
2342
          changed_mc = True
2343
          result.append(("master_candidate", "auto-demotion due to drain"))
2344
        if node.offline:
2345
          node.offline = False
2346
          result.append(("offline", "clear offline status due to drain"))
2347

    
2348
    # this will trigger configuration file update, if needed
2349
    self.cfg.Update(node)
2350
    # this will trigger job queue propagation or cleanup
2351
    if changed_mc:
2352
      self.context.ReaddNode(node)
2353

    
2354
    return result
2355

    
2356

    
2357
class LUQueryClusterInfo(NoHooksLU):
2358
  """Query cluster configuration.
2359

2360
  """
2361
  _OP_REQP = []
2362
  REQ_BGL = False
2363

    
2364
  def ExpandNames(self):
2365
    self.needed_locks = {}
2366

    
2367
  def CheckPrereq(self):
2368
    """No prerequsites needed for this LU.
2369

2370
    """
2371
    pass
2372

    
2373
  def Exec(self, feedback_fn):
2374
    """Return cluster config.
2375

2376
    """
2377
    cluster = self.cfg.GetClusterInfo()
2378
    result = {
2379
      "software_version": constants.RELEASE_VERSION,
2380
      "protocol_version": constants.PROTOCOL_VERSION,
2381
      "config_version": constants.CONFIG_VERSION,
2382
      "os_api_version": constants.OS_API_VERSION,
2383
      "export_version": constants.EXPORT_VERSION,
2384
      "architecture": (platform.architecture()[0], platform.machine()),
2385
      "name": cluster.cluster_name,
2386
      "master": cluster.master_node,
2387
      "default_hypervisor": cluster.default_hypervisor,
2388
      "enabled_hypervisors": cluster.enabled_hypervisors,
2389
      "hvparams": dict([(hypervisor, cluster.hvparams[hypervisor])
2390
                        for hypervisor in cluster.enabled_hypervisors]),
2391
      "beparams": cluster.beparams,
2392
      "candidate_pool_size": cluster.candidate_pool_size,
2393
      }
2394

    
2395
    return result
2396

    
2397

    
2398
class LUQueryConfigValues(NoHooksLU):
2399
  """Return configuration values.
2400

2401
  """
2402
  _OP_REQP = []
2403
  REQ_BGL = False
2404
  _FIELDS_DYNAMIC = utils.FieldSet()
2405
  _FIELDS_STATIC = utils.FieldSet("cluster_name", "master_node", "drain_flag")
2406

    
2407
  def ExpandNames(self):
2408
    self.needed_locks = {}
2409

    
2410
    _CheckOutputFields(static=self._FIELDS_STATIC,
2411
                       dynamic=self._FIELDS_DYNAMIC,
2412
                       selected=self.op.output_fields)
2413

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

2417
    """
2418
    pass
2419

    
2420
  def Exec(self, feedback_fn):
2421
    """Dump a representation of the cluster config to the standard output.
2422

2423
    """
2424
    values = []
2425
    for field in self.op.output_fields:
2426
      if field == "cluster_name":
2427
        entry = self.cfg.GetClusterName()
2428
      elif field == "master_node":
2429
        entry = self.cfg.GetMasterNode()
2430
      elif field == "drain_flag":
2431
        entry = os.path.exists(constants.JOB_QUEUE_DRAIN_FILE)
2432
      else:
2433
        raise errors.ParameterError(field)
2434
      values.append(entry)
2435
    return values
2436

    
2437

    
2438
class LUActivateInstanceDisks(NoHooksLU):
2439
  """Bring up an instance's disks.
2440

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

    
2445
  def ExpandNames(self):
2446
    self._ExpandAndLockInstance()
2447
    self.needed_locks[locking.LEVEL_NODE] = []
2448
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2449

    
2450
  def DeclareLocks(self, level):
2451
    if level == locking.LEVEL_NODE:
2452
      self._LockInstancesNodes()
2453

    
2454
  def CheckPrereq(self):
2455
    """Check prerequisites.
2456

2457
    This checks that the instance is in the cluster.
2458

2459
    """
2460
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2461
    assert self.instance is not None, \
2462
      "Cannot retrieve locked instance %s" % self.op.instance_name
2463
    _CheckNodeOnline(self, self.instance.primary_node)
2464

    
2465
  def Exec(self, feedback_fn):
2466
    """Activate the disks.
2467

2468
    """
2469
    disks_ok, disks_info = _AssembleInstanceDisks(self, self.instance)
2470
    if not disks_ok:
2471
      raise errors.OpExecError("Cannot activate block devices")
2472

    
2473
    return disks_info
2474

    
2475

    
2476
def _AssembleInstanceDisks(lu, instance, ignore_secondaries=False):
2477
  """Prepare the block devices for an instance.
2478

2479
  This sets up the block devices on all nodes.
2480

2481
  @type lu: L{LogicalUnit}
2482
  @param lu: the logical unit on whose behalf we execute
2483
  @type instance: L{objects.Instance}
2484
  @param instance: the instance for whose disks we assemble
2485
  @type ignore_secondaries: boolean
2486
  @param ignore_secondaries: if true, errors on secondary nodes
2487
      won't result in an error return from the function
2488
  @return: False if the operation failed, otherwise a list of
2489
      (host, instance_visible_name, node_visible_name)
2490
      with the mapping from node devices to instance devices
2491

2492
  """
2493
  device_info = []
2494
  disks_ok = True
2495
  iname = instance.name
2496
  # With the two passes mechanism we try to reduce the window of
2497
  # opportunity for the race condition of switching DRBD to primary
2498
  # before handshaking occured, but we do not eliminate it
2499

    
2500
  # The proper fix would be to wait (with some limits) until the
2501
  # connection has been made and drbd transitions from WFConnection
2502
  # into any other network-connected state (Connected, SyncTarget,
2503
  # SyncSource, etc.)
2504

    
2505
  # 1st pass, assemble on all nodes in secondary mode
2506
  for inst_disk in instance.disks:
2507
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
2508
      lu.cfg.SetDiskID(node_disk, node)
2509
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, False)
2510
      msg = result.RemoteFailMsg()
2511
      if msg:
2512
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
2513
                           " (is_primary=False, pass=1): %s",
2514
                           inst_disk.iv_name, node, msg)
2515
        if not ignore_secondaries:
2516
          disks_ok = False
2517

    
2518
  # FIXME: race condition on drbd migration to primary
2519

    
2520
  # 2nd pass, do only the primary node
2521
  for inst_disk in instance.disks:
2522
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
2523
      if node != instance.primary_node:
2524
        continue
2525
      lu.cfg.SetDiskID(node_disk, node)
2526
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, True)
2527
      msg = result.RemoteFailMsg()
2528
      if msg:
2529
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
2530
                           " (is_primary=True, pass=2): %s",
2531
                           inst_disk.iv_name, node, msg)
2532
        disks_ok = False
2533
    device_info.append((instance.primary_node, inst_disk.iv_name,
2534
                        result.payload))
2535

    
2536
  # leave the disks configured for the primary node
2537
  # this is a workaround that would be fixed better by
2538
  # improving the logical/physical id handling
2539
  for disk in instance.disks:
2540
    lu.cfg.SetDiskID(disk, instance.primary_node)
2541

    
2542
  return disks_ok, device_info
2543

    
2544

    
2545
def _StartInstanceDisks(lu, instance, force):
2546
  """Start the disks of an instance.
2547

2548
  """
2549
  disks_ok, dummy = _AssembleInstanceDisks(lu, instance,
2550
                                           ignore_secondaries=force)
2551
  if not disks_ok:
2552
    _ShutdownInstanceDisks(lu, instance)
2553
    if force is not None and not force:
2554
      lu.proc.LogWarning("", hint="If the message above refers to a"
2555
                         " secondary node,"
2556
                         " you can retry the operation using '--force'.")
2557
    raise errors.OpExecError("Disk consistency error")
2558

    
2559

    
2560
class LUDeactivateInstanceDisks(NoHooksLU):
2561
  """Shutdown an instance's disks.
2562

2563
  """
2564
  _OP_REQP = ["instance_name"]
2565
  REQ_BGL = False
2566

    
2567
  def ExpandNames(self):
2568
    self._ExpandAndLockInstance()
2569
    self.needed_locks[locking.LEVEL_NODE] = []
2570
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2571

    
2572
  def DeclareLocks(self, level):
2573
    if level == locking.LEVEL_NODE:
2574
      self._LockInstancesNodes()
2575

    
2576
  def CheckPrereq(self):
2577
    """Check prerequisites.
2578

2579
    This checks that the instance is in the cluster.
2580

2581
    """
2582
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2583
    assert self.instance is not None, \
2584
      "Cannot retrieve locked instance %s" % self.op.instance_name
2585

    
2586
  def Exec(self, feedback_fn):
2587
    """Deactivate the disks
2588

2589
    """
2590
    instance = self.instance
2591
    _SafeShutdownInstanceDisks(self, instance)
2592

    
2593

    
2594
def _SafeShutdownInstanceDisks(lu, instance):
2595
  """Shutdown block devices of an instance.
2596

2597
  This function checks if an instance is running, before calling
2598
  _ShutdownInstanceDisks.
2599

2600
  """
2601
  ins_l = lu.rpc.call_instance_list([instance.primary_node],
2602
                                      [instance.hypervisor])
2603
  ins_l = ins_l[instance.primary_node]
2604
  if ins_l.failed or not isinstance(ins_l.data, list):
2605
    raise errors.OpExecError("Can't contact node '%s'" %
2606
                             instance.primary_node)
2607

    
2608
  if instance.name in ins_l.data:
2609
    raise errors.OpExecError("Instance is running, can't shutdown"
2610
                             " block devices.")
2611

    
2612
  _ShutdownInstanceDisks(lu, instance)
2613

    
2614

    
2615
def _ShutdownInstanceDisks(lu, instance, ignore_primary=False):
2616
  """Shutdown block devices of an instance.
2617

2618
  This does the shutdown on all nodes of the instance.
2619

2620
  If the ignore_primary is false, errors on the primary node are
2621
  ignored.
2622

2623
  """
2624
  all_result = True
2625
  for disk in instance.disks:
2626
    for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
2627
      lu.cfg.SetDiskID(top_disk, node)
2628
      result = lu.rpc.call_blockdev_shutdown(node, top_disk)
2629
      msg = result.RemoteFailMsg()
2630
      if msg:
2631
        lu.LogWarning("Could not shutdown block device %s on node %s: %s",
2632
                      disk.iv_name, node, msg)
2633
        if not ignore_primary or node != instance.primary_node:
2634
          all_result = False
2635
  return all_result
2636

    
2637

    
2638
def _CheckNodeFreeMemory(lu, node, reason, requested, hypervisor_name):
2639
  """Checks if a node has enough free memory.
2640

2641
  This function check if a given node has the needed amount of free
2642
  memory. In case the node has less memory or we cannot get the
2643
  information from the node, this function raise an OpPrereqError
2644
  exception.
2645

2646
  @type lu: C{LogicalUnit}
2647
  @param lu: a logical unit from which we get configuration data
2648
  @type node: C{str}
2649
  @param node: the node to check
2650
  @type reason: C{str}
2651
  @param reason: string to use in the error message
2652
  @type requested: C{int}
2653
  @param requested: the amount of memory in MiB to check for
2654
  @type hypervisor_name: C{str}
2655
  @param hypervisor_name: the hypervisor to ask for memory stats
2656
  @raise errors.OpPrereqError: if the node doesn't have enough memory, or
2657
      we cannot check the node
2658

2659
  """
2660
  nodeinfo = lu.rpc.call_node_info([node], lu.cfg.GetVGName(), hypervisor_name)
2661
  nodeinfo[node].Raise()
2662
  free_mem = nodeinfo[node].data.get('memory_free')
2663
  if not isinstance(free_mem, int):
2664
    raise errors.OpPrereqError("Can't compute free memory on node %s, result"
2665
                             " was '%s'" % (node, free_mem))
2666
  if requested > free_mem:
2667
    raise errors.OpPrereqError("Not enough memory on node %s for %s:"
2668
                             " needed %s MiB, available %s MiB" %
2669
                             (node, reason, requested, free_mem))
2670

    
2671

    
2672
class LUStartupInstance(LogicalUnit):
2673
  """Starts an instance.
2674

2675
  """
2676
  HPATH = "instance-start"
2677
  HTYPE = constants.HTYPE_INSTANCE
2678
  _OP_REQP = ["instance_name", "force"]
2679
  REQ_BGL = False
2680

    
2681
  def ExpandNames(self):
2682
    self._ExpandAndLockInstance()
2683

    
2684
  def BuildHooksEnv(self):
2685
    """Build hooks env.
2686

2687
    This runs on master, primary and secondary nodes of the instance.
2688

2689
    """
2690
    env = {
2691
      "FORCE": self.op.force,
2692
      }
2693
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
2694
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2695
    return env, nl, nl
2696

    
2697
  def CheckPrereq(self):
2698
    """Check prerequisites.
2699

2700
    This checks that the instance is in the cluster.
2701

2702
    """
2703
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2704
    assert self.instance is not None, \
2705
      "Cannot retrieve locked instance %s" % self.op.instance_name
2706

    
2707
    _CheckNodeOnline(self, instance.primary_node)
2708

    
2709
    bep = self.cfg.GetClusterInfo().FillBE(instance)
2710
    # check bridges existance
2711
    _CheckInstanceBridgesExist(self, instance)
2712

    
2713
    _CheckNodeFreeMemory(self, instance.primary_node,
2714
                         "starting instance %s" % instance.name,
2715
                         bep[constants.BE_MEMORY], instance.hypervisor)
2716

    
2717
  def Exec(self, feedback_fn):
2718
    """Start the instance.
2719

2720
    """
2721
    instance = self.instance
2722
    force = self.op.force
2723

    
2724
    self.cfg.MarkInstanceUp(instance.name)
2725

    
2726
    node_current = instance.primary_node
2727

    
2728
    _StartInstanceDisks(self, instance, force)
2729

    
2730
    result = self.rpc.call_instance_start(node_current, instance)
2731
    msg = result.RemoteFailMsg()
2732
    if msg:
2733
      _ShutdownInstanceDisks(self, instance)
2734
      raise errors.OpExecError("Could not start instance: %s" % msg)
2735

    
2736

    
2737
class LURebootInstance(LogicalUnit):
2738
  """Reboot an instance.
2739

2740
  """
2741
  HPATH = "instance-reboot"
2742
  HTYPE = constants.HTYPE_INSTANCE
2743
  _OP_REQP = ["instance_name", "ignore_secondaries", "reboot_type"]
2744
  REQ_BGL = False
2745

    
2746
  def ExpandNames(self):
2747
    if self.op.reboot_type not in [constants.INSTANCE_REBOOT_SOFT,
2748
                                   constants.INSTANCE_REBOOT_HARD,
2749
                                   constants.INSTANCE_REBOOT_FULL]:
2750
      raise errors.ParameterError("reboot type not in [%s, %s, %s]" %
2751
                                  (constants.INSTANCE_REBOOT_SOFT,
2752
                                   constants.INSTANCE_REBOOT_HARD,
2753
                                   constants.INSTANCE_REBOOT_FULL))
2754
    self._ExpandAndLockInstance()
2755

    
2756
  def BuildHooksEnv(self):
2757
    """Build hooks env.
2758

2759
    This runs on master, primary and secondary nodes of the instance.
2760

2761
    """
2762
    env = {
2763
      "IGNORE_SECONDARIES": self.op.ignore_secondaries,
2764
      }
2765
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
2766
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2767
    return env, nl, nl
2768

    
2769
  def CheckPrereq(self):
2770
    """Check prerequisites.
2771

2772
    This checks that the instance is in the cluster.
2773

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

    
2779
    _CheckNodeOnline(self, instance.primary_node)
2780

    
2781
    # check bridges existance
2782
    _CheckInstanceBridgesExist(self, instance)
2783

    
2784
  def Exec(self, feedback_fn):
2785
    """Reboot the instance.
2786

2787
    """
2788
    instance = self.instance
2789
    ignore_secondaries = self.op.ignore_secondaries
2790
    reboot_type = self.op.reboot_type
2791

    
2792
    node_current = instance.primary_node
2793

    
2794
    if reboot_type in [constants.INSTANCE_REBOOT_SOFT,
2795
                       constants.INSTANCE_REBOOT_HARD]:
2796
      for disk in instance.disks:
2797
        self.cfg.SetDiskID(disk, node_current)
2798
      result = self.rpc.call_instance_reboot(node_current, instance,
2799
                                             reboot_type)
2800
      msg = result.RemoteFailMsg()
2801
      if msg:
2802
        raise errors.OpExecError("Could not reboot instance: %s" % msg)
2803
    else:
2804
      result = self.rpc.call_instance_shutdown(node_current, instance)
2805
      msg = result.RemoteFailMsg()
2806
      if msg:
2807
        raise errors.OpExecError("Could not shutdown instance for"
2808
                                 " full reboot: %s" % msg)
2809
      _ShutdownInstanceDisks(self, instance)
2810
      _StartInstanceDisks(self, instance, ignore_secondaries)
2811
      result = self.rpc.call_instance_start(node_current, instance)
2812
      msg = result.RemoteFailMsg()
2813
      if msg:
2814
        _ShutdownInstanceDisks(self, instance)
2815
        raise errors.OpExecError("Could not start instance for"
2816
                                 " full reboot: %s" % msg)
2817

    
2818
    self.cfg.MarkInstanceUp(instance.name)
2819

    
2820

    
2821
class LUShutdownInstance(LogicalUnit):
2822
  """Shutdown an instance.
2823

2824
  """
2825
  HPATH = "instance-stop"
2826
  HTYPE = constants.HTYPE_INSTANCE
2827
  _OP_REQP = ["instance_name"]
2828
  REQ_BGL = False
2829

    
2830
  def ExpandNames(self):
2831
    self._ExpandAndLockInstance()
2832

    
2833
  def BuildHooksEnv(self):
2834
    """Build hooks env.
2835

2836
    This runs on master, primary and secondary nodes of the instance.
2837

2838
    """
2839
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2840
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2841
    return env, nl, nl
2842

    
2843
  def CheckPrereq(self):
2844
    """Check prerequisites.
2845

2846
    This checks that the instance is in the cluster.
2847

2848
    """
2849
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2850
    assert self.instance is not None, \
2851
      "Cannot retrieve locked instance %s" % self.op.instance_name
2852
    _CheckNodeOnline(self, self.instance.primary_node)
2853

    
2854
  def Exec(self, feedback_fn):
2855
    """Shutdown the instance.
2856

2857
    """
2858
    instance = self.instance
2859
    node_current = instance.primary_node
2860
    self.cfg.MarkInstanceDown(instance.name)
2861
    result = self.rpc.call_instance_shutdown(node_current, instance)
2862
    msg = result.RemoteFailMsg()
2863
    if msg:
2864
      self.proc.LogWarning("Could not shutdown instance: %s" % msg)
2865

    
2866
    _ShutdownInstanceDisks(self, instance)
2867

    
2868

    
2869
class LUReinstallInstance(LogicalUnit):
2870
  """Reinstall an instance.
2871

2872
  """
2873
  HPATH = "instance-reinstall"
2874
  HTYPE = constants.HTYPE_INSTANCE
2875
  _OP_REQP = ["instance_name"]
2876
  REQ_BGL = False
2877

    
2878
  def ExpandNames(self):
2879
    self._ExpandAndLockInstance()
2880

    
2881
  def BuildHooksEnv(self):
2882
    """Build hooks env.
2883

2884
    This runs on master, primary and secondary nodes of the instance.
2885

2886
    """
2887
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2888
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2889
    return env, nl, nl
2890

    
2891
  def CheckPrereq(self):
2892
    """Check prerequisites.
2893

2894
    This checks that the instance is in the cluster and is not running.
2895

2896
    """
2897
    instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2898
    assert instance is not None, \
2899
      "Cannot retrieve locked instance %s" % self.op.instance_name
2900
    _CheckNodeOnline(self, instance.primary_node)
2901

    
2902
    if instance.disk_template == constants.DT_DISKLESS:
2903
      raise errors.OpPrereqError("Instance '%s' has no disks" %
2904
                                 self.op.instance_name)
2905
    if instance.admin_up:
2906
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2907
                                 self.op.instance_name)
2908
    remote_info = self.rpc.call_instance_info(instance.primary_node,
2909
                                              instance.name,
2910
                                              instance.hypervisor)
2911
    if remote_info.failed or remote_info.data:
2912
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
2913
                                 (self.op.instance_name,
2914
                                  instance.primary_node))
2915

    
2916
    self.op.os_type = getattr(self.op, "os_type", None)
2917
    if self.op.os_type is not None:
2918
      # OS verification
2919
      pnode = self.cfg.GetNodeInfo(
2920
        self.cfg.ExpandNodeName(instance.primary_node))
2921
      if pnode is None:
2922
        raise errors.OpPrereqError("Primary node '%s' is unknown" %
2923
                                   self.op.pnode)
2924
      result = self.rpc.call_os_get(pnode.name, self.op.os_type)
2925
      result.Raise()
2926
      if not isinstance(result.data, objects.OS):
2927
        raise errors.OpPrereqError("OS '%s' not in supported OS list for"
2928
                                   " primary node"  % self.op.os_type)
2929

    
2930
    self.instance = instance
2931

    
2932
  def Exec(self, feedback_fn):
2933
    """Reinstall the instance.
2934

2935
    """
2936
    inst = self.instance
2937

    
2938
    if self.op.os_type is not None:
2939
      feedback_fn("Changing OS to '%s'..." % self.op.os_type)
2940
      inst.os = self.op.os_type
2941
      self.cfg.Update(inst)
2942

    
2943
    _StartInstanceDisks(self, inst, None)
2944
    try:
2945
      feedback_fn("Running the instance OS create scripts...")
2946
      result = self.rpc.call_instance_os_add(inst.primary_node, inst)
2947
      msg = result.RemoteFailMsg()
2948
      if msg:
2949
        raise errors.OpExecError("Could not install OS for instance %s"
2950
                                 " on node %s: %s" %
2951
                                 (inst.name, inst.primary_node, msg))
2952
    finally:
2953
      _ShutdownInstanceDisks(self, inst)
2954

    
2955

    
2956
class LURenameInstance(LogicalUnit):
2957
  """Rename an instance.
2958

2959
  """
2960
  HPATH = "instance-rename"
2961
  HTYPE = constants.HTYPE_INSTANCE
2962
  _OP_REQP = ["instance_name", "new_name"]
2963

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

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

2969
    """
2970
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2971
    env["INSTANCE_NEW_NAME"] = self.op.new_name
2972
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2973
    return env, nl, nl
2974

    
2975
  def CheckPrereq(self):
2976
    """Check prerequisites.
2977

2978
    This checks that the instance is in the cluster and is not running.
2979

2980
    """
2981
    instance = self.cfg.GetInstanceInfo(
2982
      self.cfg.ExpandInstanceName(self.op.instance_name))
2983
    if instance is None:
2984
      raise errors.OpPrereqError("Instance '%s' not known" %
2985
                                 self.op.instance_name)
2986
    _CheckNodeOnline(self, instance.primary_node)
2987

    
2988
    if instance.admin_up:
2989
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2990
                                 self.op.instance_name)
2991
    remote_info = self.rpc.call_instance_info(instance.primary_node,
2992
                                              instance.name,
2993
                                              instance.hypervisor)
2994
    remote_info.Raise()
2995
    if remote_info.data:
2996
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
2997
                                 (self.op.instance_name,
2998
                                  instance.primary_node))
2999
    self.instance = instance
3000

    
3001
    # new name verification
3002
    name_info = utils.HostInfo(self.op.new_name)
3003

    
3004
    self.op.new_name = new_name = name_info.name
3005
    instance_list = self.cfg.GetInstanceList()
3006
    if new_name in instance_list:
3007
      raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
3008
                                 new_name)
3009

    
3010
    if not getattr(self.op, "ignore_ip", False):
3011
      if utils.TcpPing(name_info.ip, constants.DEFAULT_NODED_PORT):
3012
        raise errors.OpPrereqError("IP %s of instance %s already in use" %
3013
                                   (name_info.ip, new_name))
3014

    
3015

    
3016
  def Exec(self, feedback_fn):
3017
    """Reinstall the instance.
3018

3019
    """
3020
    inst = self.instance
3021
    old_name = inst.name
3022

    
3023
    if inst.disk_template == constants.DT_FILE:
3024
      old_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
3025

    
3026
    self.cfg.RenameInstance(inst.name, self.op.new_name)
3027
    # Change the instance lock. This is definitely safe while we hold the BGL
3028
    self.context.glm.remove(locking.LEVEL_INSTANCE, old_name)
3029
    self.context.glm.add(locking.LEVEL_INSTANCE, self.op.new_name)
3030

    
3031
    # re-read the instance from the configuration after rename
3032
    inst = self.cfg.GetInstanceInfo(self.op.new_name)
3033

    
3034
    if inst.disk_template == constants.DT_FILE:
3035
      new_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
3036
      result = self.rpc.call_file_storage_dir_rename(inst.primary_node,
3037
                                                     old_file_storage_dir,
3038
                                                     new_file_storage_dir)
3039
      result.Raise()
3040
      if not result.data:
3041
        raise errors.OpExecError("Could not connect to node '%s' to rename"
3042
                                 " directory '%s' to '%s' (but the instance"
3043
                                 " has been renamed in Ganeti)" % (
3044
                                 inst.primary_node, old_file_storage_dir,
3045
                                 new_file_storage_dir))
3046

    
3047
      if not result.data[0]:
3048
        raise errors.OpExecError("Could not rename directory '%s' to '%s'"
3049
                                 " (but the instance has been renamed in"
3050
                                 " Ganeti)" % (old_file_storage_dir,
3051
                                               new_file_storage_dir))
3052

    
3053
    _StartInstanceDisks(self, inst, None)
3054
    try:
3055
      result = self.rpc.call_instance_run_rename(inst.primary_node, inst,
3056
                                                 old_name)
3057
      msg = result.RemoteFailMsg()
3058
      if msg:
3059
        msg = ("Could not run OS rename script for instance %s on node %s"
3060
               " (but the instance has been renamed in Ganeti): %s" %
3061
               (inst.name, inst.primary_node, msg))
3062
        self.proc.LogWarning(msg)
3063
    finally:
3064
      _ShutdownInstanceDisks(self, inst)
3065

    
3066

    
3067
class LURemoveInstance(LogicalUnit):
3068
  """Remove an instance.
3069

3070
  """
3071
  HPATH = "instance-remove"
3072
  HTYPE = constants.HTYPE_INSTANCE
3073
  _OP_REQP = ["instance_name", "ignore_failures"]
3074
  REQ_BGL = False
3075

    
3076
  def ExpandNames(self):
3077
    self._ExpandAndLockInstance()
3078
    self.needed_locks[locking.LEVEL_NODE] = []
3079
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3080

    
3081
  def DeclareLocks(self, level):
3082
    if level == locking.LEVEL_NODE:
3083
      self._LockInstancesNodes()
3084

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

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

3090
    """
3091
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3092
    nl = [self.cfg.GetMasterNode()]
3093
    return env, nl, nl
3094

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

3098
    This checks that the instance is in the cluster.
3099

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

    
3105
  def Exec(self, feedback_fn):
3106
    """Remove the instance.
3107

3108
    """
3109
    instance = self.instance
3110
    logging.info("Shutting down instance %s on node %s",
3111
                 instance.name, instance.primary_node)
3112

    
3113
    result = self.rpc.call_instance_shutdown(instance.primary_node, instance)
3114
    msg = result.RemoteFailMsg()
3115
    if msg:
3116
      if self.op.ignore_failures:
3117
        feedback_fn("Warning: can't shutdown instance: %s" % msg)
3118
      else:
3119
        raise errors.OpExecError("Could not shutdown instance %s on"
3120
                                 " node %s: %s" %
3121
                                 (instance.name, instance.primary_node, msg))
3122

    
3123
    logging.info("Removing block devices for instance %s", instance.name)
3124

    
3125
    if not _RemoveDisks(self, instance):
3126
      if self.op.ignore_failures:
3127
        feedback_fn("Warning: can't remove instance's disks")
3128
      else:
3129
        raise errors.OpExecError("Can't remove instance's disks")
3130

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

    
3133
    self.cfg.RemoveInstance(instance.name)
3134
    self.remove_locks[locking.LEVEL_INSTANCE] = instance.name
3135

    
3136

    
3137
class LUQueryInstances(NoHooksLU):
3138
  """Logical unit for querying instances.
3139

3140
  """
3141
  _OP_REQP = ["output_fields", "names", "use_locking"]
3142
  REQ_BGL = False
3143
  _FIELDS_STATIC = utils.FieldSet(*["name", "os", "pnode", "snodes",
3144
                                    "admin_state",
3145
                                    "disk_template", "ip", "mac", "bridge",
3146
                                    "sda_size", "sdb_size", "vcpus", "tags",
3147
                                    "network_port", "beparams",
3148
                                    r"(disk)\.(size)/([0-9]+)",
3149
                                    r"(disk)\.(sizes)", "disk_usage",
3150
                                    r"(nic)\.(mac|ip|bridge)/([0-9]+)",
3151
                                    r"(nic)\.(macs|ips|bridges)",
3152
                                    r"(disk|nic)\.(count)",
3153
                                    "serial_no", "hypervisor", "hvparams",] +
3154
                                  ["hv/%s" % name
3155
                                   for name in constants.HVS_PARAMETERS] +
3156
                                  ["be/%s" % name
3157
                                   for name in constants.BES_PARAMETERS])
3158
  _FIELDS_DYNAMIC = utils.FieldSet("oper_state", "oper_ram", "status")
3159

    
3160

    
3161
  def ExpandNames(self):
3162
    _CheckOutputFields(static=self._FIELDS_STATIC,
3163
                       dynamic=self._FIELDS_DYNAMIC,
3164
                       selected=self.op.output_fields)
3165

    
3166
    self.needed_locks = {}
3167
    self.share_locks[locking.LEVEL_INSTANCE] = 1
3168
    self.share_locks[locking.LEVEL_NODE] = 1
3169

    
3170
    if self.op.names:
3171
      self.wanted = _GetWantedInstances(self, self.op.names)
3172
    else:
3173
      self.wanted = locking.ALL_SET
3174

    
3175
    self.do_node_query = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
3176
    self.do_locking = self.do_node_query and self.op.use_locking
3177
    if self.do_locking:
3178
      self.needed_locks[locking.LEVEL_INSTANCE] = self.wanted
3179
      self.needed_locks[locking.LEVEL_NODE] = []
3180
      self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3181

    
3182
  def DeclareLocks(self, level):
3183
    if level == locking.LEVEL_NODE and self.do_locking:
3184
      self._LockInstancesNodes()
3185

    
3186
  def CheckPrereq(self):
3187
    """Check prerequisites.
3188

3189
    """
3190
    pass
3191

    
3192
  def Exec(self, feedback_fn):
3193
    """Computes the list of nodes and their attributes.
3194

3195
    """
3196
    all_info = self.cfg.GetAllInstancesInfo()
3197
    if self.wanted == locking.ALL_SET:
3198
      # caller didn't specify instance names, so ordering is not important
3199
      if self.do_locking:
3200
        instance_names = self.acquired_locks[locking.LEVEL_INSTANCE]
3201
      else:
3202
        instance_names = all_info.keys()
3203
      instance_names = utils.NiceSort(instance_names)
3204
    else:
3205
      # caller did specify names, so we must keep the ordering
3206
      if self.do_locking:
3207
        tgt_set = self.acquired_locks[locking.LEVEL_INSTANCE]
3208
      else:
3209
        tgt_set = all_info.keys()
3210
      missing = set(self.wanted).difference(tgt_set)
3211
      if missing:
3212
        raise errors.OpExecError("Some instances were removed before"
3213
                                 " retrieving their data: %s" % missing)
3214
      instance_names = self.wanted
3215

    
3216
    instance_list = [all_info[iname] for iname in instance_names]
3217

    
3218
    # begin data gathering
3219

    
3220
    nodes = frozenset([inst.primary_node for inst in instance_list])
3221
    hv_list = list(set([inst.hypervisor for inst in instance_list]))
3222

    
3223
    bad_nodes = []
3224
    off_nodes = []
3225
    if self.do_node_query:
3226
      live_data = {}
3227
      node_data = self.rpc.call_all_instances_info(nodes, hv_list)
3228
      for name in nodes:
3229
        result = node_data[name]
3230
        if result.offline:
3231
          # offline nodes will be in both lists
3232
          off_nodes.append(name)
3233
        if result.failed:
3234
          bad_nodes.append(name)
3235
        else:
3236
          if result.data:
3237
            live_data.update(result.data)
3238
            # else no instance is alive
3239
    else:
3240
      live_data = dict([(name, {}) for name in instance_names])
3241

    
3242
    # end data gathering
3243

    
3244
    HVPREFIX = "hv/"
3245
    BEPREFIX = "be/"
3246
    output = []
3247
    for instance in instance_list:
3248
      iout = []
3249
      i_hv = self.cfg.GetClusterInfo().FillHV(instance)
3250
      i_be = self.cfg.GetClusterInfo().FillBE(instance)
3251
      for field in self.op.output_fields:
3252
        st_match = self._FIELDS_STATIC.Matches(field)
3253
        if field == "name":
3254
          val = instance.name
3255
        elif field == "os":
3256
          val = instance.os
3257
        elif field == "pnode":
3258
          val = instance.primary_node
3259
        elif field == "snodes":
3260
          val = list(instance.secondary_nodes)
3261
        elif field == "admin_state":
3262
          val = instance.admin_up
3263
        elif field == "oper_state":
3264
          if instance.primary_node in bad_nodes:
3265
            val = None
3266
          else:
3267
            val = bool(live_data.get(instance.name))
3268
        elif field == "status":
3269
          if instance.primary_node in off_nodes:
3270
            val = "ERROR_nodeoffline"
3271
          elif instance.primary_node in bad_nodes:
3272
            val = "ERROR_nodedown"
3273
          else:
3274
            running = bool(live_data.get(instance.name))
3275
            if running:
3276
              if instance.admin_up:
3277
                val = "running"
3278
              else:
3279
                val = "ERROR_up"
3280
            else:
3281
              if instance.admin_up:
3282
                val = "ERROR_down"
3283
              else:
3284
                val = "ADMIN_down"
3285
        elif field == "oper_ram":
3286
          if instance.primary_node in bad_nodes:
3287
            val = None
3288
          elif instance.name in live_data:
3289
            val = live_data[instance.name].get("memory", "?")
3290
          else:
3291
            val = "-"
3292
        elif field == "disk_template":
3293
          val = instance.disk_template
3294
        elif field == "ip":
3295
          val = instance.nics[0].ip
3296
        elif field == "bridge":
3297
          val = instance.nics[0].bridge
3298
        elif field == "mac":
3299
          val = instance.nics[0].mac
3300
        elif field == "sda_size" or field == "sdb_size":
3301
          idx = ord(field[2]) - ord('a')
3302
          try:
3303
            val = instance.FindDisk(idx).size
3304
          except errors.OpPrereqError:
3305
            val = None
3306
        elif field == "disk_usage": # total disk usage per node
3307
          disk_sizes = [{'size': disk.size} for disk in instance.disks]
3308
          val = _ComputeDiskSize(instance.disk_template, disk_sizes)
3309
        elif field == "tags":
3310
          val = list(instance.GetTags())
3311
        elif field == "serial_no":
3312
          val = instance.serial_no
3313
        elif field == "network_port":
3314
          val = instance.network_port
3315
        elif field == "hypervisor":
3316
          val = instance.hypervisor
3317
        elif field == "hvparams":
3318
          val = i_hv
3319
        elif (field.startswith(HVPREFIX) and
3320
              field[len(HVPREFIX):] in constants.HVS_PARAMETERS):
3321
          val = i_hv.get(field[len(HVPREFIX):], None)
3322
        elif field == "beparams":
3323
          val = i_be
3324
        elif (field.startswith(BEPREFIX) and
3325
              field[len(BEPREFIX):] in constants.BES_PARAMETERS):
3326
          val = i_be.get(field[len(BEPREFIX):], None)
3327
        elif st_match and st_match.groups():
3328
          # matches a variable list
3329
          st_groups = st_match.groups()
3330
          if st_groups and st_groups[0] == "disk":
3331
            if st_groups[1] == "count":
3332
              val = len(instance.disks)
3333
            elif st_groups[1] == "sizes":
3334
              val = [disk.size for disk in instance.disks]
3335
            elif st_groups[1] == "size":
3336
              try:
3337
                val = instance.FindDisk(st_groups[2]).size
3338
              except errors.OpPrereqError:
3339
                val = None
3340
            else:
3341
              assert False, "Unhandled disk parameter"
3342
          elif st_groups[0] == "nic":
3343
            if st_groups[1] == "count":
3344
              val = len(instance.nics)
3345
            elif st_groups[1] == "macs":
3346
              val = [nic.mac for nic in instance.nics]
3347
            elif st_groups[1] == "ips":
3348
              val = [nic.ip for nic in instance.nics]
3349
            elif st_groups[1] == "bridges":
3350
              val = [nic.bridge for nic in instance.nics]
3351
            else:
3352
              # index-based item
3353
              nic_idx = int(st_groups[2])
3354
              if nic_idx >= len(instance.nics):
3355
                val = None
3356
              else:
3357
                if st_groups[1] == "mac":
3358
                  val = instance.nics[nic_idx].mac
3359
                elif st_groups[1] == "ip":
3360
                  val = instance.nics[nic_idx].ip
3361
                elif st_groups[1] == "bridge":
3362
                  val = instance.nics[nic_idx].bridge
3363
                else:
3364
                  assert False, "Unhandled NIC parameter"
3365
          else:
3366
            assert False, "Unhandled variable parameter"
3367
        else:
3368
          raise errors.ParameterError(field)
3369
        iout.append(val)
3370
      output.append(iout)
3371

    
3372
    return output
3373

    
3374

    
3375
class LUFailoverInstance(LogicalUnit):
3376
  """Failover an instance.
3377

3378
  """
3379
  HPATH = "instance-failover"
3380
  HTYPE = constants.HTYPE_INSTANCE
3381
  _OP_REQP = ["instance_name", "ignore_consistency"]
3382
  REQ_BGL = False
3383

    
3384
  def ExpandNames(self):
3385
    self._ExpandAndLockInstance()
3386
    self.needed_locks[locking.LEVEL_NODE] = []
3387
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3388

    
3389
  def DeclareLocks(self, level):
3390
    if level == locking.LEVEL_NODE:
3391
      self._LockInstancesNodes()
3392

    
3393
  def BuildHooksEnv(self):
3394
    """Build hooks env.
3395

3396
    This runs on master, primary and secondary nodes of the instance.
3397

3398
    """
3399
    env = {
3400
      "IGNORE_CONSISTENCY": self.op.ignore_consistency,
3401
      }
3402
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3403
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3404
    return env, nl, nl
3405

    
3406
  def CheckPrereq(self):
3407
    """Check prerequisites.
3408

3409
    This checks that the instance is in the cluster.
3410

3411
    """
3412
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3413
    assert self.instance is not None, \
3414
      "Cannot retrieve locked instance %s" % self.op.instance_name
3415

    
3416
    bep = self.cfg.GetClusterInfo().FillBE(instance)
3417
    if instance.disk_template not in constants.DTS_NET_MIRROR:
3418
      raise errors.OpPrereqError("Instance's disk layout is not"
3419
                                 " network mirrored, cannot failover.")
3420

    
3421
    secondary_nodes = instance.secondary_nodes
3422
    if not secondary_nodes:
3423
      raise errors.ProgrammerError("no secondary node but using "
3424
                                   "a mirrored disk template")
3425

    
3426
    target_node = secondary_nodes[0]
3427
    _CheckNodeOnline(self, target_node)
3428
    _CheckNodeNotDrained(self, target_node)
3429
    # check memory requirements on the secondary node
3430
    _CheckNodeFreeMemory(self, target_node, "failing over instance %s" %
3431
                         instance.name, bep[constants.BE_MEMORY],
3432
                         instance.hypervisor)
3433

    
3434
    # check bridge existance
3435
    brlist = [nic.bridge for nic in instance.nics]
3436
    result = self.rpc.call_bridges_exist(target_node, brlist)
3437
    result.Raise()
3438
    if not result.data:
3439
      raise errors.OpPrereqError("One or more target bridges %s does not"
3440
                                 " exist on destination node '%s'" %
3441
                                 (brlist, target_node))
3442

    
3443
  def Exec(self, feedback_fn):
3444
    """Failover an instance.
3445

3446
    The failover is done by shutting it down on its present node and
3447
    starting it on the secondary.
3448

3449
    """
3450
    instance = self.instance
3451

    
3452
    source_node = instance.primary_node
3453
    target_node = instance.secondary_nodes[0]
3454

    
3455
    feedback_fn("* checking disk consistency between source and target")
3456
    for dev in instance.disks:
3457
      # for drbd, these are drbd over lvm
3458
      if not _CheckDiskConsistency(self, dev, target_node, False):
3459
        if instance.admin_up and not self.op.ignore_consistency:
3460
          raise errors.OpExecError("Disk %s is degraded on target node,"
3461
                                   " aborting failover." % dev.iv_name)
3462

    
3463
    feedback_fn("* shutting down instance on source node")
3464
    logging.info("Shutting down instance %s on node %s",
3465
                 instance.name, source_node)
3466

    
3467
    result = self.rpc.call_instance_shutdown(source_node, instance)
3468
    msg = result.RemoteFailMsg()
3469
    if msg:
3470
      if self.op.ignore_consistency:
3471
        self.proc.LogWarning("Could not shutdown instance %s on node %s."
3472
                             " Proceeding anyway. Please make sure node"
3473
                             " %s is down. Error details: %s",
3474
                             instance.name, source_node, source_node, msg)
3475
      else:
3476
        raise errors.OpExecError("Could not shutdown instance %s on"
3477
                                 " node %s: %s" %
3478
                                 (instance.name, source_node, msg))
3479

    
3480
    feedback_fn("* deactivating the instance's disks on source node")
3481
    if not _ShutdownInstanceDisks(self, instance, ignore_primary=True):
3482
      raise errors.OpExecError("Can't shut down the instance's disks.")
3483

    
3484
    instance.primary_node = target_node
3485
    # distribute new instance config to the other nodes
3486
    self.cfg.Update(instance)
3487

    
3488
    # Only start the instance if it's marked as up
3489
    if instance.admin_up:
3490
      feedback_fn("* activating the instance's disks on target node")
3491
      logging.info("Starting instance %s on node %s",
3492
                   instance.name, target_node)
3493

    
3494
      disks_ok, dummy = _AssembleInstanceDisks(self, instance,
3495
                                               ignore_secondaries=True)
3496
      if not disks_ok:
3497
        _ShutdownInstanceDisks(self, instance)
3498
        raise errors.OpExecError("Can't activate the instance's disks")
3499

    
3500
      feedback_fn("* starting the instance on the target node")
3501
      result = self.rpc.call_instance_start(target_node, instance)
3502
      msg = result.RemoteFailMsg()
3503
      if msg:
3504
        _ShutdownInstanceDisks(self, instance)
3505
        raise errors.OpExecError("Could not start instance %s on node %s: %s" %
3506
                                 (instance.name, target_node, msg))
3507

    
3508

    
3509
class LUMigrateInstance(LogicalUnit):
3510
  """Migrate an instance.
3511

3512
  This is migration without shutting down, compared to the failover,
3513
  which is done with shutdown.
3514

3515
  """
3516
  HPATH = "instance-migrate"
3517
  HTYPE = constants.HTYPE_INSTANCE
3518
  _OP_REQP = ["instance_name", "live", "cleanup"]
3519

    
3520
  REQ_BGL = False
3521

    
3522
  def ExpandNames(self):
3523
    self._ExpandAndLockInstance()
3524
    self.needed_locks[locking.LEVEL_NODE] = []
3525
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3526

    
3527
  def DeclareLocks(self, level):
3528
    if level == locking.LEVEL_NODE:
3529
      self._LockInstancesNodes()
3530

    
3531
  def BuildHooksEnv(self):
3532
    """Build hooks env.
3533

3534
    This runs on master, primary and secondary nodes of the instance.
3535

3536
    """
3537
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3538
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3539
    return env, nl, nl
3540

    
3541
  def CheckPrereq(self):
3542
    """Check prerequisites.
3543

3544
    This checks that the instance is in the cluster.
3545

3546
    """
3547
    instance = self.cfg.GetInstanceInfo(
3548
      self.cfg.ExpandInstanceName(self.op.instance_name))
3549
    if instance is None:
3550
      raise errors.OpPrereqError("Instance '%s' not known" %
3551
                                 self.op.instance_name)
3552

    
3553
    if instance.disk_template != constants.DT_DRBD8:
3554
      raise errors.OpPrereqError("Instance's disk layout is not"
3555
                                 " drbd8, cannot migrate.")
3556

    
3557
    secondary_nodes = instance.secondary_nodes
3558
    if not secondary_nodes:
3559
      raise errors.ConfigurationError("No secondary node but using"
3560
                                      " drbd8 disk template")
3561

    
3562
    i_be = self.cfg.GetClusterInfo().FillBE(instance)
3563

    
3564
    target_node = secondary_nodes[0]
3565
    # check memory requirements on the secondary node
3566
    _CheckNodeFreeMemory(self, target_node, "migrating instance %s" %
3567
                         instance.name, i_be[constants.BE_MEMORY],
3568
                         instance.hypervisor)
3569

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

    
3578
    if not self.op.cleanup:
3579
      _CheckNodeNotDrained(self, target_node)
3580
      result = self.rpc.call_instance_migratable(instance.primary_node,
3581
                                                 instance)
3582
      msg = result.RemoteFailMsg()
3583
      if msg:
3584
        raise errors.OpPrereqError("Can't migrate: %s - please use failover" %
3585
                                   msg)
3586

    
3587
    self.instance = instance
3588

    
3589
  def _WaitUntilSync(self):
3590
    """Poll with custom rpc for disk sync.
3591

3592
    This uses our own step-based rpc call.
3593

3594
    """
3595
    self.feedback_fn("* wait until resync is done")
3596
    all_done = False
3597
    while not all_done:
3598
      all_done = True
3599
      result = self.rpc.call_drbd_wait_sync(self.all_nodes,
3600
                                            self.nodes_ip,
3601
                                            self.instance.disks)
3602
      min_percent = 100
3603
      for node, nres in result.items():
3604
        msg = nres.RemoteFailMsg()
3605
        if msg:
3606
          raise errors.OpExecError("Cannot resync disks on node %s: %s" %
3607
                                   (node, msg))
3608
        node_done, node_percent = nres.payload
3609
        all_done = all_done and node_done
3610
        if node_percent is not None:
3611
          min_percent = min(min_percent, node_percent)
3612
      if not all_done:
3613
        if min_percent < 100:
3614
          self.feedback_fn("   - progress: %.1f%%" % min_percent)
3615
        time.sleep(2)
3616

    
3617
  def _EnsureSecondary(self, node):
3618
    """Demote a node to secondary.
3619

3620
    """
3621
    self.feedback_fn("* switching node %s to secondary mode" % node)
3622

    
3623
    for dev in self.instance.disks:
3624
      self.cfg.SetDiskID(dev, node)
3625

    
3626
    result = self.rpc.call_blockdev_close(node, self.instance.name,
3627
                                          self.instance.disks)
3628
    msg = result.RemoteFailMsg()
3629
    if msg:
3630
      raise errors.OpExecError("Cannot change disk to secondary on node %s,"
3631
                               " error %s" % (node, msg))
3632

    
3633
  def _GoStandalone(self):
3634
    """Disconnect from the network.
3635

3636
    """
3637
    self.feedback_fn("* changing into standalone mode")
3638
    result = self.rpc.call_drbd_disconnect_net(self.all_nodes, self.nodes_ip,
3639
                                               self.instance.disks)
3640
    for node, nres in result.items():
3641
      msg = nres.RemoteFailMsg()
3642
      if msg:
3643
        raise errors.OpExecError("Cannot disconnect disks node %s,"
3644
                                 " error %s" % (node, msg))
3645

    
3646
  def _GoReconnect(self, multimaster):
3647
    """Reconnect to the network.
3648

3649
    """
3650
    if multimaster:
3651
      msg = "dual-master"
3652
    else:
3653
      msg = "single-master"
3654
    self.feedback_fn("* changing disks into %s mode" % msg)
3655
    result = self.rpc.call_drbd_attach_net(self.all_nodes, self.nodes_ip,
3656
                                           self.instance.disks,
3657
                                           self.instance.name, multimaster)
3658
    for node, nres in result.items():
3659
      msg = nres.RemoteFailMsg()
3660
      if msg:
3661
        raise errors.OpExecError("Cannot change disks config on node %s,"
3662
                                 " error: %s" % (node, msg))
3663

    
3664
  def _ExecCleanup(self):
3665
    """Try to cleanup after a failed migration.
3666

3667
    The cleanup is done by:
3668
      - check that the instance is running only on one node
3669
        (and update the config if needed)
3670
      - change disks on its secondary node to secondary
3671
      - wait until disks are fully synchronized
3672
      - disconnect from the network
3673
      - change disks into single-master mode
3674
      - wait again until disks are fully synchronized
3675

3676
    """
3677
    instance = self.instance
3678
    target_node = self.target_node
3679
    source_node = self.source_node
3680

    
3681
    # check running on only one node
3682
    self.feedback_fn("* checking where the instance actually runs"
3683
                     " (if this hangs, the hypervisor might be in"
3684
                     " a bad state)")
3685
    ins_l = self.rpc.call_instance_list(self.all_nodes, [instance.hypervisor])
3686
    for node, result in ins_l.items():
3687
      result.Raise()
3688
      if not isinstance(result.data, list):
3689
        raise errors.OpExecError("Can't contact node '%s'" % node)
3690

    
3691
    runningon_source = instance.name in ins_l[source_node].data
3692
    runningon_target = instance.name in ins_l[target_node].data
3693

    
3694
    if runningon_source and runningon_target:
3695
      raise errors.OpExecError("Instance seems to be running on two nodes,"
3696
                               " or the hypervisor is confused. You will have"
3697
                               " to ensure manually that it runs only on one"
3698
                               " and restart this operation.")
3699

    
3700
    if not (runningon_source or runningon_target):
3701
      raise errors.OpExecError("Instance does not seem to be running at all."
3702
                               " In this case, it's safer to repair by"
3703
                               " running 'gnt-instance stop' to ensure disk"
3704
                               " shutdown, and then restarting it.")
3705

    
3706
    if runningon_target:
3707
      # the migration has actually succeeded, we need to update the config
3708
      self.feedback_fn("* instance running on secondary node (%s),"
3709
                       " updating config" % target_node)
3710
      instance.primary_node = target_node
3711
      self.cfg.Update(instance)
3712
      demoted_node = source_node
3713
    else:
3714
      self.feedback_fn("* instance confirmed to be running on its"
3715
                       " primary node (%s)" % source_node)
3716
      demoted_node = target_node
3717

    
3718
    self._EnsureSecondary(demoted_node)
3719
    try:
3720
      self._WaitUntilSync()
3721
    except errors.OpExecError:
3722
      # we ignore here errors, since if the device is standalone, it
3723
      # won't be able to sync
3724
      pass
3725
    self._GoStandalone()
3726
    self._GoReconnect(False)
3727
    self._WaitUntilSync()
3728

    
3729
    self.feedback_fn("* done")
3730

    
3731
  def _RevertDiskStatus(self):
3732
    """Try to revert the disk status after a failed migration.
3733

3734
    """
3735
    target_node = self.target_node
3736
    try:
3737
      self._EnsureSecondary(target_node)
3738
      self._GoStandalone()
3739
      self._GoReconnect(False)
3740
      self._WaitUntilSync()
3741
    except errors.OpExecError, err:
3742
      self.LogWarning("Migration failed and I can't reconnect the"
3743
                      " drives: error '%s'\n"
3744
                      "Please look and recover the instance status" %
3745
                      str(err))
3746

    
3747
  def _AbortMigration(self):
3748
    """Call the hypervisor code to abort a started migration.
3749

3750
    """
3751
    instance = self.instance
3752
    target_node = self.target_node
3753
    migration_info = self.migration_info
3754

    
3755
    abort_result = self.rpc.call_finalize_migration(target_node,
3756
                                                    instance,
3757
                                                    migration_info,
3758
                                                    False)
3759
    abort_msg = abort_result.RemoteFailMsg()
3760
    if abort_msg:
3761
      logging.error("Aborting migration failed on target node %s: %s" %
3762
                    (target_node, abort_msg))
3763
      # Don't raise an exception here, as we stil have to try to revert the
3764
      # disk status, even if this step failed.
3765

    
3766
  def _ExecMigration(self):
3767
    """Migrate an instance.
3768

3769
    The migrate is done by:
3770
      - change the disks into dual-master mode
3771
      - wait until disks are fully synchronized again
3772
      - migrate the instance
3773
      - change disks on the new secondary node (the old primary) to secondary
3774
      - wait until disks are fully synchronized
3775
      - change disks into single-master mode
3776

3777
    """
3778
    instance = self.instance
3779
    target_node = self.target_node
3780
    source_node = self.source_node
3781

    
3782
    self.feedback_fn("* checking disk consistency between source and target")
3783
    for dev in instance.disks:
3784
      if not _CheckDiskConsistency(self, dev, target_node, False):
3785
        raise errors.OpExecError("Disk %s is degraded or not fully"
3786
                                 " synchronized on target node,"
3787
                                 " aborting migrate." % dev.iv_name)
3788

    
3789
    # First get the migration information from the remote node
3790
    result = self.rpc.call_migration_info(source_node, instance)
3791
    msg = result.RemoteFailMsg()
3792
    if msg:
3793
      log_err = ("Failed fetching source migration information from %s: %s" %
3794
                 (source_node, msg))
3795
      logging.error(log_err)
3796
      raise errors.OpExecError(log_err)
3797

    
3798
    self.migration_info = migration_info = result.payload
3799

    
3800
    # Then switch the disks to master/master mode
3801
    self._EnsureSecondary(target_node)
3802
    self._GoStandalone()
3803
    self._GoReconnect(True)
3804
    self._WaitUntilSync()
3805

    
3806
    self.feedback_fn("* preparing %s to accept the instance" % target_node)
3807
    result = self.rpc.call_accept_instance(target_node,
3808
                                           instance,
3809
                                           migration_info,
3810
                                           self.nodes_ip[target_node])
3811

    
3812
    msg = result.RemoteFailMsg()
3813
    if msg:
3814
      logging.error("Instance pre-migration failed, trying to revert"
3815
                    " disk status: %s", msg)
3816
      self._AbortMigration()
3817
      self._RevertDiskStatus()
3818
      raise errors.OpExecError("Could not pre-migrate instance %s: %s" %
3819
                               (instance.name, msg))
3820

    
3821
    self.feedback_fn("* migrating instance to %s" % target_node)
3822
    time.sleep(10)
3823
    result = self.rpc.call_instance_migrate(source_node, instance,
3824
                                            self.nodes_ip[target_node],
3825
                                            self.op.live)
3826
    msg = result.RemoteFailMsg()
3827
    if msg:
3828
      logging.error("Instance migration failed, trying to revert"
3829
                    " disk status: %s", msg)
3830
      self._AbortMigration()
3831
      self._RevertDiskStatus()
3832
      raise errors.OpExecError("Could not migrate instance %s: %s" %
3833
                               (instance.name, msg))
3834
    time.sleep(10)
3835

    
3836
    instance.primary_node = target_node
3837
    # distribute new instance config to the other nodes
3838
    self.cfg.Update(instance)
3839

    
3840
    result = self.rpc.call_finalize_migration(target_node,
3841
                                              instance,
3842
                                              migration_info,
3843
                                              True)
3844
    msg = result.RemoteFailMsg()
3845
    if msg:
3846
      logging.error("Instance migration succeeded, but finalization failed:"
3847
                    " %s" % msg)
3848
      raise errors.OpExecError("Could not finalize instance migration: %s" %
3849
                               msg)
3850

    
3851
    self._EnsureSecondary(source_node)
3852
    self._WaitUntilSync()
3853
    self._GoStandalone()
3854
    self._GoReconnect(False)
3855
    self._WaitUntilSync()
3856

    
3857
    self.feedback_fn("* done")
3858

    
3859
  def Exec(self, feedback_fn):
3860
    """Perform the migration.
3861

3862
    """
3863
    self.feedback_fn = feedback_fn
3864

    
3865
    self.source_node = self.instance.primary_node
3866
    self.target_node = self.instance.secondary_nodes[0]
3867
    self.all_nodes = [self.source_node, self.target_node]
3868
    self.nodes_ip = {
3869
      self.source_node: self.cfg.GetNodeInfo(self.source_node).secondary_ip,
3870
      self.target_node: self.cfg.GetNodeInfo(self.target_node).secondary_ip,
3871
      }
3872
    if self.op.cleanup:
3873
      return self._ExecCleanup()
3874
    else:
3875
      return self._ExecMigration()
3876

    
3877

    
3878
def _CreateBlockDev(lu, node, instance, device, force_create,
3879
                    info, force_open):
3880
  """Create a tree of block devices on a given node.
3881

3882
  If this device type has to be created on secondaries, create it and
3883
  all its children.
3884

3885
  If not, just recurse to children keeping the same 'force' value.
3886

3887
  @param lu: the lu on whose behalf we execute
3888
  @param node: the node on which to create the device
3889
  @type instance: L{objects.Instance}
3890
  @param instance: the instance which owns the device
3891
  @type device: L{objects.Disk}
3892
  @param device: the device to create
3893
  @type force_create: boolean
3894
  @param force_create: whether to force creation of this device; this
3895
      will be change to True whenever we find a device which has
3896
      CreateOnSecondary() attribute
3897
  @param info: the extra 'metadata' we should attach to the device
3898
      (this will be represented as a LVM tag)
3899
  @type force_open: boolean
3900
  @param force_open: this parameter will be passes to the
3901
      L{backend.BlockdevCreate} function where it specifies
3902
      whether we run on primary or not, and it affects both
3903
      the child assembly and the device own Open() execution
3904

3905
  """
3906
  if device.CreateOnSecondary():
3907
    force_create = True
3908

    
3909
  if device.children:
3910
    for child in device.children:
3911
      _CreateBlockDev(lu, node, instance, child, force_create,
3912
                      info, force_open)
3913

    
3914
  if not force_create:
3915
    return
3916

    
3917
  _CreateSingleBlockDev(lu, node, instance, device, info, force_open)
3918

    
3919

    
3920
def _CreateSingleBlockDev(lu, node, instance, device, info, force_open):
3921
  """Create a single block device on a given node.
3922

3923
  This will not recurse over children of the device, so they must be
3924
  created in advance.
3925

3926
  @param lu: the lu on whose behalf we execute
3927
  @param node: the node on which to create the device
3928
  @type instance: L{objects.Instance}
3929
  @param instance: the instance which owns the device
3930
  @type device: L{objects.Disk}
3931
  @param device: the device to create
3932
  @param info: the extra 'metadata' we should attach to the device
3933
      (this will be represented as a LVM tag)
3934
  @type force_open: boolean
3935
  @param force_open: this parameter will be passes to the
3936
      L{backend.BlockdevCreate} function where it specifies
3937
      whether we run on primary or not, and it affects both
3938
      the child assembly and the device own Open() execution
3939

3940
  """
3941
  lu.cfg.SetDiskID(device, node)
3942
  result = lu.rpc.call_blockdev_create(node, device, device.size,
3943
                                       instance.name, force_open, info)
3944
  msg = result.RemoteFailMsg()
3945
  if msg:
3946
    raise errors.OpExecError("Can't create block device %s on"
3947
                             " node %s for instance %s: %s" %
3948
                             (device, node, instance.name, msg))
3949
  if device.physical_id is None:
3950
    device.physical_id = result.payload
3951

    
3952

    
3953
def _GenerateUniqueNames(lu, exts):
3954
  """Generate a suitable LV name.
3955

3956
  This will generate a logical volume name for the given instance.
3957

3958
  """
3959
  results = []
3960
  for val in exts:
3961
    new_id = lu.cfg.GenerateUniqueID()
3962
    results.append("%s%s" % (new_id, val))
3963
  return results
3964

    
3965

    
3966
def _GenerateDRBD8Branch(lu, primary, secondary, size, names, iv_name,
3967
                         p_minor, s_minor):
3968
  """Generate a drbd8 device complete with its children.
3969

3970
  """
3971
  port = lu.cfg.AllocatePort()
3972
  vgname = lu.cfg.GetVGName()
3973
  shared_secret = lu.cfg.GenerateDRBDSecret()
3974
  dev_data = objects.Disk(dev_type=constants.LD_LV, size=size,
3975
                          logical_id=(vgname, names[0]))
3976
  dev_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
3977
                          logical_id=(vgname, names[1]))
3978
  drbd_dev = objects.Disk(dev_type=constants.LD_DRBD8, size=size,
3979
                          logical_id=(primary, secondary, port,
3980
                                      p_minor, s_minor,
3981
                                      shared_secret),
3982
                          children=[dev_data, dev_meta],
3983
                          iv_name=iv_name)
3984
  return drbd_dev
3985

    
3986

    
3987
def _GenerateDiskTemplate(lu, template_name,
3988
                          instance_name, primary_node,
3989
                          secondary_nodes, disk_info,
3990
                          file_storage_dir, file_driver,
3991
                          base_index):
3992
  """Generate the entire disk layout for a given template type.
3993

3994
  """
3995
  #TODO: compute space requirements
3996

    
3997
  vgname = lu.cfg.GetVGName()
3998
  disk_count = len(disk_info)
3999
  disks = []
4000
  if template_name == constants.DT_DISKLESS:
4001
    pass
4002
  elif template_name == constants.DT_PLAIN:
4003
    if len(secondary_nodes) != 0:
4004
      raise errors.ProgrammerError("Wrong template configuration")
4005

    
4006
    names = _GenerateUniqueNames(lu, [".disk%d" % i
4007
                                      for i in range(disk_count)])
4008
    for idx, disk in enumerate(disk_info):
4009
      disk_index = idx + base_index
4010
      disk_dev = objects.Disk(dev_type=constants.LD_LV, size=disk["size"],
4011
                              logical_id=(vgname, names[idx]),
4012
                              iv_name="disk/%d" % disk_index,
4013
                              mode=disk["mode"])
4014
      disks.append(disk_dev)
4015
  elif template_name == constants.DT_DRBD8:
4016
    if len(secondary_nodes) != 1:
4017
      raise errors.ProgrammerError("Wrong template configuration