Statistics
| Branch: | Tag: | Revision:

root / lib / cmdlib.py @ 6906a9d8

History | View | Annotate | Download (231.9 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 = lu.cfg.GetInstanceList()
396
  return utils.NiceSort(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 nodes 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 _BuildInstanceHookEnv(name, primary_node, secondary_nodes, os_type, status,
445
                          memory, vcpus, nics):
446
  """Builds instance related env variables for hooks
447

448
  This builds the hook environment from individual variables.
449

450
  @type name: string
451
  @param name: the name of the instance
452
  @type primary_node: string
453
  @param primary_node: the name of the instance's primary node
454
  @type secondary_nodes: list
455
  @param secondary_nodes: list of secondary nodes as strings
456
  @type os_type: string
457
  @param os_type: the name of the instance's OS
458
  @type status: string
459
  @param status: the desired status of the instances
460
  @type memory: string
461
  @param memory: the memory size of the instance
462
  @type vcpus: string
463
  @param vcpus: the count of VCPUs the instance has
464
  @type nics: list
465
  @param nics: list of tuples (ip, bridge, mac) representing
466
      the NICs the instance  has
467
  @rtype: dict
468
  @return: the hook environment for this instance
469

470
  """
471
  env = {
472
    "OP_TARGET": name,
473
    "INSTANCE_NAME": name,
474
    "INSTANCE_PRIMARY": primary_node,
475
    "INSTANCE_SECONDARIES": " ".join(secondary_nodes),
476
    "INSTANCE_OS_TYPE": os_type,
477
    "INSTANCE_STATUS": status,
478
    "INSTANCE_MEMORY": memory,
479
    "INSTANCE_VCPUS": vcpus,
480
  }
481

    
482
  if nics:
483
    nic_count = len(nics)
484
    for idx, (ip, bridge, mac) in enumerate(nics):
485
      if ip is None:
486
        ip = ""
487
      env["INSTANCE_NIC%d_IP" % idx] = ip
488
      env["INSTANCE_NIC%d_BRIDGE" % idx] = bridge
489
      env["INSTANCE_NIC%d_HWADDR" % idx] = mac
490
  else:
491
    nic_count = 0
492

    
493
  env["INSTANCE_NIC_COUNT"] = nic_count
494

    
495
  return env
496

    
497

    
498
def _BuildInstanceHookEnvByObject(lu, instance, override=None):
499
  """Builds instance related env variables for hooks from an object.
500

501
  @type lu: L{LogicalUnit}
502
  @param lu: the logical unit on whose behalf we execute
503
  @type instance: L{objects.Instance}
504
  @param instance: the instance for which we should build the
505
      environment
506
  @type override: dict
507
  @param override: dictionary with key/values that will override
508
      our values
509
  @rtype: dict
510
  @return: the hook environment dictionary
511

512
  """
513
  bep = lu.cfg.GetClusterInfo().FillBE(instance)
514
  args = {
515
    'name': instance.name,
516
    'primary_node': instance.primary_node,
517
    'secondary_nodes': instance.secondary_nodes,
518
    'os_type': instance.os,
519
    'status': instance.os,
520
    'memory': bep[constants.BE_MEMORY],
521
    'vcpus': bep[constants.BE_VCPUS],
522
    'nics': [(nic.ip, nic.bridge, nic.mac) for nic in instance.nics],
523
  }
524
  if override:
525
    args.update(override)
526
  return _BuildInstanceHookEnv(**args)
527

    
528

    
529
def _AdjustCandidatePool(lu):
530
  """Adjust the candidate pool after node operations.
531

532
  """
533
  mod_list = lu.cfg.MaintainCandidatePool()
534
  if mod_list:
535
    lu.LogInfo("Promoted nodes to master candidate role: %s",
536
               ", ".join(node.name for node in mod_list))
537
    for name in mod_list:
538
      lu.context.ReaddNode(name)
539
  mc_now, mc_max = lu.cfg.GetMasterCandidateStats()
540
  if mc_now > mc_max:
541
    lu.LogInfo("Note: more nodes are candidates (%d) than desired (%d)" %
542
               (mc_now, mc_max))
543

    
544

    
545
def _CheckInstanceBridgesExist(lu, instance):
546
  """Check that the brigdes needed by an instance exist.
547

548
  """
549
  # check bridges existance
550
  brlist = [nic.bridge for nic in instance.nics]
551
  result = lu.rpc.call_bridges_exist(instance.primary_node, brlist)
552
  result.Raise()
553
  if not result.data:
554
    raise errors.OpPrereqError("One or more target bridges %s does not"
555
                               " exist on destination node '%s'" %
556
                               (brlist, instance.primary_node))
557

    
558

    
559
class LUDestroyCluster(NoHooksLU):
560
  """Logical unit for destroying the cluster.
561

562
  """
563
  _OP_REQP = []
564

    
565
  def CheckPrereq(self):
566
    """Check prerequisites.
567

568
    This checks whether the cluster is empty.
569

570
    Any errors are signalled by raising errors.OpPrereqError.
571

572
    """
573
    master = self.cfg.GetMasterNode()
574

    
575
    nodelist = self.cfg.GetNodeList()
576
    if len(nodelist) != 1 or nodelist[0] != master:
577
      raise errors.OpPrereqError("There are still %d node(s) in"
578
                                 " this cluster." % (len(nodelist) - 1))
579
    instancelist = self.cfg.GetInstanceList()
580
    if instancelist:
581
      raise errors.OpPrereqError("There are still %d instance(s) in"
582
                                 " this cluster." % len(instancelist))
583

    
584
  def Exec(self, feedback_fn):
585
    """Destroys the cluster.
586

587
    """
588
    master = self.cfg.GetMasterNode()
589
    result = self.rpc.call_node_stop_master(master, False)
590
    result.Raise()
591
    if not result.data:
592
      raise errors.OpExecError("Could not disable the master role")
593
    priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
594
    utils.CreateBackup(priv_key)
595
    utils.CreateBackup(pub_key)
596
    return master
597

    
598

    
599
class LUVerifyCluster(LogicalUnit):
600
  """Verifies the cluster status.
601

602
  """
603
  HPATH = "cluster-verify"
604
  HTYPE = constants.HTYPE_CLUSTER
605
  _OP_REQP = ["skip_checks"]
606
  REQ_BGL = False
607

    
608
  def ExpandNames(self):
609
    self.needed_locks = {
610
      locking.LEVEL_NODE: locking.ALL_SET,
611
      locking.LEVEL_INSTANCE: locking.ALL_SET,
612
    }
613
    self.share_locks = dict(((i, 1) for i in locking.LEVELS))
614

    
615
  def _VerifyNode(self, nodeinfo, file_list, local_cksum,
616
                  node_result, feedback_fn, master_files,
617
                  drbd_map):
618
    """Run multiple tests against a node.
619

620
    Test list:
621

622
      - compares ganeti version
623
      - checks vg existance and size > 20G
624
      - checks config file checksum
625
      - checks ssh to other nodes
626

627
    @type nodeinfo: L{objects.Node}
628
    @param nodeinfo: the node to check
629
    @param file_list: required list of files
630
    @param local_cksum: dictionary of local files and their checksums
631
    @param node_result: the results from the node
632
    @param feedback_fn: function used to accumulate results
633
    @param master_files: list of files that only masters should have
634
    @param drbd_map: the useddrbd minors for this node, in
635
        form of minor: (instance, must_exist) which correspond to instances
636
        and their running status
637

638
    """
639
    node = nodeinfo.name
640

    
641
    # main result, node_result should be a non-empty dict
642
    if not node_result or not isinstance(node_result, dict):
643
      feedback_fn("  - ERROR: unable to verify node %s." % (node,))
644
      return True
645

    
646
    # compares ganeti version
647
    local_version = constants.PROTOCOL_VERSION
648
    remote_version = node_result.get('version', None)
649
    if not remote_version:
650
      feedback_fn("  - ERROR: connection to %s failed" % (node))
651
      return True
652

    
653
    if local_version != remote_version:
654
      feedback_fn("  - ERROR: sw version mismatch: master %s, node(%s) %s" %
655
                      (local_version, node, remote_version))
656
      return True
657

    
658
    # checks vg existance and size > 20G
659

    
660
    bad = False
661
    vglist = node_result.get(constants.NV_VGLIST, None)
662
    if not vglist:
663
      feedback_fn("  - ERROR: unable to check volume groups on node %s." %
664
                      (node,))
665
      bad = True
666
    else:
667
      vgstatus = utils.CheckVolumeGroupSize(vglist, self.cfg.GetVGName(),
668
                                            constants.MIN_VG_SIZE)
669
      if vgstatus:
670
        feedback_fn("  - ERROR: %s on node %s" % (vgstatus, node))
671
        bad = True
672

    
673
    # checks config file checksum
674

    
675
    remote_cksum = node_result.get(constants.NV_FILELIST, None)
676
    if not isinstance(remote_cksum, dict):
677
      bad = True
678
      feedback_fn("  - ERROR: node hasn't returned file checksum data")
679
    else:
680
      for file_name in file_list:
681
        node_is_mc = nodeinfo.master_candidate
682
        must_have_file = file_name not in master_files
683
        if file_name not in remote_cksum:
684
          if node_is_mc or must_have_file:
685
            bad = True
686
            feedback_fn("  - ERROR: file '%s' missing" % file_name)
687
        elif remote_cksum[file_name] != local_cksum[file_name]:
688
          if node_is_mc or must_have_file:
689
            bad = True
690
            feedback_fn("  - ERROR: file '%s' has wrong checksum" % file_name)
691
          else:
692
            # not candidate and this is not a must-have file
693
            bad = True
694
            feedback_fn("  - ERROR: non master-candidate has old/wrong file"
695
                        " '%s'" % file_name)
696
        else:
697
          # all good, except non-master/non-must have combination
698
          if not node_is_mc and not must_have_file:
699
            feedback_fn("  - ERROR: file '%s' should not exist on non master"
700
                        " candidates" % file_name)
701

    
702
    # checks ssh to any
703

    
704
    if constants.NV_NODELIST not in node_result:
705
      bad = True
706
      feedback_fn("  - ERROR: node hasn't returned node ssh connectivity data")
707
    else:
708
      if node_result[constants.NV_NODELIST]:
709
        bad = True
710
        for node in node_result[constants.NV_NODELIST]:
711
          feedback_fn("  - ERROR: ssh communication with node '%s': %s" %
712
                          (node, node_result[constants.NV_NODELIST][node]))
713

    
714
    if constants.NV_NODENETTEST not in node_result:
715
      bad = True
716
      feedback_fn("  - ERROR: node hasn't returned node tcp connectivity data")
717
    else:
718
      if node_result[constants.NV_NODENETTEST]:
719
        bad = True
720
        nlist = utils.NiceSort(node_result[constants.NV_NODENETTEST].keys())
721
        for node in nlist:
722
          feedback_fn("  - ERROR: tcp communication with node '%s': %s" %
723
                          (node, node_result[constants.NV_NODENETTEST][node]))
724

    
725
    hyp_result = node_result.get(constants.NV_HYPERVISOR, None)
726
    if isinstance(hyp_result, dict):
727
      for hv_name, hv_result in hyp_result.iteritems():
728
        if hv_result is not None:
729
          feedback_fn("  - ERROR: hypervisor %s verify failure: '%s'" %
730
                      (hv_name, hv_result))
731

    
732
    # check used drbd list
733
    used_minors = node_result.get(constants.NV_DRBDLIST, [])
734
    for minor, (iname, must_exist) in drbd_map.items():
735
      if minor not in used_minors and must_exist:
736
        feedback_fn("  - ERROR: drbd minor %d of instance %s is not active" %
737
                    (minor, iname))
738
        bad = True
739
    for minor in used_minors:
740
      if minor not in drbd_map:
741
        feedback_fn("  - ERROR: unallocated drbd minor %d is in use" % minor)
742
        bad = True
743

    
744
    return bad
745

    
746
  def _VerifyInstance(self, instance, instanceconfig, node_vol_is,
747
                      node_instance, feedback_fn, n_offline):
748
    """Verify an instance.
749

750
    This function checks to see if the required block devices are
751
    available on the instance's node.
752

753
    """
754
    bad = False
755

    
756
    node_current = instanceconfig.primary_node
757

    
758
    node_vol_should = {}
759
    instanceconfig.MapLVsByNode(node_vol_should)
760

    
761
    for node in node_vol_should:
762
      if node in n_offline:
763
        # ignore missing volumes on offline nodes
764
        continue
765
      for volume in node_vol_should[node]:
766
        if node not in node_vol_is or volume not in node_vol_is[node]:
767
          feedback_fn("  - ERROR: volume %s missing on node %s" %
768
                          (volume, node))
769
          bad = True
770

    
771
    if not instanceconfig.status == 'down':
772
      if ((node_current not in node_instance or
773
          not instance in node_instance[node_current]) and
774
          node_current not in n_offline):
775
        feedback_fn("  - ERROR: instance %s not running on node %s" %
776
                        (instance, node_current))
777
        bad = True
778

    
779
    for node in node_instance:
780
      if (not node == node_current):
781
        if instance in node_instance[node]:
782
          feedback_fn("  - ERROR: instance %s should not run on node %s" %
783
                          (instance, node))
784
          bad = True
785

    
786
    return bad
787

    
788
  def _VerifyOrphanVolumes(self, node_vol_should, node_vol_is, feedback_fn):
789
    """Verify if there are any unknown volumes in the cluster.
790

791
    The .os, .swap and backup volumes are ignored. All other volumes are
792
    reported as unknown.
793

794
    """
795
    bad = False
796

    
797
    for node in node_vol_is:
798
      for volume in node_vol_is[node]:
799
        if node not in node_vol_should or volume not in node_vol_should[node]:
800
          feedback_fn("  - ERROR: volume %s on node %s should not exist" %
801
                      (volume, node))
802
          bad = True
803
    return bad
804

    
805
  def _VerifyOrphanInstances(self, instancelist, node_instance, feedback_fn):
806
    """Verify the list of running instances.
807

808
    This checks what instances are running but unknown to the cluster.
809

810
    """
811
    bad = False
812
    for node in node_instance:
813
      for runninginstance in node_instance[node]:
814
        if runninginstance not in instancelist:
815
          feedback_fn("  - ERROR: instance %s on node %s should not exist" %
816
                          (runninginstance, node))
817
          bad = True
818
    return bad
819

    
820
  def _VerifyNPlusOneMemory(self, node_info, instance_cfg, feedback_fn):
821
    """Verify N+1 Memory Resilience.
822

823
    Check that if one single node dies we can still start all the instances it
824
    was primary for.
825

826
    """
827
    bad = False
828

    
829
    for node, nodeinfo in node_info.iteritems():
830
      # This code checks that every node which is now listed as secondary has
831
      # enough memory to host all instances it is supposed to should a single
832
      # other node in the cluster fail.
833
      # FIXME: not ready for failover to an arbitrary node
834
      # FIXME: does not support file-backed instances
835
      # WARNING: we currently take into account down instances as well as up
836
      # ones, considering that even if they're down someone might want to start
837
      # them even in the event of a node failure.
838
      for prinode, instances in nodeinfo['sinst-by-pnode'].iteritems():
839
        needed_mem = 0
840
        for instance in instances:
841
          bep = self.cfg.GetClusterInfo().FillBE(instance_cfg[instance])
842
          if bep[constants.BE_AUTO_BALANCE]:
843
            needed_mem += bep[constants.BE_MEMORY]
844
        if nodeinfo['mfree'] < needed_mem:
845
          feedback_fn("  - ERROR: not enough memory on node %s to accomodate"
846
                      " failovers should node %s fail" % (node, prinode))
847
          bad = True
848
    return bad
849

    
850
  def CheckPrereq(self):
851
    """Check prerequisites.
852

853
    Transform the list of checks we're going to skip into a set and check that
854
    all its members are valid.
855

856
    """
857
    self.skip_set = frozenset(self.op.skip_checks)
858
    if not constants.VERIFY_OPTIONAL_CHECKS.issuperset(self.skip_set):
859
      raise errors.OpPrereqError("Invalid checks to be skipped specified")
860

    
861
  def BuildHooksEnv(self):
862
    """Build hooks env.
863

864
    Cluster-Verify hooks just rone in the post phase and their failure makes
865
    the output be logged in the verify output and the verification to fail.
866

867
    """
868
    all_nodes = self.cfg.GetNodeList()
869
    # TODO: populate the environment with useful information for verify hooks
870
    env = {}
871
    return env, [], all_nodes
872

    
873
  def Exec(self, feedback_fn):
874
    """Verify integrity of cluster, performing various test on nodes.
875

876
    """
877
    bad = False
878
    feedback_fn("* Verifying global settings")
879
    for msg in self.cfg.VerifyConfig():
880
      feedback_fn("  - ERROR: %s" % msg)
881

    
882
    vg_name = self.cfg.GetVGName()
883
    hypervisors = self.cfg.GetClusterInfo().enabled_hypervisors
884
    nodelist = utils.NiceSort(self.cfg.GetNodeList())
885
    nodeinfo = [self.cfg.GetNodeInfo(nname) for nname in nodelist]
886
    instancelist = utils.NiceSort(self.cfg.GetInstanceList())
887
    instanceinfo = dict((iname, self.cfg.GetInstanceInfo(iname))
888
                        for iname in instancelist)
889
    i_non_redundant = [] # Non redundant instances
890
    i_non_a_balanced = [] # Non auto-balanced instances
891
    n_offline = [] # List of offline nodes
892
    node_volume = {}
893
    node_instance = {}
894
    node_info = {}
895
    instance_cfg = {}
896

    
897
    # FIXME: verify OS list
898
    # do local checksums
899
    master_files = [constants.CLUSTER_CONF_FILE]
900

    
901
    file_names = ssconf.SimpleStore().GetFileList()
902
    file_names.append(constants.SSL_CERT_FILE)
903
    file_names.append(constants.RAPI_CERT_FILE)
904
    file_names.extend(master_files)
905

    
906
    local_checksums = utils.FingerprintFiles(file_names)
907

    
908
    feedback_fn("* Gathering data (%d nodes)" % len(nodelist))
909
    node_verify_param = {
910
      constants.NV_FILELIST: file_names,
911
      constants.NV_NODELIST: [node.name for node in nodeinfo
912
                              if not node.offline],
913
      constants.NV_HYPERVISOR: hypervisors,
914
      constants.NV_NODENETTEST: [(node.name, node.primary_ip,
915
                                  node.secondary_ip) for node in nodeinfo
916
                                 if not node.offline],
917
      constants.NV_LVLIST: vg_name,
918
      constants.NV_INSTANCELIST: hypervisors,
919
      constants.NV_VGLIST: None,
920
      constants.NV_VERSION: None,
921
      constants.NV_HVINFO: self.cfg.GetHypervisorType(),
922
      constants.NV_DRBDLIST: None,
923
      }
924
    all_nvinfo = self.rpc.call_node_verify(nodelist, node_verify_param,
925
                                           self.cfg.GetClusterName())
926

    
927
    cluster = self.cfg.GetClusterInfo()
928
    master_node = self.cfg.GetMasterNode()
929
    all_drbd_map = self.cfg.ComputeDRBDMap()
930

    
931
    for node_i in nodeinfo:
932
      node = node_i.name
933
      nresult = all_nvinfo[node].data
934

    
935
      if node_i.offline:
936
        feedback_fn("* Skipping offline node %s" % (node,))
937
        n_offline.append(node)
938
        continue
939

    
940
      if node == master_node:
941
        ntype = "master"
942
      elif node_i.master_candidate:
943
        ntype = "master candidate"
944
      else:
945
        ntype = "regular"
946
      feedback_fn("* Verifying node %s (%s)" % (node, ntype))
947

    
948
      if all_nvinfo[node].failed or not isinstance(nresult, dict):
949
        feedback_fn("  - ERROR: connection to %s failed" % (node,))
950
        bad = True
951
        continue
952

    
953
      node_drbd = {}
954
      for minor, instance in all_drbd_map[node].items():
955
        instance = instanceinfo[instance]
956
        node_drbd[minor] = (instance.name, instance.status == "up")
957
      result = self._VerifyNode(node_i, file_names, local_checksums,
958
                                nresult, feedback_fn, master_files,
959
                                node_drbd)
960
      bad = bad or result
961

    
962
      lvdata = nresult.get(constants.NV_LVLIST, "Missing LV data")
963
      if isinstance(lvdata, basestring):
964
        feedback_fn("  - ERROR: LVM problem on node %s: %s" %
965
                    (node, lvdata.encode('string_escape')))
966
        bad = True
967
        node_volume[node] = {}
968
      elif not isinstance(lvdata, dict):
969
        feedback_fn("  - ERROR: connection to %s failed (lvlist)" % (node,))
970
        bad = True
971
        continue
972
      else:
973
        node_volume[node] = lvdata
974

    
975
      # node_instance
976
      idata = nresult.get(constants.NV_INSTANCELIST, None)
977
      if not isinstance(idata, list):
978
        feedback_fn("  - ERROR: connection to %s failed (instancelist)" %
979
                    (node,))
980
        bad = True
981
        continue
982

    
983
      node_instance[node] = idata
984

    
985
      # node_info
986
      nodeinfo = nresult.get(constants.NV_HVINFO, None)
987
      if not isinstance(nodeinfo, dict):
988
        feedback_fn("  - ERROR: connection to %s failed (hvinfo)" % (node,))
989
        bad = True
990
        continue
991

    
992
      try:
993
        node_info[node] = {
994
          "mfree": int(nodeinfo['memory_free']),
995
          "dfree": int(nresult[constants.NV_VGLIST][vg_name]),
996
          "pinst": [],
997
          "sinst": [],
998
          # dictionary holding all instances this node is secondary for,
999
          # grouped by their primary node. Each key is a cluster node, and each
1000
          # value is a list of instances which have the key as primary and the
1001
          # current node as secondary.  this is handy to calculate N+1 memory
1002
          # availability if you can only failover from a primary to its
1003
          # secondary.
1004
          "sinst-by-pnode": {},
1005
        }
1006
      except ValueError:
1007
        feedback_fn("  - ERROR: invalid value returned from node %s" % (node,))
1008
        bad = True
1009
        continue
1010

    
1011
    node_vol_should = {}
1012

    
1013
    for instance in instancelist:
1014
      feedback_fn("* Verifying instance %s" % instance)
1015
      inst_config = instanceinfo[instance]
1016
      result =  self._VerifyInstance(instance, inst_config, node_volume,
1017
                                     node_instance, feedback_fn, n_offline)
1018
      bad = bad or result
1019
      inst_nodes_offline = []
1020

    
1021
      inst_config.MapLVsByNode(node_vol_should)
1022

    
1023
      instance_cfg[instance] = inst_config
1024

    
1025
      pnode = inst_config.primary_node
1026
      if pnode in node_info:
1027
        node_info[pnode]['pinst'].append(instance)
1028
      elif pnode not in n_offline:
1029
        feedback_fn("  - ERROR: instance %s, connection to primary node"
1030
                    " %s failed" % (instance, pnode))
1031
        bad = True
1032

    
1033
      if pnode in n_offline:
1034
        inst_nodes_offline.append(pnode)
1035

    
1036
      # If the instance is non-redundant we cannot survive losing its primary
1037
      # node, so we are not N+1 compliant. On the other hand we have no disk
1038
      # templates with more than one secondary so that situation is not well
1039
      # supported either.
1040
      # FIXME: does not support file-backed instances
1041
      if len(inst_config.secondary_nodes) == 0:
1042
        i_non_redundant.append(instance)
1043
      elif len(inst_config.secondary_nodes) > 1:
1044
        feedback_fn("  - WARNING: multiple secondaries for instance %s"
1045
                    % instance)
1046

    
1047
      if not cluster.FillBE(inst_config)[constants.BE_AUTO_BALANCE]:
1048
        i_non_a_balanced.append(instance)
1049

    
1050
      for snode in inst_config.secondary_nodes:
1051
        if snode in node_info:
1052
          node_info[snode]['sinst'].append(instance)
1053
          if pnode not in node_info[snode]['sinst-by-pnode']:
1054
            node_info[snode]['sinst-by-pnode'][pnode] = []
1055
          node_info[snode]['sinst-by-pnode'][pnode].append(instance)
1056
        elif snode not in n_offline:
1057
          feedback_fn("  - ERROR: instance %s, connection to secondary node"
1058
                      " %s failed" % (instance, snode))
1059
          bad = True
1060
        if snode in n_offline:
1061
          inst_nodes_offline.append(snode)
1062

    
1063
      if inst_nodes_offline:
1064
        # warn that the instance lives on offline nodes, and set bad=True
1065
        feedback_fn("  - ERROR: instance lives on offline node(s) %s" %
1066
                    ", ".join(inst_nodes_offline))
1067
        bad = True
1068

    
1069
    feedback_fn("* Verifying orphan volumes")
1070
    result = self._VerifyOrphanVolumes(node_vol_should, node_volume,
1071
                                       feedback_fn)
1072
    bad = bad or result
1073

    
1074
    feedback_fn("* Verifying remaining instances")
1075
    result = self._VerifyOrphanInstances(instancelist, node_instance,
1076
                                         feedback_fn)
1077
    bad = bad or result
1078

    
1079
    if constants.VERIFY_NPLUSONE_MEM not in self.skip_set:
1080
      feedback_fn("* Verifying N+1 Memory redundancy")
1081
      result = self._VerifyNPlusOneMemory(node_info, instance_cfg, feedback_fn)
1082
      bad = bad or result
1083

    
1084
    feedback_fn("* Other Notes")
1085
    if i_non_redundant:
1086
      feedback_fn("  - NOTICE: %d non-redundant instance(s) found."
1087
                  % len(i_non_redundant))
1088

    
1089
    if i_non_a_balanced:
1090
      feedback_fn("  - NOTICE: %d non-auto-balanced instance(s) found."
1091
                  % len(i_non_a_balanced))
1092

    
1093
    if n_offline:
1094
      feedback_fn("  - NOTICE: %d offline node(s) found." % len(n_offline))
1095

    
1096
    return not bad
1097

    
1098
  def HooksCallBack(self, phase, hooks_results, feedback_fn, lu_result):
1099
    """Analize the post-hooks' result
1100

1101
    This method analyses the hook result, handles it, and sends some
1102
    nicely-formatted feedback back to the user.
1103

1104
    @param phase: one of L{constants.HOOKS_PHASE_POST} or
1105
        L{constants.HOOKS_PHASE_PRE}; it denotes the hooks phase
1106
    @param hooks_results: the results of the multi-node hooks rpc call
1107
    @param feedback_fn: function used send feedback back to the caller
1108
    @param lu_result: previous Exec result
1109
    @return: the new Exec result, based on the previous result
1110
        and hook results
1111

1112
    """
1113
    # We only really run POST phase hooks, and are only interested in
1114
    # their results
1115
    if phase == constants.HOOKS_PHASE_POST:
1116
      # Used to change hooks' output to proper indentation
1117
      indent_re = re.compile('^', re.M)
1118
      feedback_fn("* Hooks Results")
1119
      if not hooks_results:
1120
        feedback_fn("  - ERROR: general communication failure")
1121
        lu_result = 1
1122
      else:
1123
        for node_name in hooks_results:
1124
          show_node_header = True
1125
          res = hooks_results[node_name]
1126
          if res.failed or res.data is False or not isinstance(res.data, list):
1127
            if res.offline:
1128
              # no need to warn or set fail return value
1129
              continue
1130
            feedback_fn("    Communication failure in hooks execution")
1131
            lu_result = 1
1132
            continue
1133
          for script, hkr, output in res.data:
1134
            if hkr == constants.HKR_FAIL:
1135
              # The node header is only shown once, if there are
1136
              # failing hooks on that node
1137
              if show_node_header:
1138
                feedback_fn("  Node %s:" % node_name)
1139
                show_node_header = False
1140
              feedback_fn("    ERROR: Script %s failed, output:" % script)
1141
              output = indent_re.sub('      ', output)
1142
              feedback_fn("%s" % output)
1143
              lu_result = 1
1144

    
1145
      return lu_result
1146

    
1147

    
1148
class LUVerifyDisks(NoHooksLU):
1149
  """Verifies the cluster disks status.
1150

1151
  """
1152
  _OP_REQP = []
1153
  REQ_BGL = False
1154

    
1155
  def ExpandNames(self):
1156
    self.needed_locks = {
1157
      locking.LEVEL_NODE: locking.ALL_SET,
1158
      locking.LEVEL_INSTANCE: locking.ALL_SET,
1159
    }
1160
    self.share_locks = dict(((i, 1) for i in locking.LEVELS))
1161

    
1162
  def CheckPrereq(self):
1163
    """Check prerequisites.
1164

1165
    This has no prerequisites.
1166

1167
    """
1168
    pass
1169

    
1170
  def Exec(self, feedback_fn):
1171
    """Verify integrity of cluster disks.
1172

1173
    """
1174
    result = res_nodes, res_nlvm, res_instances, res_missing = [], {}, [], {}
1175

    
1176
    vg_name = self.cfg.GetVGName()
1177
    nodes = utils.NiceSort(self.cfg.GetNodeList())
1178
    instances = [self.cfg.GetInstanceInfo(name)
1179
                 for name in self.cfg.GetInstanceList()]
1180

    
1181
    nv_dict = {}
1182
    for inst in instances:
1183
      inst_lvs = {}
1184
      if (inst.status != "up" or
1185
          inst.disk_template not in constants.DTS_NET_MIRROR):
1186
        continue
1187
      inst.MapLVsByNode(inst_lvs)
1188
      # transform { iname: {node: [vol,],},} to {(node, vol): iname}
1189
      for node, vol_list in inst_lvs.iteritems():
1190
        for vol in vol_list:
1191
          nv_dict[(node, vol)] = inst
1192

    
1193
    if not nv_dict:
1194
      return result
1195

    
1196
    node_lvs = self.rpc.call_volume_list(nodes, vg_name)
1197

    
1198
    to_act = set()
1199
    for node in nodes:
1200
      # node_volume
1201
      lvs = node_lvs[node]
1202
      if lvs.failed:
1203
        if not lvs.offline:
1204
          self.LogWarning("Connection to node %s failed: %s" %
1205
                          (node, lvs.data))
1206
        continue
1207
      lvs = lvs.data
1208
      if isinstance(lvs, basestring):
1209
        logging.warning("Error enumerating LVs on node %s: %s", node, lvs)
1210
        res_nlvm[node] = lvs
1211
      elif not isinstance(lvs, dict):
1212
        logging.warning("Connection to node %s failed or invalid data"
1213
                        " returned", node)
1214
        res_nodes.append(node)
1215
        continue
1216

    
1217
      for lv_name, (_, lv_inactive, lv_online) in lvs.iteritems():
1218
        inst = nv_dict.pop((node, lv_name), None)
1219
        if (not lv_online and inst is not None
1220
            and inst.name not in res_instances):
1221
          res_instances.append(inst.name)
1222

    
1223
    # any leftover items in nv_dict are missing LVs, let's arrange the
1224
    # data better
1225
    for key, inst in nv_dict.iteritems():
1226
      if inst.name not in res_missing:
1227
        res_missing[inst.name] = []
1228
      res_missing[inst.name].append(key)
1229

    
1230
    return result
1231

    
1232

    
1233
class LURenameCluster(LogicalUnit):
1234
  """Rename the cluster.
1235

1236
  """
1237
  HPATH = "cluster-rename"
1238
  HTYPE = constants.HTYPE_CLUSTER
1239
  _OP_REQP = ["name"]
1240

    
1241
  def BuildHooksEnv(self):
1242
    """Build hooks env.
1243

1244
    """
1245
    env = {
1246
      "OP_TARGET": self.cfg.GetClusterName(),
1247
      "NEW_NAME": self.op.name,
1248
      }
1249
    mn = self.cfg.GetMasterNode()
1250
    return env, [mn], [mn]
1251

    
1252
  def CheckPrereq(self):
1253
    """Verify that the passed name is a valid one.
1254

1255
    """
1256
    hostname = utils.HostInfo(self.op.name)
1257

    
1258
    new_name = hostname.name
1259
    self.ip = new_ip = hostname.ip
1260
    old_name = self.cfg.GetClusterName()
1261
    old_ip = self.cfg.GetMasterIP()
1262
    if new_name == old_name and new_ip == old_ip:
1263
      raise errors.OpPrereqError("Neither the name nor the IP address of the"
1264
                                 " cluster has changed")
1265
    if new_ip != old_ip:
1266
      if utils.TcpPing(new_ip, constants.DEFAULT_NODED_PORT):
1267
        raise errors.OpPrereqError("The given cluster IP address (%s) is"
1268
                                   " reachable on the network. Aborting." %
1269
                                   new_ip)
1270

    
1271
    self.op.name = new_name
1272

    
1273
  def Exec(self, feedback_fn):
1274
    """Rename the cluster.
1275

1276
    """
1277
    clustername = self.op.name
1278
    ip = self.ip
1279

    
1280
    # shutdown the master IP
1281
    master = self.cfg.GetMasterNode()
1282
    result = self.rpc.call_node_stop_master(master, False)
1283
    if result.failed or not result.data:
1284
      raise errors.OpExecError("Could not disable the master role")
1285

    
1286
    try:
1287
      cluster = self.cfg.GetClusterInfo()
1288
      cluster.cluster_name = clustername
1289
      cluster.master_ip = ip
1290
      self.cfg.Update(cluster)
1291

    
1292
      # update the known hosts file
1293
      ssh.WriteKnownHostsFile(self.cfg, constants.SSH_KNOWN_HOSTS_FILE)
1294
      node_list = self.cfg.GetNodeList()
1295
      try:
1296
        node_list.remove(master)
1297
      except ValueError:
1298
        pass
1299
      result = self.rpc.call_upload_file(node_list,
1300
                                         constants.SSH_KNOWN_HOSTS_FILE)
1301
      for to_node, to_result in result.iteritems():
1302
        if to_result.failed or not to_result.data:
1303
          logging.error("Copy of file %s to node %s failed",
1304
                        constants.SSH_KNOWN_HOSTS_FILE, to_node)
1305

    
1306
    finally:
1307
      result = self.rpc.call_node_start_master(master, False)
1308
      if result.failed or not result.data:
1309
        self.LogWarning("Could not re-enable the master role on"
1310
                        " the master, please restart manually.")
1311

    
1312

    
1313
def _RecursiveCheckIfLVMBased(disk):
1314
  """Check if the given disk or its children are lvm-based.
1315

1316
  @type disk: L{objects.Disk}
1317
  @param disk: the disk to check
1318
  @rtype: booleean
1319
  @return: boolean indicating whether a LD_LV dev_type was found or not
1320

1321
  """
1322
  if disk.children:
1323
    for chdisk in disk.children:
1324
      if _RecursiveCheckIfLVMBased(chdisk):
1325
        return True
1326
  return disk.dev_type == constants.LD_LV
1327

    
1328

    
1329
class LUSetClusterParams(LogicalUnit):
1330
  """Change the parameters of the cluster.
1331

1332
  """
1333
  HPATH = "cluster-modify"
1334
  HTYPE = constants.HTYPE_CLUSTER
1335
  _OP_REQP = []
1336
  REQ_BGL = False
1337

    
1338
  def CheckParameters(self):
1339
    """Check parameters
1340

1341
    """
1342
    if not hasattr(self.op, "candidate_pool_size"):
1343
      self.op.candidate_pool_size = None
1344
    if self.op.candidate_pool_size is not None:
1345
      try:
1346
        self.op.candidate_pool_size = int(self.op.candidate_pool_size)
1347
      except ValueError, err:
1348
        raise errors.OpPrereqError("Invalid candidate_pool_size value: %s" %
1349
                                   str(err))
1350
      if self.op.candidate_pool_size < 1:
1351
        raise errors.OpPrereqError("At least one master candidate needed")
1352

    
1353
  def ExpandNames(self):
1354
    # FIXME: in the future maybe other cluster params won't require checking on
1355
    # all nodes to be modified.
1356
    self.needed_locks = {
1357
      locking.LEVEL_NODE: locking.ALL_SET,
1358
    }
1359
    self.share_locks[locking.LEVEL_NODE] = 1
1360

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

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

    
1372
  def CheckPrereq(self):
1373
    """Check prerequisites.
1374

1375
    This checks whether the given params don't conflict and
1376
    if the given volume group is valid.
1377

1378
    """
1379
    # FIXME: This only works because there is only one parameter that can be
1380
    # changed or removed.
1381
    if self.op.vg_name is not None and not self.op.vg_name:
1382
      instances = self.cfg.GetAllInstancesInfo().values()
1383
      for inst in instances:
1384
        for disk in inst.disks:
1385
          if _RecursiveCheckIfLVMBased(disk):
1386
            raise errors.OpPrereqError("Cannot disable lvm storage while"
1387
                                       " lvm-based instances exist")
1388

    
1389
    node_list = self.acquired_locks[locking.LEVEL_NODE]
1390

    
1391
    # if vg_name not None, checks given volume group on all nodes
1392
    if self.op.vg_name:
1393
      vglist = self.rpc.call_vg_list(node_list)
1394
      for node in node_list:
1395
        if vglist[node].failed:
1396
          # ignoring down node
1397
          self.LogWarning("Node %s unreachable/error, ignoring" % node)
1398
          continue
1399
        vgstatus = utils.CheckVolumeGroupSize(vglist[node].data,
1400
                                              self.op.vg_name,
1401
                                              constants.MIN_VG_SIZE)
1402
        if vgstatus:
1403
          raise errors.OpPrereqError("Error on node '%s': %s" %
1404
                                     (node, vgstatus))
1405

    
1406
    self.cluster = cluster = self.cfg.GetClusterInfo()
1407
    # validate beparams changes
1408
    if self.op.beparams:
1409
      utils.CheckBEParams(self.op.beparams)
1410
      self.new_beparams = cluster.FillDict(
1411
        cluster.beparams[constants.BEGR_DEFAULT], self.op.beparams)
1412

    
1413
    # hypervisor list/parameters
1414
    self.new_hvparams = cluster.FillDict(cluster.hvparams, {})
1415
    if self.op.hvparams:
1416
      if not isinstance(self.op.hvparams, dict):
1417
        raise errors.OpPrereqError("Invalid 'hvparams' parameter on input")
1418
      for hv_name, hv_dict in self.op.hvparams.items():
1419
        if hv_name not in self.new_hvparams:
1420
          self.new_hvparams[hv_name] = hv_dict
1421
        else:
1422
          self.new_hvparams[hv_name].update(hv_dict)
1423

    
1424
    if self.op.enabled_hypervisors is not None:
1425
      self.hv_list = self.op.enabled_hypervisors
1426
    else:
1427
      self.hv_list = cluster.enabled_hypervisors
1428

    
1429
    if self.op.hvparams or self.op.enabled_hypervisors is not None:
1430
      # either the enabled list has changed, or the parameters have, validate
1431
      for hv_name, hv_params in self.new_hvparams.items():
1432
        if ((self.op.hvparams and hv_name in self.op.hvparams) or
1433
            (self.op.enabled_hypervisors and
1434
             hv_name in self.op.enabled_hypervisors)):
1435
          # either this is a new hypervisor, or its parameters have changed
1436
          hv_class = hypervisor.GetHypervisor(hv_name)
1437
          hv_class.CheckParameterSyntax(hv_params)
1438
          _CheckHVParams(self, node_list, hv_name, hv_params)
1439

    
1440
  def Exec(self, feedback_fn):
1441
    """Change the parameters of the cluster.
1442

1443
    """
1444
    if self.op.vg_name is not None:
1445
      if self.op.vg_name != self.cfg.GetVGName():
1446
        self.cfg.SetVGName(self.op.vg_name)
1447
      else:
1448
        feedback_fn("Cluster LVM configuration already in desired"
1449
                    " state, not changing")
1450
    if self.op.hvparams:
1451
      self.cluster.hvparams = self.new_hvparams
1452
    if self.op.enabled_hypervisors is not None:
1453
      self.cluster.enabled_hypervisors = self.op.enabled_hypervisors
1454
    if self.op.beparams:
1455
      self.cluster.beparams[constants.BEGR_DEFAULT] = self.new_beparams
1456
    if self.op.candidate_pool_size is not None:
1457
      self.cluster.candidate_pool_size = self.op.candidate_pool_size
1458

    
1459
    self.cfg.Update(self.cluster)
1460

    
1461
    # we want to update nodes after the cluster so that if any errors
1462
    # happen, we have recorded and saved the cluster info
1463
    if self.op.candidate_pool_size is not None:
1464
      _AdjustCandidatePool(self)
1465

    
1466

    
1467
class LURedistributeConfig(NoHooksLU):
1468
  """Force the redistribution of cluster configuration.
1469

1470
  This is a very simple LU.
1471

1472
  """
1473
  _OP_REQP = []
1474
  REQ_BGL = False
1475

    
1476
  def ExpandNames(self):
1477
    self.needed_locks = {
1478
      locking.LEVEL_NODE: locking.ALL_SET,
1479
    }
1480
    self.share_locks[locking.LEVEL_NODE] = 1
1481

    
1482
  def CheckPrereq(self):
1483
    """Check prerequisites.
1484

1485
    """
1486

    
1487
  def Exec(self, feedback_fn):
1488
    """Redistribute the configuration.
1489

1490
    """
1491
    self.cfg.Update(self.cfg.GetClusterInfo())
1492

    
1493

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

1497
  """
1498
  if not instance.disks:
1499
    return True
1500

    
1501
  if not oneshot:
1502
    lu.proc.LogInfo("Waiting for instance %s to sync disks." % instance.name)
1503

    
1504
  node = instance.primary_node
1505

    
1506
  for dev in instance.disks:
1507
    lu.cfg.SetDiskID(dev, node)
1508

    
1509
  retries = 0
1510
  while True:
1511
    max_time = 0
1512
    done = True
1513
    cumul_degraded = False
1514
    rstats = lu.rpc.call_blockdev_getmirrorstatus(node, instance.disks)
1515
    if rstats.failed or not rstats.data:
1516
      lu.LogWarning("Can't get any data from node %s", node)
1517
      retries += 1
1518
      if retries >= 10:
1519
        raise errors.RemoteError("Can't contact node %s for mirror data,"
1520
                                 " aborting." % node)
1521
      time.sleep(6)
1522
      continue
1523
    rstats = rstats.data
1524
    retries = 0
1525
    for i, mstat in enumerate(rstats):
1526
      if mstat is None:
1527
        lu.LogWarning("Can't compute data for node %s/%s",
1528
                           node, instance.disks[i].iv_name)
1529
        continue
1530
      # we ignore the ldisk parameter
1531
      perc_done, est_time, is_degraded, _ = mstat
1532
      cumul_degraded = cumul_degraded or (is_degraded and perc_done is None)
1533
      if perc_done is not None:
1534
        done = False
1535
        if est_time is not None:
1536
          rem_time = "%d estimated seconds remaining" % est_time
1537
          max_time = est_time
1538
        else:
1539
          rem_time = "no time estimate"
1540
        lu.proc.LogInfo("- device %s: %5.2f%% done, %s" %
1541
                        (instance.disks[i].iv_name, perc_done, rem_time))
1542
    if done or oneshot:
1543
      break
1544

    
1545
    time.sleep(min(60, max_time))
1546

    
1547
  if done:
1548
    lu.proc.LogInfo("Instance %s's disks are in sync." % instance.name)
1549
  return not cumul_degraded
1550

    
1551

    
1552
def _CheckDiskConsistency(lu, dev, node, on_primary, ldisk=False):
1553
  """Check that mirrors are not degraded.
1554

1555
  The ldisk parameter, if True, will change the test from the
1556
  is_degraded attribute (which represents overall non-ok status for
1557
  the device(s)) to the ldisk (representing the local storage status).
1558

1559
  """
1560
  lu.cfg.SetDiskID(dev, node)
1561
  if ldisk:
1562
    idx = 6
1563
  else:
1564
    idx = 5
1565

    
1566
  result = True
1567
  if on_primary or dev.AssembleOnSecondary():
1568
    rstats = lu.rpc.call_blockdev_find(node, dev)
1569
    if rstats.failed or not rstats.data:
1570
      logging.warning("Node %s: disk degraded, not found or node down", node)
1571
      result = False
1572
    else:
1573
      result = result and (not rstats.data[idx])
1574
  if dev.children:
1575
    for child in dev.children:
1576
      result = result and _CheckDiskConsistency(lu, child, node, on_primary)
1577

    
1578
  return result
1579

    
1580

    
1581
class LUDiagnoseOS(NoHooksLU):
1582
  """Logical unit for OS diagnose/query.
1583

1584
  """
1585
  _OP_REQP = ["output_fields", "names"]
1586
  REQ_BGL = False
1587
  _FIELDS_STATIC = utils.FieldSet()
1588
  _FIELDS_DYNAMIC = utils.FieldSet("name", "valid", "node_status")
1589

    
1590
  def ExpandNames(self):
1591
    if self.op.names:
1592
      raise errors.OpPrereqError("Selective OS query not supported")
1593

    
1594
    _CheckOutputFields(static=self._FIELDS_STATIC,
1595
                       dynamic=self._FIELDS_DYNAMIC,
1596
                       selected=self.op.output_fields)
1597

    
1598
    # Lock all nodes, in shared mode
1599
    self.needed_locks = {}
1600
    self.share_locks[locking.LEVEL_NODE] = 1
1601
    self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
1602

    
1603
  def CheckPrereq(self):
1604
    """Check prerequisites.
1605

1606
    """
1607

    
1608
  @staticmethod
1609
  def _DiagnoseByOS(node_list, rlist):
1610
    """Remaps a per-node return list into an a per-os per-node dictionary
1611

1612
    @param node_list: a list with the names of all nodes
1613
    @param rlist: a map with node names as keys and OS objects as values
1614

1615
    @rtype: dict
1616
    @returns: a dictionary with osnames as keys and as value another map, with
1617
        nodes as keys and list of OS objects as values, eg::
1618

1619
          {"debian-etch": {"node1": [<object>,...],
1620
                           "node2": [<object>,]}
1621
          }
1622

1623
    """
1624
    all_os = {}
1625
    for node_name, nr in rlist.iteritems():
1626
      if nr.failed or not nr.data:
1627
        continue
1628
      for os_obj in nr.data:
1629
        if os_obj.name not in all_os:
1630
          # build a list of nodes for this os containing empty lists
1631
          # for each node in node_list
1632
          all_os[os_obj.name] = {}
1633
          for nname in node_list:
1634
            all_os[os_obj.name][nname] = []
1635
        all_os[os_obj.name][node_name].append(os_obj)
1636
    return all_os
1637

    
1638
  def Exec(self, feedback_fn):
1639
    """Compute the list of OSes.
1640

1641
    """
1642
    node_list = self.acquired_locks[locking.LEVEL_NODE]
1643
    valid_nodes = [node for node in self.cfg.GetOnlineNodeList()
1644
                   if node in node_list]
1645
    node_data = self.rpc.call_os_diagnose(valid_nodes)
1646
    if node_data == False:
1647
      raise errors.OpExecError("Can't gather the list of OSes")
1648
    pol = self._DiagnoseByOS(valid_nodes, node_data)
1649
    output = []
1650
    for os_name, os_data in pol.iteritems():
1651
      row = []
1652
      for field in self.op.output_fields:
1653
        if field == "name":
1654
          val = os_name
1655
        elif field == "valid":
1656
          val = utils.all([osl and osl[0] for osl in os_data.values()])
1657
        elif field == "node_status":
1658
          val = {}
1659
          for node_name, nos_list in os_data.iteritems():
1660
            val[node_name] = [(v.status, v.path) for v in nos_list]
1661
        else:
1662
          raise errors.ParameterError(field)
1663
        row.append(val)
1664
      output.append(row)
1665

    
1666
    return output
1667

    
1668

    
1669
class LURemoveNode(LogicalUnit):
1670
  """Logical unit for removing a node.
1671

1672
  """
1673
  HPATH = "node-remove"
1674
  HTYPE = constants.HTYPE_NODE
1675
  _OP_REQP = ["node_name"]
1676

    
1677
  def BuildHooksEnv(self):
1678
    """Build hooks env.
1679

1680
    This doesn't run on the target node in the pre phase as a failed
1681
    node would then be impossible to remove.
1682

1683
    """
1684
    env = {
1685
      "OP_TARGET": self.op.node_name,
1686
      "NODE_NAME": self.op.node_name,
1687
      }
1688
    all_nodes = self.cfg.GetNodeList()
1689
    all_nodes.remove(self.op.node_name)
1690
    return env, all_nodes, all_nodes
1691

    
1692
  def CheckPrereq(self):
1693
    """Check prerequisites.
1694

1695
    This checks:
1696
     - the node exists in the configuration
1697
     - it does not have primary or secondary instances
1698
     - it's not the master
1699

1700
    Any errors are signalled by raising errors.OpPrereqError.
1701

1702
    """
1703
    node = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.node_name))
1704
    if node is None:
1705
      raise errors.OpPrereqError, ("Node '%s' is unknown." % self.op.node_name)
1706

    
1707
    instance_list = self.cfg.GetInstanceList()
1708

    
1709
    masternode = self.cfg.GetMasterNode()
1710
    if node.name == masternode:
1711
      raise errors.OpPrereqError("Node is the master node,"
1712
                                 " you need to failover first.")
1713

    
1714
    for instance_name in instance_list:
1715
      instance = self.cfg.GetInstanceInfo(instance_name)
1716
      if node.name in instance.all_nodes:
1717
        raise errors.OpPrereqError("Instance %s is still running on the node,"
1718
                                   " please remove first." % instance_name)
1719
    self.op.node_name = node.name
1720
    self.node = node
1721

    
1722
  def Exec(self, feedback_fn):
1723
    """Removes the node from the cluster.
1724

1725
    """
1726
    node = self.node
1727
    logging.info("Stopping the node daemon and removing configs from node %s",
1728
                 node.name)
1729

    
1730
    self.context.RemoveNode(node.name)
1731

    
1732
    self.rpc.call_node_leave_cluster(node.name)
1733

    
1734
    # Promote nodes to master candidate as needed
1735
    _AdjustCandidatePool(self)
1736

    
1737

    
1738
class LUQueryNodes(NoHooksLU):
1739
  """Logical unit for querying nodes.
1740

1741
  """
1742
  _OP_REQP = ["output_fields", "names"]
1743
  REQ_BGL = False
1744
  _FIELDS_DYNAMIC = utils.FieldSet(
1745
    "dtotal", "dfree",
1746
    "mtotal", "mnode", "mfree",
1747
    "bootid",
1748
    "ctotal",
1749
    )
1750

    
1751
  _FIELDS_STATIC = utils.FieldSet(
1752
    "name", "pinst_cnt", "sinst_cnt",
1753
    "pinst_list", "sinst_list",
1754
    "pip", "sip", "tags",
1755
    "serial_no",
1756
    "master_candidate",
1757
    "master",
1758
    "offline",
1759
    )
1760

    
1761
  def ExpandNames(self):
1762
    _CheckOutputFields(static=self._FIELDS_STATIC,
1763
                       dynamic=self._FIELDS_DYNAMIC,
1764
                       selected=self.op.output_fields)
1765

    
1766
    self.needed_locks = {}
1767
    self.share_locks[locking.LEVEL_NODE] = 1
1768

    
1769
    if self.op.names:
1770
      self.wanted = _GetWantedNodes(self, self.op.names)
1771
    else:
1772
      self.wanted = locking.ALL_SET
1773

    
1774
    self.do_locking = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
1775
    if self.do_locking:
1776
      # if we don't request only static fields, we need to lock the nodes
1777
      self.needed_locks[locking.LEVEL_NODE] = self.wanted
1778

    
1779

    
1780
  def CheckPrereq(self):
1781
    """Check prerequisites.
1782

1783
    """
1784
    # The validation of the node list is done in the _GetWantedNodes,
1785
    # if non empty, and if empty, there's no validation to do
1786
    pass
1787

    
1788
  def Exec(self, feedback_fn):
1789
    """Computes the list of nodes and their attributes.
1790

1791
    """
1792
    all_info = self.cfg.GetAllNodesInfo()
1793
    if self.do_locking:
1794
      nodenames = self.acquired_locks[locking.LEVEL_NODE]
1795
    elif self.wanted != locking.ALL_SET:
1796
      nodenames = self.wanted
1797
      missing = set(nodenames).difference(all_info.keys())
1798
      if missing:
1799
        raise errors.OpExecError(
1800
          "Some nodes were removed before retrieving their data: %s" % missing)
1801
    else:
1802
      nodenames = all_info.keys()
1803

    
1804
    nodenames = utils.NiceSort(nodenames)
1805
    nodelist = [all_info[name] for name in nodenames]
1806

    
1807
    # begin data gathering
1808

    
1809
    if self.do_locking:
1810
      live_data = {}
1811
      node_data = self.rpc.call_node_info(nodenames, self.cfg.GetVGName(),
1812
                                          self.cfg.GetHypervisorType())
1813
      for name in nodenames:
1814
        nodeinfo = node_data[name]
1815
        if not nodeinfo.failed and nodeinfo.data:
1816
          nodeinfo = nodeinfo.data
1817
          fn = utils.TryConvert
1818
          live_data[name] = {
1819
            "mtotal": fn(int, nodeinfo.get('memory_total', None)),
1820
            "mnode": fn(int, nodeinfo.get('memory_dom0', None)),
1821
            "mfree": fn(int, nodeinfo.get('memory_free', None)),
1822
            "dtotal": fn(int, nodeinfo.get('vg_size', None)),
1823
            "dfree": fn(int, nodeinfo.get('vg_free', None)),
1824
            "ctotal": fn(int, nodeinfo.get('cpu_total', None)),
1825
            "bootid": nodeinfo.get('bootid', None),
1826
            }
1827
        else:
1828
          live_data[name] = {}
1829
    else:
1830
      live_data = dict.fromkeys(nodenames, {})
1831

    
1832
    node_to_primary = dict([(name, set()) for name in nodenames])
1833
    node_to_secondary = dict([(name, set()) for name in nodenames])
1834

    
1835
    inst_fields = frozenset(("pinst_cnt", "pinst_list",
1836
                             "sinst_cnt", "sinst_list"))
1837
    if inst_fields & frozenset(self.op.output_fields):
1838
      instancelist = self.cfg.GetInstanceList()
1839

    
1840
      for instance_name in instancelist:
1841
        inst = self.cfg.GetInstanceInfo(instance_name)
1842
        if inst.primary_node in node_to_primary:
1843
          node_to_primary[inst.primary_node].add(inst.name)
1844
        for secnode in inst.secondary_nodes:
1845
          if secnode in node_to_secondary:
1846
            node_to_secondary[secnode].add(inst.name)
1847

    
1848
    master_node = self.cfg.GetMasterNode()
1849

    
1850
    # end data gathering
1851

    
1852
    output = []
1853
    for node in nodelist:
1854
      node_output = []
1855
      for field in self.op.output_fields:
1856
        if field == "name":
1857
          val = node.name
1858
        elif field == "pinst_list":
1859
          val = list(node_to_primary[node.name])
1860
        elif field == "sinst_list":
1861
          val = list(node_to_secondary[node.name])
1862
        elif field == "pinst_cnt":
1863
          val = len(node_to_primary[node.name])
1864
        elif field == "sinst_cnt":
1865
          val = len(node_to_secondary[node.name])
1866
        elif field == "pip":
1867
          val = node.primary_ip
1868
        elif field == "sip":
1869
          val = node.secondary_ip
1870
        elif field == "tags":
1871
          val = list(node.GetTags())
1872
        elif field == "serial_no":
1873
          val = node.serial_no
1874
        elif field == "master_candidate":
1875
          val = node.master_candidate
1876
        elif field == "master":
1877
          val = node.name == master_node
1878
        elif field == "offline":
1879
          val = node.offline
1880
        elif self._FIELDS_DYNAMIC.Matches(field):
1881
          val = live_data[node.name].get(field, None)
1882
        else:
1883
          raise errors.ParameterError(field)
1884
        node_output.append(val)
1885
      output.append(node_output)
1886

    
1887
    return output
1888

    
1889

    
1890
class LUQueryNodeVolumes(NoHooksLU):
1891
  """Logical unit for getting volumes on node(s).
1892

1893
  """
1894
  _OP_REQP = ["nodes", "output_fields"]
1895
  REQ_BGL = False
1896
  _FIELDS_DYNAMIC = utils.FieldSet("phys", "vg", "name", "size", "instance")
1897
  _FIELDS_STATIC = utils.FieldSet("node")
1898

    
1899
  def ExpandNames(self):
1900
    _CheckOutputFields(static=self._FIELDS_STATIC,
1901
                       dynamic=self._FIELDS_DYNAMIC,
1902
                       selected=self.op.output_fields)
1903

    
1904
    self.needed_locks = {}
1905
    self.share_locks[locking.LEVEL_NODE] = 1
1906
    if not self.op.nodes:
1907
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
1908
    else:
1909
      self.needed_locks[locking.LEVEL_NODE] = \
1910
        _GetWantedNodes(self, self.op.nodes)
1911

    
1912
  def CheckPrereq(self):
1913
    """Check prerequisites.
1914

1915
    This checks that the fields required are valid output fields.
1916

1917
    """
1918
    self.nodes = self.acquired_locks[locking.LEVEL_NODE]
1919

    
1920
  def Exec(self, feedback_fn):
1921
    """Computes the list of nodes and their attributes.
1922

1923
    """
1924
    nodenames = self.nodes
1925
    volumes = self.rpc.call_node_volumes(nodenames)
1926

    
1927
    ilist = [self.cfg.GetInstanceInfo(iname) for iname
1928
             in self.cfg.GetInstanceList()]
1929

    
1930
    lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
1931

    
1932
    output = []
1933
    for node in nodenames:
1934
      if node not in volumes or volumes[node].failed or not volumes[node].data:
1935
        continue
1936

    
1937
      node_vols = volumes[node].data[:]
1938
      node_vols.sort(key=lambda vol: vol['dev'])
1939

    
1940
      for vol in node_vols:
1941
        node_output = []
1942
        for field in self.op.output_fields:
1943
          if field == "node":
1944
            val = node
1945
          elif field == "phys":
1946
            val = vol['dev']
1947
          elif field == "vg":
1948
            val = vol['vg']
1949
          elif field == "name":
1950
            val = vol['name']
1951
          elif field == "size":
1952
            val = int(float(vol['size']))
1953
          elif field == "instance":
1954
            for inst in ilist:
1955
              if node not in lv_by_node[inst]:
1956
                continue
1957
              if vol['name'] in lv_by_node[inst][node]:
1958
                val = inst.name
1959
                break
1960
            else:
1961
              val = '-'
1962
          else:
1963
            raise errors.ParameterError(field)
1964
          node_output.append(str(val))
1965

    
1966
        output.append(node_output)
1967

    
1968
    return output
1969

    
1970

    
1971
class LUAddNode(LogicalUnit):
1972
  """Logical unit for adding node to the cluster.
1973

1974
  """
1975
  HPATH = "node-add"
1976
  HTYPE = constants.HTYPE_NODE
1977
  _OP_REQP = ["node_name"]
1978

    
1979
  def BuildHooksEnv(self):
1980
    """Build hooks env.
1981

1982
    This will run on all nodes before, and on all nodes + the new node after.
1983

1984
    """
1985
    env = {
1986
      "OP_TARGET": self.op.node_name,
1987
      "NODE_NAME": self.op.node_name,
1988
      "NODE_PIP": self.op.primary_ip,
1989
      "NODE_SIP": self.op.secondary_ip,
1990
      }
1991
    nodes_0 = self.cfg.GetNodeList()
1992
    nodes_1 = nodes_0 + [self.op.node_name, ]
1993
    return env, nodes_0, nodes_1
1994

    
1995
  def CheckPrereq(self):
1996
    """Check prerequisites.
1997

1998
    This checks:
1999
     - the new node is not already in the config
2000
     - it is resolvable
2001
     - its parameters (single/dual homed) matches the cluster
2002

2003
    Any errors are signalled by raising errors.OpPrereqError.
2004

2005
    """
2006
    node_name = self.op.node_name
2007
    cfg = self.cfg
2008

    
2009
    dns_data = utils.HostInfo(node_name)
2010

    
2011
    node = dns_data.name
2012
    primary_ip = self.op.primary_ip = dns_data.ip
2013
    secondary_ip = getattr(self.op, "secondary_ip", None)
2014
    if secondary_ip is None:
2015
      secondary_ip = primary_ip
2016
    if not utils.IsValidIP(secondary_ip):
2017
      raise errors.OpPrereqError("Invalid secondary IP given")
2018
    self.op.secondary_ip = secondary_ip
2019

    
2020
    node_list = cfg.GetNodeList()
2021
    if not self.op.readd and node in node_list:
2022
      raise errors.OpPrereqError("Node %s is already in the configuration" %
2023
                                 node)
2024
    elif self.op.readd and node not in node_list:
2025
      raise errors.OpPrereqError("Node %s is not in the configuration" % node)
2026

    
2027
    for existing_node_name in node_list:
2028
      existing_node = cfg.GetNodeInfo(existing_node_name)
2029

    
2030
      if self.op.readd and node == existing_node_name:
2031
        if (existing_node.primary_ip != primary_ip or
2032
            existing_node.secondary_ip != secondary_ip):
2033
          raise errors.OpPrereqError("Readded node doesn't have the same IP"
2034
                                     " address configuration as before")
2035
        continue
2036

    
2037
      if (existing_node.primary_ip == primary_ip or
2038
          existing_node.secondary_ip == primary_ip or
2039
          existing_node.primary_ip == secondary_ip or
2040
          existing_node.secondary_ip == secondary_ip):
2041
        raise errors.OpPrereqError("New node ip address(es) conflict with"
2042
                                   " existing node %s" % existing_node.name)
2043

    
2044
    # check that the type of the node (single versus dual homed) is the
2045
    # same as for the master
2046
    myself = cfg.GetNodeInfo(self.cfg.GetMasterNode())
2047
    master_singlehomed = myself.secondary_ip == myself.primary_ip
2048
    newbie_singlehomed = secondary_ip == primary_ip
2049
    if master_singlehomed != newbie_singlehomed:
2050
      if master_singlehomed:
2051
        raise errors.OpPrereqError("The master has no private ip but the"
2052
                                   " new node has one")
2053
      else:
2054
        raise errors.OpPrereqError("The master has a private ip but the"
2055
                                   " new node doesn't have one")
2056

    
2057
    # checks reachablity
2058
    if not utils.TcpPing(primary_ip, constants.DEFAULT_NODED_PORT):
2059
      raise errors.OpPrereqError("Node not reachable by ping")
2060

    
2061
    if not newbie_singlehomed:
2062
      # check reachability from my secondary ip to newbie's secondary ip
2063
      if not utils.TcpPing(secondary_ip, constants.DEFAULT_NODED_PORT,
2064
                           source=myself.secondary_ip):
2065
        raise errors.OpPrereqError("Node secondary ip not reachable by TCP"
2066
                                   " based ping to noded port")
2067

    
2068
    cp_size = self.cfg.GetClusterInfo().candidate_pool_size
2069
    mc_now, _ = self.cfg.GetMasterCandidateStats()
2070
    master_candidate = mc_now < cp_size
2071

    
2072
    self.new_node = objects.Node(name=node,
2073
                                 primary_ip=primary_ip,
2074
                                 secondary_ip=secondary_ip,
2075
                                 master_candidate=master_candidate,
2076
                                 offline=False)
2077

    
2078
  def Exec(self, feedback_fn):
2079
    """Adds the new node to the cluster.
2080

2081
    """
2082
    new_node = self.new_node
2083
    node = new_node.name
2084

    
2085
    # check connectivity
2086
    result = self.rpc.call_version([node])[node]
2087
    result.Raise()
2088
    if result.data:
2089
      if constants.PROTOCOL_VERSION == result.data:
2090
        logging.info("Communication to node %s fine, sw version %s match",
2091
                     node, result.data)
2092
      else:
2093
        raise errors.OpExecError("Version mismatch master version %s,"
2094
                                 " node version %s" %
2095
                                 (constants.PROTOCOL_VERSION, result.data))
2096
    else:
2097
      raise errors.OpExecError("Cannot get version from the new node")
2098

    
2099
    # setup ssh on node
2100
    logging.info("Copy ssh key to node %s", node)
2101
    priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
2102
    keyarray = []
2103
    keyfiles = [constants.SSH_HOST_DSA_PRIV, constants.SSH_HOST_DSA_PUB,
2104
                constants.SSH_HOST_RSA_PRIV, constants.SSH_HOST_RSA_PUB,
2105
                priv_key, pub_key]
2106

    
2107
    for i in keyfiles:
2108
      f = open(i, 'r')
2109
      try:
2110
        keyarray.append(f.read())
2111
      finally:
2112
        f.close()
2113

    
2114
    result = self.rpc.call_node_add(node, keyarray[0], keyarray[1],
2115
                                    keyarray[2],
2116
                                    keyarray[3], keyarray[4], keyarray[5])
2117

    
2118
    if result.failed or not result.data:
2119
      raise errors.OpExecError("Cannot transfer ssh keys to the new node")
2120

    
2121
    # Add node to our /etc/hosts, and add key to known_hosts
2122
    utils.AddHostToEtcHosts(new_node.name)
2123

    
2124
    if new_node.secondary_ip != new_node.primary_ip:
2125
      result = self.rpc.call_node_has_ip_address(new_node.name,
2126
                                                 new_node.secondary_ip)
2127
      if result.failed or not result.data:
2128
        raise errors.OpExecError("Node claims it doesn't have the secondary ip"
2129
                                 " you gave (%s). Please fix and re-run this"
2130
                                 " command." % new_node.secondary_ip)
2131

    
2132
    node_verify_list = [self.cfg.GetMasterNode()]
2133
    node_verify_param = {
2134
      'nodelist': [node],
2135
      # TODO: do a node-net-test as well?
2136
    }
2137

    
2138
    result = self.rpc.call_node_verify(node_verify_list, node_verify_param,
2139
                                       self.cfg.GetClusterName())
2140
    for verifier in node_verify_list:
2141
      if result[verifier].failed or not result[verifier].data:
2142
        raise errors.OpExecError("Cannot communicate with %s's node daemon"
2143
                                 " for remote verification" % verifier)
2144
      if result[verifier].data['nodelist']:
2145
        for failed in result[verifier].data['nodelist']:
2146
          feedback_fn("ssh/hostname verification failed %s -> %s" %
2147
                      (verifier, result[verifier]['nodelist'][failed]))
2148
        raise errors.OpExecError("ssh/hostname verification failed.")
2149

    
2150
    # Distribute updated /etc/hosts and known_hosts to all nodes,
2151
    # including the node just added
2152
    myself = self.cfg.GetNodeInfo(self.cfg.GetMasterNode())
2153
    dist_nodes = self.cfg.GetNodeList()
2154
    if not self.op.readd:
2155
      dist_nodes.append(node)
2156
    if myself.name in dist_nodes:
2157
      dist_nodes.remove(myself.name)
2158

    
2159
    logging.debug("Copying hosts and known_hosts to all nodes")
2160
    for fname in (constants.ETC_HOSTS, constants.SSH_KNOWN_HOSTS_FILE):
2161
      result = self.rpc.call_upload_file(dist_nodes, fname)
2162
      for to_node, to_result in result.iteritems():
2163
        if to_result.failed or not to_result.data:
2164
          logging.error("Copy of file %s to node %s failed", fname, to_node)
2165

    
2166
    to_copy = []
2167
    if constants.HT_XEN_HVM in self.cfg.GetClusterInfo().enabled_hypervisors:
2168
      to_copy.append(constants.VNC_PASSWORD_FILE)
2169
    for fname in to_copy:
2170
      result = self.rpc.call_upload_file([node], fname)
2171
      if result[node].failed or not result[node]:
2172
        logging.error("Could not copy file %s to node %s", fname, node)
2173

    
2174
    if self.op.readd:
2175
      self.context.ReaddNode(new_node)
2176
    else:
2177
      self.context.AddNode(new_node)
2178

    
2179

    
2180
class LUSetNodeParams(LogicalUnit):
2181
  """Modifies the parameters of a node.
2182

2183
  """
2184
  HPATH = "node-modify"
2185
  HTYPE = constants.HTYPE_NODE
2186
  _OP_REQP = ["node_name"]
2187
  REQ_BGL = False
2188

    
2189
  def CheckArguments(self):
2190
    node_name = self.cfg.ExpandNodeName(self.op.node_name)
2191
    if node_name is None:
2192
      raise errors.OpPrereqError("Invalid node name '%s'" % self.op.node_name)
2193
    self.op.node_name = node_name
2194
    _CheckBooleanOpField(self.op, 'master_candidate')
2195
    _CheckBooleanOpField(self.op, 'offline')
2196
    if self.op.master_candidate is None and self.op.offline is None:
2197
      raise errors.OpPrereqError("Please pass at least one modification")
2198
    if self.op.offline == True and self.op.master_candidate == True:
2199
      raise errors.OpPrereqError("Can't set the node into offline and"
2200
                                 " master_candidate at the same time")
2201

    
2202
  def ExpandNames(self):
2203
    self.needed_locks = {locking.LEVEL_NODE: self.op.node_name}
2204

    
2205
  def BuildHooksEnv(self):
2206
    """Build hooks env.
2207

2208
    This runs on the master node.
2209

2210
    """
2211
    env = {
2212
      "OP_TARGET": self.op.node_name,
2213
      "MASTER_CANDIDATE": str(self.op.master_candidate),
2214
      "OFFLINE": str(self.op.offline),
2215
      }
2216
    nl = [self.cfg.GetMasterNode(),
2217
          self.op.node_name]
2218
    return env, nl, nl
2219

    
2220
  def CheckPrereq(self):
2221
    """Check prerequisites.
2222

2223
    This only checks the instance list against the existing names.
2224

2225
    """
2226
    node = self.node = self.cfg.GetNodeInfo(self.op.node_name)
2227

    
2228
    if ((self.op.master_candidate == False or self.op.offline == True)
2229
        and node.master_candidate):
2230
      # we will demote the node from master_candidate
2231
      if self.op.node_name == self.cfg.GetMasterNode():
2232
        raise errors.OpPrereqError("The master node has to be a"
2233
                                   " master candidate and online")
2234
      cp_size = self.cfg.GetClusterInfo().candidate_pool_size
2235
      num_candidates, _ = self.cfg.GetMasterCandidateStats()
2236
      if num_candidates <= cp_size:
2237
        msg = ("Not enough master candidates (desired"
2238
               " %d, new value will be %d)" % (cp_size, num_candidates-1))
2239
        if self.op.force:
2240
          self.LogWarning(msg)
2241
        else:
2242
          raise errors.OpPrereqError(msg)
2243

    
2244
    if (self.op.master_candidate == True and node.offline and
2245
        not self.op.offline == False):
2246
      raise errors.OpPrereqError("Can't set an offline node to"
2247
                                 " master_candidate")
2248

    
2249
    return
2250

    
2251
  def Exec(self, feedback_fn):
2252
    """Modifies a node.
2253

2254
    """
2255
    node = self.node
2256

    
2257
    result = []
2258

    
2259
    if self.op.offline is not None:
2260
      node.offline = self.op.offline
2261
      result.append(("offline", str(self.op.offline)))
2262
      if self.op.offline == True and node.master_candidate:
2263
        node.master_candidate = False
2264
        result.append(("master_candidate", "auto-demotion due to offline"))
2265

    
2266
    if self.op.master_candidate is not None:
2267
      node.master_candidate = self.op.master_candidate
2268
      result.append(("master_candidate", str(self.op.master_candidate)))
2269
      if self.op.master_candidate == False:
2270
        rrc = self.rpc.call_node_demote_from_mc(node.name)
2271
        if (rrc.failed or not isinstance(rrc.data, (tuple, list))
2272
            or len(rrc.data) != 2):
2273
          self.LogWarning("Node rpc error: %s" % rrc.error)
2274
        elif not rrc.data[0]:
2275
          self.LogWarning("Node failed to demote itself: %s" % rrc.data[1])
2276

    
2277
    # this will trigger configuration file update, if needed
2278
    self.cfg.Update(node)
2279
    # this will trigger job queue propagation or cleanup
2280
    if self.op.node_name != self.cfg.GetMasterNode():
2281
      self.context.ReaddNode(node)
2282

    
2283
    return result
2284

    
2285

    
2286
class LUQueryClusterInfo(NoHooksLU):
2287
  """Query cluster configuration.
2288

2289
  """
2290
  _OP_REQP = []
2291
  REQ_BGL = False
2292

    
2293
  def ExpandNames(self):
2294
    self.needed_locks = {}
2295

    
2296
  def CheckPrereq(self):
2297
    """No prerequsites needed for this LU.
2298

2299
    """
2300
    pass
2301

    
2302
  def Exec(self, feedback_fn):
2303
    """Return cluster config.
2304

2305
    """
2306
    cluster = self.cfg.GetClusterInfo()
2307
    result = {
2308
      "software_version": constants.RELEASE_VERSION,
2309
      "protocol_version": constants.PROTOCOL_VERSION,
2310
      "config_version": constants.CONFIG_VERSION,
2311
      "os_api_version": constants.OS_API_VERSION,
2312
      "export_version": constants.EXPORT_VERSION,
2313
      "architecture": (platform.architecture()[0], platform.machine()),
2314
      "name": cluster.cluster_name,
2315
      "master": cluster.master_node,
2316
      "default_hypervisor": cluster.default_hypervisor,
2317
      "enabled_hypervisors": cluster.enabled_hypervisors,
2318
      "hvparams": cluster.hvparams,
2319
      "beparams": cluster.beparams,
2320
      "candidate_pool_size": cluster.candidate_pool_size,
2321
      }
2322

    
2323
    return result
2324

    
2325

    
2326
class LUQueryConfigValues(NoHooksLU):
2327
  """Return configuration values.
2328

2329
  """
2330
  _OP_REQP = []
2331
  REQ_BGL = False
2332
  _FIELDS_DYNAMIC = utils.FieldSet()
2333
  _FIELDS_STATIC = utils.FieldSet("cluster_name", "master_node", "drain_flag")
2334

    
2335
  def ExpandNames(self):
2336
    self.needed_locks = {}
2337

    
2338
    _CheckOutputFields(static=self._FIELDS_STATIC,
2339
                       dynamic=self._FIELDS_DYNAMIC,
2340
                       selected=self.op.output_fields)
2341

    
2342
  def CheckPrereq(self):
2343
    """No prerequisites.
2344

2345
    """
2346
    pass
2347

    
2348
  def Exec(self, feedback_fn):
2349
    """Dump a representation of the cluster config to the standard output.
2350

2351
    """
2352
    values = []
2353
    for field in self.op.output_fields:
2354
      if field == "cluster_name":
2355
        entry = self.cfg.GetClusterName()
2356
      elif field == "master_node":
2357
        entry = self.cfg.GetMasterNode()
2358
      elif field == "drain_flag":
2359
        entry = os.path.exists(constants.JOB_QUEUE_DRAIN_FILE)
2360
      else:
2361
        raise errors.ParameterError(field)
2362
      values.append(entry)
2363
    return values
2364

    
2365

    
2366
class LUActivateInstanceDisks(NoHooksLU):
2367
  """Bring up an instance's disks.
2368

2369
  """
2370
  _OP_REQP = ["instance_name"]
2371
  REQ_BGL = False
2372

    
2373
  def ExpandNames(self):
2374
    self._ExpandAndLockInstance()
2375
    self.needed_locks[locking.LEVEL_NODE] = []
2376
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2377

    
2378
  def DeclareLocks(self, level):
2379
    if level == locking.LEVEL_NODE:
2380
      self._LockInstancesNodes()
2381

    
2382
  def CheckPrereq(self):
2383
    """Check prerequisites.
2384

2385
    This checks that the instance is in the cluster.
2386

2387
    """
2388
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2389
    assert self.instance is not None, \
2390
      "Cannot retrieve locked instance %s" % self.op.instance_name
2391
    _CheckNodeOnline(self, self.instance.primary_node)
2392

    
2393
  def Exec(self, feedback_fn):
2394
    """Activate the disks.
2395

2396
    """
2397
    disks_ok, disks_info = _AssembleInstanceDisks(self, self.instance)
2398
    if not disks_ok:
2399
      raise errors.OpExecError("Cannot activate block devices")
2400

    
2401
    return disks_info
2402

    
2403

    
2404
def _AssembleInstanceDisks(lu, instance, ignore_secondaries=False):
2405
  """Prepare the block devices for an instance.
2406

2407
  This sets up the block devices on all nodes.
2408

2409
  @type lu: L{LogicalUnit}
2410
  @param lu: the logical unit on whose behalf we execute
2411
  @type instance: L{objects.Instance}
2412
  @param instance: the instance for whose disks we assemble
2413
  @type ignore_secondaries: boolean
2414
  @param ignore_secondaries: if true, errors on secondary nodes
2415
      won't result in an error return from the function
2416
  @return: False if the operation failed, otherwise a list of
2417
      (host, instance_visible_name, node_visible_name)
2418
      with the mapping from node devices to instance devices
2419

2420
  """
2421
  device_info = []
2422
  disks_ok = True
2423
  iname = instance.name
2424
  # With the two passes mechanism we try to reduce the window of
2425
  # opportunity for the race condition of switching DRBD to primary
2426
  # before handshaking occured, but we do not eliminate it
2427

    
2428
  # The proper fix would be to wait (with some limits) until the
2429
  # connection has been made and drbd transitions from WFConnection
2430
  # into any other network-connected state (Connected, SyncTarget,
2431
  # SyncSource, etc.)
2432

    
2433
  # 1st pass, assemble on all nodes in secondary mode
2434
  for inst_disk in instance.disks:
2435
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
2436
      lu.cfg.SetDiskID(node_disk, node)
2437
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, False)
2438
      if result.failed or not result:
2439
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
2440
                           " (is_primary=False, pass=1)",
2441
                           inst_disk.iv_name, node)
2442
        if not ignore_secondaries:
2443
          disks_ok = False
2444

    
2445
  # FIXME: race condition on drbd migration to primary
2446

    
2447
  # 2nd pass, do only the primary node
2448
  for inst_disk in instance.disks:
2449
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
2450
      if node != instance.primary_node:
2451
        continue
2452
      lu.cfg.SetDiskID(node_disk, node)
2453
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, True)
2454
      if result.failed or not result:
2455
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
2456
                           " (is_primary=True, pass=2)",
2457
                           inst_disk.iv_name, node)
2458
        disks_ok = False
2459
    device_info.append((instance.primary_node, inst_disk.iv_name, result.data))
2460

    
2461
  # leave the disks configured for the primary node
2462
  # this is a workaround that would be fixed better by
2463
  # improving the logical/physical id handling
2464
  for disk in instance.disks:
2465
    lu.cfg.SetDiskID(disk, instance.primary_node)
2466

    
2467
  return disks_ok, device_info
2468

    
2469

    
2470
def _StartInstanceDisks(lu, instance, force):
2471
  """Start the disks of an instance.
2472

2473
  """
2474
  disks_ok, dummy = _AssembleInstanceDisks(lu, instance,
2475
                                           ignore_secondaries=force)
2476
  if not disks_ok:
2477
    _ShutdownInstanceDisks(lu, instance)
2478
    if force is not None and not force:
2479
      lu.proc.LogWarning("", hint="If the message above refers to a"
2480
                         " secondary node,"
2481
                         " you can retry the operation using '--force'.")
2482
    raise errors.OpExecError("Disk consistency error")
2483

    
2484

    
2485
class LUDeactivateInstanceDisks(NoHooksLU):
2486
  """Shutdown an instance's disks.
2487

2488
  """
2489
  _OP_REQP = ["instance_name"]
2490
  REQ_BGL = False
2491

    
2492
  def ExpandNames(self):
2493
    self._ExpandAndLockInstance()
2494
    self.needed_locks[locking.LEVEL_NODE] = []
2495
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2496

    
2497
  def DeclareLocks(self, level):
2498
    if level == locking.LEVEL_NODE:
2499
      self._LockInstancesNodes()
2500

    
2501
  def CheckPrereq(self):
2502
    """Check prerequisites.
2503

2504
    This checks that the instance is in the cluster.
2505

2506
    """
2507
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2508
    assert self.instance is not None, \
2509
      "Cannot retrieve locked instance %s" % self.op.instance_name
2510

    
2511
  def Exec(self, feedback_fn):
2512
    """Deactivate the disks
2513

2514
    """
2515
    instance = self.instance
2516
    _SafeShutdownInstanceDisks(self, instance)
2517

    
2518

    
2519
def _SafeShutdownInstanceDisks(lu, instance):
2520
  """Shutdown block devices of an instance.
2521

2522
  This function checks if an instance is running, before calling
2523
  _ShutdownInstanceDisks.
2524

2525
  """
2526
  ins_l = lu.rpc.call_instance_list([instance.primary_node],
2527
                                      [instance.hypervisor])
2528
  ins_l = ins_l[instance.primary_node]
2529
  if ins_l.failed or not isinstance(ins_l.data, list):
2530
    raise errors.OpExecError("Can't contact node '%s'" %
2531
                             instance.primary_node)
2532

    
2533
  if instance.name in ins_l.data:
2534
    raise errors.OpExecError("Instance is running, can't shutdown"
2535
                             " block devices.")
2536

    
2537
  _ShutdownInstanceDisks(lu, instance)
2538

    
2539

    
2540
def _ShutdownInstanceDisks(lu, instance, ignore_primary=False):
2541
  """Shutdown block devices of an instance.
2542

2543
  This does the shutdown on all nodes of the instance.
2544

2545
  If the ignore_primary is false, errors on the primary node are
2546
  ignored.
2547

2548
  """
2549
  result = True
2550
  for disk in instance.disks:
2551
    for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
2552
      lu.cfg.SetDiskID(top_disk, node)
2553
      result = lu.rpc.call_blockdev_shutdown(node, top_disk)
2554
      if result.failed or not result.data:
2555
        logging.error("Could not shutdown block device %s on node %s",
2556
                      disk.iv_name, node)
2557
        if not ignore_primary or node != instance.primary_node:
2558
          result = False
2559
  return result
2560

    
2561

    
2562
def _CheckNodeFreeMemory(lu, node, reason, requested, hypervisor_name):
2563
  """Checks if a node has enough free memory.
2564

2565
  This function check if a given node has the needed amount of free
2566
  memory. In case the node has less memory or we cannot get the
2567
  information from the node, this function raise an OpPrereqError
2568
  exception.
2569

2570
  @type lu: C{LogicalUnit}
2571
  @param lu: a logical unit from which we get configuration data
2572
  @type node: C{str}
2573
  @param node: the node to check
2574
  @type reason: C{str}
2575
  @param reason: string to use in the error message
2576
  @type requested: C{int}
2577
  @param requested: the amount of memory in MiB to check for
2578
  @type hypervisor_name: C{str}
2579
  @param hypervisor_name: the hypervisor to ask for memory stats
2580
  @raise errors.OpPrereqError: if the node doesn't have enough memory, or
2581
      we cannot check the node
2582

2583
  """
2584
  nodeinfo = lu.rpc.call_node_info([node], lu.cfg.GetVGName(), hypervisor_name)
2585
  nodeinfo[node].Raise()
2586
  free_mem = nodeinfo[node].data.get('memory_free')
2587
  if not isinstance(free_mem, int):
2588
    raise errors.OpPrereqError("Can't compute free memory on node %s, result"
2589
                             " was '%s'" % (node, free_mem))
2590
  if requested > free_mem:
2591
    raise errors.OpPrereqError("Not enough memory on node %s for %s:"
2592
                             " needed %s MiB, available %s MiB" %
2593
                             (node, reason, requested, free_mem))
2594

    
2595

    
2596
class LUStartupInstance(LogicalUnit):
2597
  """Starts an instance.
2598

2599
  """
2600
  HPATH = "instance-start"
2601
  HTYPE = constants.HTYPE_INSTANCE
2602
  _OP_REQP = ["instance_name", "force"]
2603
  REQ_BGL = False
2604

    
2605
  def ExpandNames(self):
2606
    self._ExpandAndLockInstance()
2607

    
2608
  def BuildHooksEnv(self):
2609
    """Build hooks env.
2610

2611
    This runs on master, primary and secondary nodes of the instance.
2612

2613
    """
2614
    env = {
2615
      "FORCE": self.op.force,
2616
      }
2617
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
2618
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2619
    return env, nl, nl
2620

    
2621
  def CheckPrereq(self):
2622
    """Check prerequisites.
2623

2624
    This checks that the instance is in the cluster.
2625

2626
    """
2627
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2628
    assert self.instance is not None, \
2629
      "Cannot retrieve locked instance %s" % self.op.instance_name
2630

    
2631
    _CheckNodeOnline(self, instance.primary_node)
2632

    
2633
    bep = self.cfg.GetClusterInfo().FillBE(instance)
2634
    # check bridges existance
2635
    _CheckInstanceBridgesExist(self, instance)
2636

    
2637
    _CheckNodeFreeMemory(self, instance.primary_node,
2638
                         "starting instance %s" % instance.name,
2639
                         bep[constants.BE_MEMORY], instance.hypervisor)
2640

    
2641
  def Exec(self, feedback_fn):
2642
    """Start the instance.
2643

2644
    """
2645
    instance = self.instance
2646
    force = self.op.force
2647
    extra_args = getattr(self.op, "extra_args", "")
2648

    
2649
    self.cfg.MarkInstanceUp(instance.name)
2650

    
2651
    node_current = instance.primary_node
2652

    
2653
    _StartInstanceDisks(self, instance, force)
2654

    
2655
    result = self.rpc.call_instance_start(node_current, instance, extra_args)
2656
    msg = result.RemoteFailMsg()
2657
    if msg:
2658
      _ShutdownInstanceDisks(self, instance)
2659
      raise errors.OpExecError("Could not start instance: %s" % msg)
2660

    
2661

    
2662
class LURebootInstance(LogicalUnit):
2663
  """Reboot an instance.
2664

2665
  """
2666
  HPATH = "instance-reboot"
2667
  HTYPE = constants.HTYPE_INSTANCE
2668
  _OP_REQP = ["instance_name", "ignore_secondaries", "reboot_type"]
2669
  REQ_BGL = False
2670

    
2671
  def ExpandNames(self):
2672
    if self.op.reboot_type not in [constants.INSTANCE_REBOOT_SOFT,
2673
                                   constants.INSTANCE_REBOOT_HARD,
2674
                                   constants.INSTANCE_REBOOT_FULL]:
2675
      raise errors.ParameterError("reboot type not in [%s, %s, %s]" %
2676
                                  (constants.INSTANCE_REBOOT_SOFT,
2677
                                   constants.INSTANCE_REBOOT_HARD,
2678
                                   constants.INSTANCE_REBOOT_FULL))
2679
    self._ExpandAndLockInstance()
2680

    
2681
  def BuildHooksEnv(self):
2682
    """Build hooks env.
2683

2684
    This runs on master, primary and secondary nodes of the instance.
2685

2686
    """
2687
    env = {
2688
      "IGNORE_SECONDARIES": self.op.ignore_secondaries,
2689
      }
2690
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
2691
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2692
    return env, nl, nl
2693

    
2694
  def CheckPrereq(self):
2695
    """Check prerequisites.
2696

2697
    This checks that the instance is in the cluster.
2698

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

    
2704
    _CheckNodeOnline(self, instance.primary_node)
2705

    
2706
    # check bridges existance
2707
    _CheckInstanceBridgesExist(self, instance)
2708

    
2709
  def Exec(self, feedback_fn):
2710
    """Reboot the instance.
2711

2712
    """
2713
    instance = self.instance
2714
    ignore_secondaries = self.op.ignore_secondaries
2715
    reboot_type = self.op.reboot_type
2716
    extra_args = getattr(self.op, "extra_args", "")
2717

    
2718
    node_current = instance.primary_node
2719

    
2720
    if reboot_type in [constants.INSTANCE_REBOOT_SOFT,
2721
                       constants.INSTANCE_REBOOT_HARD]:
2722
      result = self.rpc.call_instance_reboot(node_current, instance,
2723
                                             reboot_type, extra_args)
2724
      if result.failed or not result.data:
2725
        raise errors.OpExecError("Could not reboot instance")
2726
    else:
2727
      if not self.rpc.call_instance_shutdown(node_current, instance):
2728
        raise errors.OpExecError("could not shutdown instance for full reboot")
2729
      _ShutdownInstanceDisks(self, instance)
2730
      _StartInstanceDisks(self, instance, ignore_secondaries)
2731
      result = self.rpc.call_instance_start(node_current, instance, extra_args)
2732
      msg = result.RemoteFailMsg()
2733
      if msg:
2734
        _ShutdownInstanceDisks(self, instance)
2735
        raise errors.OpExecError("Could not start instance for"
2736
                                 " full reboot: %s" % msg)
2737

    
2738
    self.cfg.MarkInstanceUp(instance.name)
2739

    
2740

    
2741
class LUShutdownInstance(LogicalUnit):
2742
  """Shutdown an instance.
2743

2744
  """
2745
  HPATH = "instance-stop"
2746
  HTYPE = constants.HTYPE_INSTANCE
2747
  _OP_REQP = ["instance_name"]
2748
  REQ_BGL = False
2749

    
2750
  def ExpandNames(self):
2751
    self._ExpandAndLockInstance()
2752

    
2753
  def BuildHooksEnv(self):
2754
    """Build hooks env.
2755

2756
    This runs on master, primary and secondary nodes of the instance.
2757

2758
    """
2759
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2760
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2761
    return env, nl, nl
2762

    
2763
  def CheckPrereq(self):
2764
    """Check prerequisites.
2765

2766
    This checks that the instance is in the cluster.
2767

2768
    """
2769
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2770
    assert self.instance is not None, \
2771
      "Cannot retrieve locked instance %s" % self.op.instance_name
2772
    _CheckNodeOnline(self, self.instance.primary_node)
2773

    
2774
  def Exec(self, feedback_fn):
2775
    """Shutdown the instance.
2776

2777
    """
2778
    instance = self.instance
2779
    node_current = instance.primary_node
2780
    self.cfg.MarkInstanceDown(instance.name)
2781
    result = self.rpc.call_instance_shutdown(node_current, instance)
2782
    if result.failed or not result.data:
2783
      self.proc.LogWarning("Could not shutdown instance")
2784

    
2785
    _ShutdownInstanceDisks(self, instance)
2786

    
2787

    
2788
class LUReinstallInstance(LogicalUnit):
2789
  """Reinstall an instance.
2790

2791
  """
2792
  HPATH = "instance-reinstall"
2793
  HTYPE = constants.HTYPE_INSTANCE
2794
  _OP_REQP = ["instance_name"]
2795
  REQ_BGL = False
2796

    
2797
  def ExpandNames(self):
2798
    self._ExpandAndLockInstance()
2799

    
2800
  def BuildHooksEnv(self):
2801
    """Build hooks env.
2802

2803
    This runs on master, primary and secondary nodes of the instance.
2804

2805
    """
2806
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2807
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2808
    return env, nl, nl
2809

    
2810
  def CheckPrereq(self):
2811
    """Check prerequisites.
2812

2813
    This checks that the instance is in the cluster and is not running.
2814

2815
    """
2816
    instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2817
    assert instance is not None, \
2818
      "Cannot retrieve locked instance %s" % self.op.instance_name
2819
    _CheckNodeOnline(self, instance.primary_node)
2820

    
2821
    if instance.disk_template == constants.DT_DISKLESS:
2822
      raise errors.OpPrereqError("Instance '%s' has no disks" %
2823
                                 self.op.instance_name)
2824
    if instance.status != "down":
2825
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2826
                                 self.op.instance_name)
2827
    remote_info = self.rpc.call_instance_info(instance.primary_node,
2828
                                              instance.name,
2829
                                              instance.hypervisor)
2830
    if remote_info.failed or remote_info.data:
2831
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
2832
                                 (self.op.instance_name,
2833
                                  instance.primary_node))
2834

    
2835
    self.op.os_type = getattr(self.op, "os_type", None)
2836
    if self.op.os_type is not None:
2837
      # OS verification
2838
      pnode = self.cfg.GetNodeInfo(
2839
        self.cfg.ExpandNodeName(instance.primary_node))
2840
      if pnode is None:
2841
        raise errors.OpPrereqError("Primary node '%s' is unknown" %
2842
                                   self.op.pnode)
2843
      result = self.rpc.call_os_get(pnode.name, self.op.os_type)
2844
      result.Raise()
2845
      if not isinstance(result.data, objects.OS):
2846
        raise errors.OpPrereqError("OS '%s' not in supported OS list for"
2847
                                   " primary node"  % self.op.os_type)
2848

    
2849
    self.instance = instance
2850

    
2851
  def Exec(self, feedback_fn):
2852
    """Reinstall the instance.
2853

2854
    """
2855
    inst = self.instance
2856

    
2857
    if self.op.os_type is not None:
2858
      feedback_fn("Changing OS to '%s'..." % self.op.os_type)
2859
      inst.os = self.op.os_type
2860
      self.cfg.Update(inst)
2861

    
2862
    _StartInstanceDisks(self, inst, None)
2863
    try:
2864
      feedback_fn("Running the instance OS create scripts...")
2865
      result = self.rpc.call_instance_os_add(inst.primary_node, inst)
2866
      msg = result.RemoteFailMsg()
2867
      if msg:
2868
        raise errors.OpExecError("Could not install OS for instance %s"
2869
                                 " on node %s: %s" %
2870
                                 (inst.name, inst.primary_node, msg))
2871
    finally:
2872
      _ShutdownInstanceDisks(self, inst)
2873

    
2874

    
2875
class LURenameInstance(LogicalUnit):
2876
  """Rename an instance.
2877

2878
  """
2879
  HPATH = "instance-rename"
2880
  HTYPE = constants.HTYPE_INSTANCE
2881
  _OP_REQP = ["instance_name", "new_name"]
2882

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

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

2888
    """
2889
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2890
    env["INSTANCE_NEW_NAME"] = self.op.new_name
2891
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2892
    return env, nl, nl
2893

    
2894
  def CheckPrereq(self):
2895
    """Check prerequisites.
2896

2897
    This checks that the instance is in the cluster and is not running.
2898

2899
    """
2900
    instance = self.cfg.GetInstanceInfo(
2901
      self.cfg.ExpandInstanceName(self.op.instance_name))
2902
    if instance is None:
2903
      raise errors.OpPrereqError("Instance '%s' not known" %
2904
                                 self.op.instance_name)
2905
    _CheckNodeOnline(self, instance.primary_node)
2906

    
2907
    if instance.status != "down":
2908
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2909
                                 self.op.instance_name)
2910
    remote_info = self.rpc.call_instance_info(instance.primary_node,
2911
                                              instance.name,
2912
                                              instance.hypervisor)
2913
    remote_info.Raise()
2914
    if remote_info.data:
2915
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
2916
                                 (self.op.instance_name,
2917
                                  instance.primary_node))
2918
    self.instance = instance
2919

    
2920
    # new name verification
2921
    name_info = utils.HostInfo(self.op.new_name)
2922

    
2923
    self.op.new_name = new_name = name_info.name
2924
    instance_list = self.cfg.GetInstanceList()
2925
    if new_name in instance_list:
2926
      raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
2927
                                 new_name)
2928

    
2929
    if not getattr(self.op, "ignore_ip", False):
2930
      if utils.TcpPing(name_info.ip, constants.DEFAULT_NODED_PORT):
2931
        raise errors.OpPrereqError("IP %s of instance %s already in use" %
2932
                                   (name_info.ip, new_name))
2933

    
2934

    
2935
  def Exec(self, feedback_fn):
2936
    """Reinstall the instance.
2937

2938
    """
2939
    inst = self.instance
2940
    old_name = inst.name
2941

    
2942
    if inst.disk_template == constants.DT_FILE:
2943
      old_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
2944

    
2945
    self.cfg.RenameInstance(inst.name, self.op.new_name)
2946
    # Change the instance lock. This is definitely safe while we hold the BGL
2947
    self.context.glm.remove(locking.LEVEL_INSTANCE, old_name)
2948
    self.context.glm.add(locking.LEVEL_INSTANCE, self.op.new_name)
2949

    
2950
    # re-read the instance from the configuration after rename
2951
    inst = self.cfg.GetInstanceInfo(self.op.new_name)
2952

    
2953
    if inst.disk_template == constants.DT_FILE:
2954
      new_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
2955
      result = self.rpc.call_file_storage_dir_rename(inst.primary_node,
2956
                                                     old_file_storage_dir,
2957
                                                     new_file_storage_dir)
2958
      result.Raise()
2959
      if not result.data:
2960
        raise errors.OpExecError("Could not connect to node '%s' to rename"
2961
                                 " directory '%s' to '%s' (but the instance"
2962
                                 " has been renamed in Ganeti)" % (
2963
                                 inst.primary_node, old_file_storage_dir,
2964
                                 new_file_storage_dir))
2965

    
2966
      if not result.data[0]:
2967
        raise errors.OpExecError("Could not rename directory '%s' to '%s'"
2968
                                 " (but the instance has been renamed in"
2969
                                 " Ganeti)" % (old_file_storage_dir,
2970
                                               new_file_storage_dir))
2971

    
2972
    _StartInstanceDisks(self, inst, None)
2973
    try:
2974
      result = self.rpc.call_instance_run_rename(inst.primary_node, inst,
2975
                                                 old_name)
2976
      msg = result.RemoteFailMsg()
2977
      if msg:
2978
        msg = ("Could not run OS rename script for instance %s on node %s"
2979
               " (but the instance has been renamed in Ganeti): %s" %
2980
               (inst.name, inst.primary_node, msg))
2981
        self.proc.LogWarning(msg)
2982
    finally:
2983
      _ShutdownInstanceDisks(self, inst)
2984

    
2985

    
2986
class LURemoveInstance(LogicalUnit):
2987
  """Remove an instance.
2988

2989
  """
2990
  HPATH = "instance-remove"
2991
  HTYPE = constants.HTYPE_INSTANCE
2992
  _OP_REQP = ["instance_name", "ignore_failures"]
2993
  REQ_BGL = False
2994

    
2995
  def ExpandNames(self):
2996
    self._ExpandAndLockInstance()
2997
    self.needed_locks[locking.LEVEL_NODE] = []
2998
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2999

    
3000
  def DeclareLocks(self, level):
3001
    if level == locking.LEVEL_NODE:
3002
      self._LockInstancesNodes()
3003

    
3004
  def BuildHooksEnv(self):
3005
    """Build hooks env.
3006

3007
    This runs on master, primary and secondary nodes of the instance.
3008

3009
    """
3010
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3011
    nl = [self.cfg.GetMasterNode()]
3012
    return env, nl, nl
3013

    
3014
  def CheckPrereq(self):
3015
    """Check prerequisites.
3016

3017
    This checks that the instance is in the cluster.
3018

3019
    """
3020
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3021
    assert self.instance is not None, \
3022
      "Cannot retrieve locked instance %s" % self.op.instance_name
3023

    
3024
  def Exec(self, feedback_fn):
3025
    """Remove the instance.
3026

3027
    """
3028
    instance = self.instance
3029
    logging.info("Shutting down instance %s on node %s",
3030
                 instance.name, instance.primary_node)
3031

    
3032
    result = self.rpc.call_instance_shutdown(instance.primary_node, instance)
3033
    if result.failed or not result.data:
3034
      if self.op.ignore_failures:
3035
        feedback_fn("Warning: can't shutdown instance")
3036
      else:
3037
        raise errors.OpExecError("Could not shutdown instance %s on node %s" %
3038
                                 (instance.name, instance.primary_node))
3039

    
3040
    logging.info("Removing block devices for instance %s", instance.name)
3041

    
3042
    if not _RemoveDisks(self, instance):
3043
      if self.op.ignore_failures:
3044
        feedback_fn("Warning: can't remove instance's disks")
3045
      else:
3046
        raise errors.OpExecError("Can't remove instance's disks")
3047

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

    
3050
    self.cfg.RemoveInstance(instance.name)
3051
    self.remove_locks[locking.LEVEL_INSTANCE] = instance.name
3052

    
3053

    
3054
class LUQueryInstances(NoHooksLU):
3055
  """Logical unit for querying instances.
3056

3057
  """
3058
  _OP_REQP = ["output_fields", "names"]
3059
  REQ_BGL = False
3060
  _FIELDS_STATIC = utils.FieldSet(*["name", "os", "pnode", "snodes",
3061
                                    "admin_state", "admin_ram",
3062
                                    "disk_template", "ip", "mac", "bridge",
3063
                                    "sda_size", "sdb_size", "vcpus", "tags",
3064
                                    "network_port", "beparams",
3065
                                    "(disk).(size)/([0-9]+)",
3066
                                    "(disk).(sizes)",
3067
                                    "(nic).(mac|ip|bridge)/([0-9]+)",
3068
                                    "(nic).(macs|ips|bridges)",
3069
                                    "(disk|nic).(count)",
3070
                                    "serial_no", "hypervisor", "hvparams",] +
3071
                                  ["hv/%s" % name
3072
                                   for name in constants.HVS_PARAMETERS] +
3073
                                  ["be/%s" % name
3074
                                   for name in constants.BES_PARAMETERS])
3075
  _FIELDS_DYNAMIC = utils.FieldSet("oper_state", "oper_ram", "status")
3076

    
3077

    
3078
  def ExpandNames(self):
3079
    _CheckOutputFields(static=self._FIELDS_STATIC,
3080
                       dynamic=self._FIELDS_DYNAMIC,
3081
                       selected=self.op.output_fields)
3082

    
3083
    self.needed_locks = {}
3084
    self.share_locks[locking.LEVEL_INSTANCE] = 1
3085
    self.share_locks[locking.LEVEL_NODE] = 1
3086

    
3087
    if self.op.names:
3088
      self.wanted = _GetWantedInstances(self, self.op.names)
3089
    else:
3090
      self.wanted = locking.ALL_SET
3091

    
3092
    self.do_locking = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
3093
    if self.do_locking:
3094
      self.needed_locks[locking.LEVEL_INSTANCE] = self.wanted
3095
      self.needed_locks[locking.LEVEL_NODE] = []
3096
      self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3097

    
3098
  def DeclareLocks(self, level):
3099
    if level == locking.LEVEL_NODE and self.do_locking:
3100
      self._LockInstancesNodes()
3101

    
3102
  def CheckPrereq(self):
3103
    """Check prerequisites.
3104

3105
    """
3106
    pass
3107

    
3108
  def Exec(self, feedback_fn):
3109
    """Computes the list of nodes and their attributes.
3110

3111
    """
3112
    all_info = self.cfg.GetAllInstancesInfo()
3113
    if self.do_locking:
3114
      instance_names = self.acquired_locks[locking.LEVEL_INSTANCE]
3115
    elif self.wanted != locking.ALL_SET:
3116
      instance_names = self.wanted
3117
      missing = set(instance_names).difference(all_info.keys())
3118
      if missing:
3119
        raise errors.OpExecError(
3120
          "Some instances were removed before retrieving their data: %s"
3121
          % missing)
3122
    else:
3123
      instance_names = all_info.keys()
3124

    
3125
    instance_names = utils.NiceSort(instance_names)
3126
    instance_list = [all_info[iname] for iname in instance_names]
3127

    
3128
    # begin data gathering
3129

    
3130
    nodes = frozenset([inst.primary_node for inst in instance_list])
3131
    hv_list = list(set([inst.hypervisor for inst in instance_list]))
3132

    
3133
    bad_nodes = []
3134
    off_nodes = []
3135
    if self.do_locking:
3136
      live_data = {}
3137
      node_data = self.rpc.call_all_instances_info(nodes, hv_list)
3138
      for name in nodes:
3139
        result = node_data[name]
3140
        if result.offline:
3141
          # offline nodes will be in both lists
3142
          off_nodes.append(name)
3143
        if result.failed:
3144
          bad_nodes.append(name)
3145
        else:
3146
          if result.data:
3147
            live_data.update(result.data)
3148
            # else no instance is alive
3149
    else:
3150
      live_data = dict([(name, {}) for name in instance_names])
3151

    
3152
    # end data gathering
3153

    
3154
    HVPREFIX = "hv/"
3155
    BEPREFIX = "be/"
3156
    output = []
3157
    for instance in instance_list:
3158
      iout = []
3159
      i_hv = self.cfg.GetClusterInfo().FillHV(instance)
3160
      i_be = self.cfg.GetClusterInfo().FillBE(instance)
3161
      for field in self.op.output_fields:
3162
        st_match = self._FIELDS_STATIC.Matches(field)
3163
        if field == "name":
3164
          val = instance.name
3165
        elif field == "os":
3166
          val = instance.os
3167
        elif field == "pnode":
3168
          val = instance.primary_node
3169
        elif field == "snodes":
3170
          val = list(instance.secondary_nodes)
3171
        elif field == "admin_state":
3172
          val = (instance.status != "down")
3173
        elif field == "oper_state":
3174
          if instance.primary_node in bad_nodes:
3175
            val = None
3176
          else:
3177
            val = bool(live_data.get(instance.name))
3178
        elif field == "status":
3179
          if instance.primary_node in off_nodes:
3180
            val = "ERROR_nodeoffline"
3181
          elif instance.primary_node in bad_nodes:
3182
            val = "ERROR_nodedown"
3183
          else:
3184
            running = bool(live_data.get(instance.name))
3185
            if running:
3186
              if instance.status != "down":
3187
                val = "running"
3188
              else:
3189
                val = "ERROR_up"
3190
            else:
3191
              if instance.status != "down":
3192
                val = "ERROR_down"
3193
              else:
3194
                val = "ADMIN_down"
3195
        elif field == "oper_ram":
3196
          if instance.primary_node in bad_nodes:
3197
            val = None
3198
          elif instance.name in live_data:
3199
            val = live_data[instance.name].get("memory", "?")
3200
          else:
3201
            val = "-"
3202
        elif field == "disk_template":
3203
          val = instance.disk_template
3204
        elif field == "ip":
3205
          val = instance.nics[0].ip
3206
        elif field == "bridge":
3207
          val = instance.nics[0].bridge
3208
        elif field == "mac":
3209
          val = instance.nics[0].mac
3210
        elif field == "sda_size" or field == "sdb_size":
3211
          idx = ord(field[2]) - ord('a')
3212
          try:
3213
            val = instance.FindDisk(idx).size
3214
          except errors.OpPrereqError:
3215
            val = None
3216
        elif field == "tags":
3217
          val = list(instance.GetTags())
3218
        elif field == "serial_no":
3219
          val = instance.serial_no
3220
        elif field == "network_port":
3221
          val = instance.network_port
3222
        elif field == "hypervisor":
3223
          val = instance.hypervisor
3224
        elif field == "hvparams":
3225
          val = i_hv
3226
        elif (field.startswith(HVPREFIX) and
3227
              field[len(HVPREFIX):] in constants.HVS_PARAMETERS):
3228
          val = i_hv.get(field[len(HVPREFIX):], None)
3229
        elif field == "beparams":
3230
          val = i_be
3231
        elif (field.startswith(BEPREFIX) and
3232
              field[len(BEPREFIX):] in constants.BES_PARAMETERS):
3233
          val = i_be.get(field[len(BEPREFIX):], None)
3234
        elif st_match and st_match.groups():
3235
          # matches a variable list
3236
          st_groups = st_match.groups()
3237
          if st_groups and st_groups[0] == "disk":
3238
            if st_groups[1] == "count":
3239
              val = len(instance.disks)
3240
            elif st_groups[1] == "sizes":
3241
              val = [disk.size for disk in instance.disks]
3242
            elif st_groups[1] == "size":
3243
              try:
3244
                val = instance.FindDisk(st_groups[2]).size
3245
              except errors.OpPrereqError:
3246
                val = None
3247
            else:
3248
              assert False, "Unhandled disk parameter"
3249
          elif st_groups[0] == "nic":
3250
            if st_groups[1] == "count":
3251
              val = len(instance.nics)
3252
            elif st_groups[1] == "macs":
3253
              val = [nic.mac for nic in instance.nics]
3254
            elif st_groups[1] == "ips":
3255
              val = [nic.ip for nic in instance.nics]
3256
            elif st_groups[1] == "bridges":
3257
              val = [nic.bridge for nic in instance.nics]
3258
            else:
3259
              # index-based item
3260
              nic_idx = int(st_groups[2])
3261
              if nic_idx >= len(instance.nics):
3262
                val = None
3263
              else:
3264
                if st_groups[1] == "mac":
3265
                  val = instance.nics[nic_idx].mac
3266
                elif st_groups[1] == "ip":
3267
                  val = instance.nics[nic_idx].ip
3268
                elif st_groups[1] == "bridge":
3269
                  val = instance.nics[nic_idx].bridge
3270
                else:
3271
                  assert False, "Unhandled NIC parameter"
3272
          else:
3273
            assert False, "Unhandled variable parameter"
3274
        else:
3275
          raise errors.ParameterError(field)
3276
        iout.append(val)
3277
      output.append(iout)
3278

    
3279
    return output
3280

    
3281

    
3282
class LUFailoverInstance(LogicalUnit):
3283
  """Failover an instance.
3284

3285
  """
3286
  HPATH = "instance-failover"
3287
  HTYPE = constants.HTYPE_INSTANCE
3288
  _OP_REQP = ["instance_name", "ignore_consistency"]
3289
  REQ_BGL = False
3290

    
3291
  def ExpandNames(self):
3292
    self._ExpandAndLockInstance()
3293
    self.needed_locks[locking.LEVEL_NODE] = []
3294
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3295

    
3296
  def DeclareLocks(self, level):
3297
    if level == locking.LEVEL_NODE:
3298
      self._LockInstancesNodes()
3299

    
3300
  def BuildHooksEnv(self):
3301
    """Build hooks env.
3302

3303
    This runs on master, primary and secondary nodes of the instance.
3304

3305
    """
3306
    env = {
3307
      "IGNORE_CONSISTENCY": self.op.ignore_consistency,
3308
      }
3309
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3310
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3311
    return env, nl, nl
3312

    
3313
  def CheckPrereq(self):
3314
    """Check prerequisites.
3315

3316
    This checks that the instance is in the cluster.
3317

3318
    """
3319
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3320
    assert self.instance is not None, \
3321
      "Cannot retrieve locked instance %s" % self.op.instance_name
3322

    
3323
    bep = self.cfg.GetClusterInfo().FillBE(instance)
3324
    if instance.disk_template not in constants.DTS_NET_MIRROR:
3325
      raise errors.OpPrereqError("Instance's disk layout is not"
3326
                                 " network mirrored, cannot failover.")
3327

    
3328
    secondary_nodes = instance.secondary_nodes
3329
    if not secondary_nodes:
3330
      raise errors.ProgrammerError("no secondary node but using "
3331
                                   "a mirrored disk template")
3332

    
3333
    target_node = secondary_nodes[0]
3334
    _CheckNodeOnline(self, target_node)
3335
    # check memory requirements on the secondary node
3336
    _CheckNodeFreeMemory(self, target_node, "failing over instance %s" %
3337
                         instance.name, bep[constants.BE_MEMORY],
3338
                         instance.hypervisor)
3339

    
3340
    # check bridge existance
3341
    brlist = [nic.bridge for nic in instance.nics]
3342
    result = self.rpc.call_bridges_exist(target_node, brlist)
3343
    result.Raise()
3344
    if not result.data:
3345
      raise errors.OpPrereqError("One or more target bridges %s does not"
3346
                                 " exist on destination node '%s'" %
3347
                                 (brlist, target_node))
3348

    
3349
  def Exec(self, feedback_fn):
3350
    """Failover an instance.
3351

3352
    The failover is done by shutting it down on its present node and
3353
    starting it on the secondary.
3354

3355
    """
3356
    instance = self.instance
3357

    
3358
    source_node = instance.primary_node
3359
    target_node = instance.secondary_nodes[0]
3360

    
3361
    feedback_fn("* checking disk consistency between source and target")
3362
    for dev in instance.disks:
3363
      # for drbd, these are drbd over lvm
3364
      if not _CheckDiskConsistency(self, dev, target_node, False):
3365
        if instance.status == "up" and not self.op.ignore_consistency:
3366
          raise errors.OpExecError("Disk %s is degraded on target node,"
3367
                                   " aborting failover." % dev.iv_name)
3368

    
3369
    feedback_fn("* shutting down instance on source node")
3370
    logging.info("Shutting down instance %s on node %s",
3371
                 instance.name, source_node)
3372

    
3373
    result = self.rpc.call_instance_shutdown(source_node, instance)
3374
    if result.failed or not result.data:
3375
      if self.op.ignore_consistency:
3376
        self.proc.LogWarning("Could not shutdown instance %s on node %s."
3377
                             " Proceeding"
3378
                             " anyway. Please make sure node %s is down",
3379
                             instance.name, source_node, source_node)
3380
      else:
3381
        raise errors.OpExecError("Could not shutdown instance %s on node %s" %
3382
                                 (instance.name, source_node))
3383

    
3384
    feedback_fn("* deactivating the instance's disks on source node")
3385
    if not _ShutdownInstanceDisks(self, instance, ignore_primary=True):
3386
      raise errors.OpExecError("Can't shut down the instance's disks.")
3387

    
3388
    instance.primary_node = target_node
3389
    # distribute new instance config to the other nodes
3390
    self.cfg.Update(instance)
3391

    
3392
    # Only start the instance if it's marked as up
3393
    if instance.status == "up":
3394
      feedback_fn("* activating the instance's disks on target node")
3395
      logging.info("Starting instance %s on node %s",
3396
                   instance.name, target_node)
3397

    
3398
      disks_ok, dummy = _AssembleInstanceDisks(self, instance,
3399
                                               ignore_secondaries=True)
3400
      if not disks_ok:
3401
        _ShutdownInstanceDisks(self, instance)
3402
        raise errors.OpExecError("Can't activate the instance's disks")
3403

    
3404
      feedback_fn("* starting the instance on the target node")
3405
      result = self.rpc.call_instance_start(target_node, instance, None)
3406
      msg = result.RemoteFailMsg()
3407
      if msg:
3408
        _ShutdownInstanceDisks(self, instance)
3409
        raise errors.OpExecError("Could not start instance %s on node %s: %s" %
3410
                                 (instance.name, target_node, msg))
3411

    
3412

    
3413
class LUMigrateInstance(LogicalUnit):
3414
  """Migrate an instance.
3415

3416
  This is migration without shutting down, compared to the failover,
3417
  which is done with shutdown.
3418

3419
  """
3420
  HPATH = "instance-migrate"
3421
  HTYPE = constants.HTYPE_INSTANCE
3422
  _OP_REQP = ["instance_name", "live", "cleanup"]
3423

    
3424
  REQ_BGL = False
3425

    
3426
  def ExpandNames(self):
3427
    self._ExpandAndLockInstance()
3428
    self.needed_locks[locking.LEVEL_NODE] = []
3429
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3430

    
3431
  def DeclareLocks(self, level):
3432
    if level == locking.LEVEL_NODE:
3433
      self._LockInstancesNodes()
3434

    
3435
  def BuildHooksEnv(self):
3436
    """Build hooks env.
3437

3438
    This runs on master, primary and secondary nodes of the instance.
3439

3440
    """
3441
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3442
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3443
    return env, nl, nl
3444

    
3445
  def CheckPrereq(self):
3446
    """Check prerequisites.
3447

3448
    This checks that the instance is in the cluster.
3449

3450
    """
3451
    instance = self.cfg.GetInstanceInfo(
3452
      self.cfg.ExpandInstanceName(self.op.instance_name))
3453
    if instance is None:
3454
      raise errors.OpPrereqError("Instance '%s' not known" %
3455
                                 self.op.instance_name)
3456

    
3457
    if instance.disk_template != constants.DT_DRBD8:
3458
      raise errors.OpPrereqError("Instance's disk layout is not"
3459
                                 " drbd8, cannot migrate.")
3460

    
3461
    secondary_nodes = instance.secondary_nodes
3462
    if not secondary_nodes:
3463
      raise errors.ProgrammerError("no secondary node but using "
3464
                                   "drbd8 disk template")
3465

    
3466
    i_be = self.cfg.GetClusterInfo().FillBE(instance)
3467

    
3468
    target_node = secondary_nodes[0]
3469
    # check memory requirements on the secondary node
3470
    _CheckNodeFreeMemory(self, target_node, "migrating instance %s" %
3471
                         instance.name, i_be[constants.BE_MEMORY],
3472
                         instance.hypervisor)
3473

    
3474
    # check bridge existance
3475
    brlist = [nic.bridge for nic in instance.nics]
3476
    result = self.rpc.call_bridges_exist(target_node, brlist)
3477
    if result.failed or not result.data:
3478
      raise errors.OpPrereqError("One or more target bridges %s does not"
3479
                                 " exist on destination node '%s'" %
3480
                                 (brlist, target_node))
3481

    
3482
    if not self.op.cleanup:
3483
      result = self.rpc.call_instance_migratable(instance.primary_node,
3484
                                                 instance)
3485
      msg = result.RemoteFailMsg()
3486
      if msg:
3487
        raise errors.OpPrereqError("Can't migrate: %s - please use failover" %
3488
                                   msg)
3489

    
3490
    self.instance = instance
3491

    
3492
  def _WaitUntilSync(self):
3493
    """Poll with custom rpc for disk sync.
3494

3495
    This uses our own step-based rpc call.
3496

3497
    """
3498
    self.feedback_fn("* wait until resync is done")
3499
    all_done = False
3500
    while not all_done:
3501
      all_done = True
3502
      result = self.rpc.call_drbd_wait_sync(self.all_nodes,
3503
                                            self.nodes_ip,
3504
                                            self.instance.disks)
3505
      min_percent = 100
3506
      for node, nres in result.items():
3507
        msg = nres.RemoteFailMsg()
3508
        if msg:
3509
          raise errors.OpExecError("Cannot resync disks on node %s: %s" %
3510
                                   (node, msg))
3511
        node_done, node_percent = nres.data[1]
3512
        all_done = all_done and node_done
3513
        if node_percent is not None:
3514
          min_percent = min(min_percent, node_percent)
3515
      if not all_done:
3516
        if min_percent < 100:
3517
          self.feedback_fn("   - progress: %.1f%%" % min_percent)
3518
        time.sleep(2)
3519

    
3520
  def _EnsureSecondary(self, node):
3521
    """Demote a node to secondary.
3522

3523
    """
3524
    self.feedback_fn("* switching node %s to secondary mode" % node)
3525

    
3526
    for dev in self.instance.disks:
3527
      self.cfg.SetDiskID(dev, node)
3528

    
3529
    result = self.rpc.call_blockdev_close(node, self.instance.name,
3530
                                          self.instance.disks)
3531
    msg = result.RemoteFailMsg()
3532
    if msg:
3533
      raise errors.OpExecError("Cannot change disk to secondary on node %s,"
3534
                               " error %s" % (node, msg))
3535

    
3536
  def _GoStandalone(self):
3537
    """Disconnect from the network.
3538

3539
    """
3540
    self.feedback_fn("* changing into standalone mode")
3541
    result = self.rpc.call_drbd_disconnect_net(self.all_nodes, self.nodes_ip,
3542
                                               self.instance.disks)
3543
    for node, nres in result.items():
3544
      msg = nres.RemoteFailMsg()
3545
      if msg:
3546
        raise errors.OpExecError("Cannot disconnect disks node %s,"
3547
                                 " error %s" % (node, msg))
3548

    
3549
  def _GoReconnect(self, multimaster):
3550
    """Reconnect to the network.
3551

3552
    """
3553
    if multimaster:
3554
      msg = "dual-master"
3555
    else:
3556
      msg = "single-master"
3557
    self.feedback_fn("* changing disks into %s mode" % msg)
3558
    result = self.rpc.call_drbd_attach_net(self.all_nodes, self.nodes_ip,
3559
                                           self.instance.disks,
3560
                                           self.instance.name, multimaster)
3561
    for node, nres in result.items():
3562
      msg = nres.RemoteFailMsg()
3563
      if msg:
3564
        raise errors.OpExecError("Cannot change disks config on node %s,"
3565
                                 " error: %s" % (node, msg))
3566

    
3567
  def _ExecCleanup(self):
3568
    """Try to cleanup after a failed migration.
3569

3570
    The cleanup is done by:
3571
      - check that the instance is running only on one node
3572
        (and update the config if needed)
3573
      - change disks on its secondary node to secondary
3574
      - wait until disks are fully synchronized
3575
      - disconnect from the network
3576
      - change disks into single-master mode
3577
      - wait again until disks are fully synchronized
3578

3579
    """
3580
    instance = self.instance
3581
    target_node = self.target_node
3582
    source_node = self.source_node
3583

    
3584
    # check running on only one node
3585
    self.feedback_fn("* checking where the instance actually runs"
3586
                     " (if this hangs, the hypervisor might be in"
3587
                     " a bad state)")
3588
    ins_l = self.rpc.call_instance_list(self.all_nodes, [instance.hypervisor])
3589
    for node, result in ins_l.items():
3590
      result.Raise()
3591
      if not isinstance(result.data, list):
3592
        raise errors.OpExecError("Can't contact node '%s'" % node)
3593

    
3594
    runningon_source = instance.name in ins_l[source_node].data
3595
    runningon_target = instance.name in ins_l[target_node].data
3596

    
3597
    if runningon_source and runningon_target:
3598
      raise errors.OpExecError("Instance seems to be running on two nodes,"
3599
                               " or the hypervisor is confused. You will have"
3600
                               " to ensure manually that it runs only on one"
3601
                               " and restart this operation.")
3602

    
3603
    if not (runningon_source or runningon_target):
3604
      raise errors.OpExecError("Instance does not seem to be running at all."
3605
                               " In this case, it's safer to repair by"
3606
                               " running 'gnt-instance stop' to ensure disk"
3607
                               " shutdown, and then restarting it.")
3608

    
3609
    if runningon_target:
3610
      # the migration has actually succeeded, we need to update the config
3611
      self.feedback_fn("* instance running on secondary node (%s),"
3612
                       " updating config" % target_node)
3613
      instance.primary_node = target_node
3614
      self.cfg.Update(instance)
3615
      demoted_node = source_node
3616
    else:
3617
      self.feedback_fn("* instance confirmed to be running on its"
3618
                       " primary node (%s)" % source_node)
3619
      demoted_node = target_node
3620

    
3621
    self._EnsureSecondary(demoted_node)
3622
    try:
3623
      self._WaitUntilSync()
3624
    except errors.OpExecError:
3625
      # we ignore here errors, since if the device is standalone, it
3626
      # won't be able to sync
3627
      pass
3628
    self._GoStandalone()
3629
    self._GoReconnect(False)
3630
    self._WaitUntilSync()
3631

    
3632
    self.feedback_fn("* done")
3633

    
3634
  def _RevertDiskStatus(self):
3635
    """Try to revert the disk status after a failed migration.
3636

3637
    """
3638
    target_node = self.target_node
3639
    try:
3640
      self._EnsureSecondary(target_node)
3641
      self._GoStandalone()
3642
      self._GoReconnect(False)
3643
      self._WaitUntilSync()
3644
    except errors.OpExecError, err:
3645
      self.LogWarning("Migration failed and I can't reconnect the"
3646
                      " drives: error '%s'\n"
3647
                      "Please look and recover the instance status" %
3648
                      str(err))
3649

    
3650
  def _AbortMigration(self):
3651
    """Call the hypervisor code to abort a started migration.
3652

3653
    """
3654
    instance = self.instance
3655
    target_node = self.target_node
3656
    migration_info = self.migration_info
3657

    
3658
    abort_result = self.rpc.call_finalize_migration(target_node,
3659
                                                    instance,
3660
                                                    migration_info,
3661
                                                    False)
3662
    abort_msg = abort_result.RemoteFailMsg()
3663
    if abort_msg:
3664
      logging.error("Aborting migration failed on target node %s: %s" %
3665
                    (target_node, abort_msg))
3666
      # Don't raise an exception here, as we stil have to try to revert the
3667
      # disk status, even if this step failed.
3668

    
3669
  def _ExecMigration(self):
3670
    """Migrate an instance.
3671

3672
    The migrate is done by:
3673
      - change the disks into dual-master mode
3674
      - wait until disks are fully synchronized again
3675
      - migrate the instance
3676
      - change disks on the new secondary node (the old primary) to secondary
3677
      - wait until disks are fully synchronized
3678
      - change disks into single-master mode
3679

3680
    """
3681
    instance = self.instance
3682
    target_node = self.target_node
3683
    source_node = self.source_node
3684

    
3685
    self.feedback_fn("* checking disk consistency between source and target")
3686
    for dev in instance.disks:
3687
      if not _CheckDiskConsistency(self, dev, target_node, False):
3688
        raise errors.OpExecError("Disk %s is degraded or not fully"
3689
                                 " synchronized on target node,"
3690
                                 " aborting migrate." % dev.iv_name)
3691

    
3692
    # First get the migration information from the remote node
3693
    result = self.rpc.call_migration_info(source_node, instance)
3694
    msg = result.RemoteFailMsg()
3695
    if msg:
3696
      log_err = ("Failed fetching source migration information from %s: %s" %
3697
                  (source_node, msg))
3698
      logging.error(log_err)
3699
      raise errors.OpExecError(log_err)
3700

    
3701
    self.migration_info = migration_info = result.data[1]
3702

    
3703
    # Then switch the disks to master/master mode
3704
    self._EnsureSecondary(target_node)
3705
    self._GoStandalone()
3706
    self._GoReconnect(True)
3707
    self._WaitUntilSync()
3708

    
3709
    self.feedback_fn("* preparing %s to accept the instance" % target_node)
3710
    result = self.rpc.call_accept_instance(target_node,
3711
                                           instance,
3712
                                           migration_info,
3713
                                           self.nodes_ip[target_node])
3714

    
3715
    msg = result.RemoteFailMsg()
3716
    if msg:
3717
      logging.error("Instance pre-migration failed, trying to revert"
3718
                    " disk status: %s", msg)
3719
      self._AbortMigration()
3720
      self._RevertDiskStatus()
3721
      raise errors.OpExecError("Could not pre-migrate instance %s: %s" %
3722
                               (instance.name, msg))
3723

    
3724
    self.feedback_fn("* migrating instance to %s" % target_node)
3725
    time.sleep(10)
3726
    result = self.rpc.call_instance_migrate(source_node, instance,
3727
                                            self.nodes_ip[target_node],
3728
                                            self.op.live)
3729
    msg = result.RemoteFailMsg()
3730
    if msg:
3731
      logging.error("Instance migration failed, trying to revert"
3732
                    " disk status: %s", msg)
3733
      self._AbortMigration()
3734
      self._RevertDiskStatus()
3735
      raise errors.OpExecError("Could not migrate instance %s: %s" %
3736
                               (instance.name, msg))
3737
    time.sleep(10)
3738

    
3739
    instance.primary_node = target_node
3740
    # distribute new instance config to the other nodes
3741
    self.cfg.Update(instance)
3742

    
3743
    result = self.rpc.call_finalize_migration(target_node,
3744
                                              instance,
3745
                                              migration_info,
3746
                                              True)
3747
    msg = result.RemoteFailMsg()
3748
    if msg:
3749
      logging.error("Instance migration succeeded, but finalization failed:"
3750
                    " %s" % msg)
3751
      raise errors.OpExecError("Could not finalize instance migration: %s" %
3752
                               msg)
3753

    
3754
    self._EnsureSecondary(source_node)
3755
    self._WaitUntilSync()
3756
    self._GoStandalone()
3757
    self._GoReconnect(False)
3758
    self._WaitUntilSync()
3759

    
3760
    self.feedback_fn("* done")
3761

    
3762
  def Exec(self, feedback_fn):
3763
    """Perform the migration.
3764

3765
    """
3766
    self.feedback_fn = feedback_fn
3767

    
3768
    self.source_node = self.instance.primary_node
3769
    self.target_node = self.instance.secondary_nodes[0]
3770
    self.all_nodes = [self.source_node, self.target_node]
3771
    self.nodes_ip = {
3772
      self.source_node: self.cfg.GetNodeInfo(self.source_node).secondary_ip,
3773
      self.target_node: self.cfg.GetNodeInfo(self.target_node).secondary_ip,
3774
      }
3775
    if self.op.cleanup:
3776
      return self._ExecCleanup()
3777
    else:
3778
      return self._ExecMigration()
3779

    
3780

    
3781
def _CreateBlockDev(lu, node, instance, device, force_create,
3782
                    info, force_open):
3783
  """Create a tree of block devices on a given node.
3784

3785
  If this device type has to be created on secondaries, create it and
3786
  all its children.
3787

3788
  If not, just recurse to children keeping the same 'force' value.
3789

3790
  @param lu: the lu on whose behalf we execute
3791
  @param node: the node on which to create the device
3792
  @type instance: L{objects.Instance}
3793
  @param instance: the instance which owns the device
3794
  @type device: L{objects.Disk}
3795
  @param device: the device to create
3796
  @type force_create: boolean
3797
  @param force_create: whether to force creation of this device; this
3798
      will be change to True whenever we find a device which has
3799
      CreateOnSecondary() attribute
3800
  @param info: the extra 'metadata' we should attach to the device
3801
      (this will be represented as a LVM tag)
3802
  @type force_open: boolean
3803
  @param force_open: this parameter will be passes to the
3804
      L{backend.CreateBlockDevice} function where it specifies
3805
      whether we run on primary or not, and it affects both
3806
      the child assembly and the device own Open() execution
3807

3808
  """
3809
  if device.CreateOnSecondary():
3810
    force_create = True
3811

    
3812
  if device.children:
3813
    for child in device.children:
3814
      _CreateBlockDev(lu, node, instance, child, force_create,
3815
                      info, force_open)
3816

    
3817
  if not force_create:
3818
    return
3819

    
3820
  _CreateSingleBlockDev(lu, node, instance, device, info, force_open)
3821

    
3822

    
3823
def _CreateSingleBlockDev(lu, node, instance, device, info, force_open):
3824
  """Create a single block device on a given node.
3825

3826
  This will not recurse over children of the device, so they must be
3827
  created in advance.
3828

3829
  @param lu: the lu on whose behalf we execute
3830
  @param node: the node on which to create the device
3831
  @type instance: L{objects.Instance}
3832
  @param instance: the instance which owns the device
3833
  @type device: L{objects.Disk}
3834
  @param device: the device to create
3835
  @param info: the extra 'metadata' we should attach to the device
3836
      (this will be represented as a LVM tag)
3837
  @type force_open: boolean
3838
  @param force_open: this parameter will be passes to the
3839
      L{backend.CreateBlockDevice} function where it specifies
3840
      whether we run on primary or not, and it affects both
3841
      the child assembly and the device own Open() execution
3842

3843
  """
3844
  lu.cfg.SetDiskID(device, node)
3845
  result = lu.rpc.call_blockdev_create(node, device, device.size,
3846
                                       instance.name, force_open, info)
3847
  msg = result.RemoteFailMsg()
3848
  if msg:
3849
    raise errors.OpExecError("Can't create block device %s on"
3850
                             " node %s for instance %s: %s" %
3851
                             (device, node, instance.name, msg))
3852
  if device.physical_id is None:
3853
    device.physical_id = result.data[1]
3854

    
3855

    
3856
def _GenerateUniqueNames(lu, exts):
3857
  """Generate a suitable LV name.
3858

3859
  This will generate a logical volume name for the given instance.
3860

3861
  """
3862
  results = []
3863
  for val in exts:
3864
    new_id = lu.cfg.GenerateUniqueID()
3865
    results.append("%s%s" % (new_id, val))
3866
  return results
3867

    
3868

    
3869
def _GenerateDRBD8Branch(lu, primary, secondary, size, names, iv_name,
3870
                         p_minor, s_minor):
3871
  """Generate a drbd8 device complete with its children.
3872

3873
  """
3874
  port = lu.cfg.AllocatePort()
3875
  vgname = lu.cfg.GetVGName()
3876
  shared_secret = lu.cfg.GenerateDRBDSecret()
3877
  dev_data = objects.Disk(dev_type=constants.LD_LV, size=size,
3878
                          logical_id=(vgname, names[0]))
3879
  dev_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
3880
                          logical_id=(vgname, names[1]))
3881
  drbd_dev = objects.Disk(dev_type=constants.LD_DRBD8, size=size,
3882
                          logical_id=(primary, secondary, port,
3883
                                      p_minor, s_minor,
3884
                                      shared_secret),
3885
                          children=[dev_data, dev_meta],
3886
                          iv_name=iv_name)
3887
  return drbd_dev
3888

    
3889

    
3890
def _GenerateDiskTemplate(lu, template_name,
3891
                          instance_name, primary_node,
3892
                          secondary_nodes, disk_info,
3893
                          file_storage_dir, file_driver,
3894
                          base_index):
3895
  """Generate the entire disk layout for a given template type.
3896

3897
  """
3898
  #TODO: compute space requirements
3899

    
3900
  vgname = lu.cfg.GetVGName()
3901
  disk_count = len(disk_info)
3902
  disks = []
3903
  if template_name == constants.DT_DISKLESS:
3904
    pass
3905
  elif template_name == constants.DT_PLAIN:
3906
    if len(secondary_nodes) != 0:
3907
      raise errors.ProgrammerError("Wrong template configuration")
3908

    
3909
    names = _GenerateUniqueNames(lu, [".disk%d" % i
3910
                                      for i in range(disk_count)])
3911
    for idx, disk in enumerate(disk_info):
3912
      disk_index = idx + base_index
3913
      disk_dev = objects.Disk(dev_type=constants.LD_LV, size=disk["size"],
3914
                              logical_id=(vgname, names[idx]),
3915
                              iv_name="disk/%d" % disk_index)
3916
      disks.append(disk_dev)
3917
  elif template_name == constants.DT_DRBD8:
3918
    if len(secondary_nodes) != 1:
3919
      raise errors.ProgrammerError("Wrong template configuration")
3920
    remote_node = secondary_nodes[0]
3921
    minors = lu.cfg.AllocateDRBDMinor(
3922
      [primary_node, remote_node] * len(disk_info), instance_name)
3923

    
3924
    names = []
3925
    for lv_prefix in _GenerateUniqueNames(lu, [".disk%d" % i
3926
                                               for i in range(disk_count)]):
3927
      names.append(lv_prefix + "_data")
3928
      names.append(lv_prefix + "_meta")
3929
    for idx, disk in enumerate(disk_info):
3930
      disk_index = idx + base_index
3931
      disk_dev = _GenerateDRBD8Branch(lu, primary_node, remote_node,
3932
                                      disk["size"], names[idx*2:idx*2+2],
3933
                                      "disk/%d" % disk_index,
3934
                                      minors[idx*2], minors[idx*2+1])
3935
      disks.append(disk_dev)
3936
  elif template_name == constants.DT_FILE:
3937
    if len(secondary_nodes) != 0:
3938
      raise errors.ProgrammerError("Wrong template configuration")
3939

    
3940
    for idx, disk in enumerate(disk_info):
3941
      disk_index = idx + base_index
3942
      disk_dev = objects.Disk(dev_type=constants.LD_FILE, size=disk["size"],
3943
                              iv_name="disk/%d" % disk_index,
3944
                              logical_id=(file_driver,
3945
                                          "%s/disk%d" % (file_storage_dir,
3946
                                                         idx)))
3947
      disks.append(disk_dev)
3948
  else:
3949
    raise errors.ProgrammerError("Invalid disk template '%s'" % template_name)
3950
  return disks
3951

    
3952

    
3953
def _GetInstanceInfoText(instance):
3954
  """Compute that text that should be added to the disk's metadata.
3955

3956
  """
3957
  return "originstname+%s" % instance.name
3958

    
3959

    
3960
def _CreateDisks(lu, instance):
3961
  """Create all disks for an instance.
3962

3963
  This abstracts away some work from AddInstance.
3964

3965
  @type lu: L{LogicalUnit}
3966
  @param lu: the logical unit on whose behalf we execute
3967
  @type instance: L{objects.Instance}
3968
  @param instance: the instance whose disks we should create
3969
  @rtype: boolean
3970
  @return: the success of the creation
3971

3972
  """
3973
  info = _GetInstanceInfoText(instance)
3974
  pnode = instance.primary_node
3975

    
3976
  if instance.disk_template == constants.DT_FILE:
3977
    file_storage_dir = os.path.dirname(instance.disks[0].logical_id[1])
3978
    result = lu.rpc.call_file_storage_dir_create(pnode, file_storage_dir)
3979

    
3980
    if result.failed or not result.data:
3981
      raise errors.OpExecError("Could not connect to node '%s'" % pnode)
3982

    
3983
    if not result.data[0]:
3984
      raise errors.OpExecError("Failed to create directory '%s'" %
3985
                               file_storage_dir)
3986

    
3987
  # Note: this needs to be kept in sync with adding of disks in
3988
  # LUSetInstanceParams
3989
  for device in instance.disks:
3990
    logging.info("Creating volume %s for instance %s",
3991
                 device.iv_name, instance.name)
3992
    #HARDCODE