Statistics
| Branch: | Tag: | Revision:

root / lib / cmdlib.py @ 6b12959c

History | View | Annotate | Download (228.2 kB)

1
#
2
#
3

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

    
21

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

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

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

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

    
48

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

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

61
  Note that all commands require root permissions.
62

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

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

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

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

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

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

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

    
109
  ssh = property(fget=__GetSSH)
110

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

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

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

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

126
    """
127
    pass
128

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

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

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

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

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

150
    Examples::
151

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

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

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

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

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

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

189
    """
190

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

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

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

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

205
    """
206
    raise NotImplementedError
207

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

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

215
    """
216
    raise NotImplementedError
217

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

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

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

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

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

237
    """
238
    raise NotImplementedError
239

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

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

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

258
    """
259
    return lu_result
260

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

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

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

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

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

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

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

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

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

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

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

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

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

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

    
326
    del self.recalculate_locks[locking.LEVEL_NODE]
327

    
328

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

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

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

    
339

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

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

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

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

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

    
366
  return utils.NiceSort(wanted)
367

    
368

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

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

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

    
385
  if instances:
386
    wanted = []
387

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

    
394
  else:
395
    wanted = 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
    """Run multiple tests against a node.
618

619
    Test list:
620

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

626
    @type nodeinfo: L{objects.Node}
627
    @param nodeinfo: the node to check
628
    @param file_list: required list of files
629
    @param local_cksum: dictionary of local files and their checksums
630
    @param node_result: the results from the node
631
    @param feedback_fn: function used to accumulate results
632
    @param master_files: list of files that only masters should have
633

634
    """
635
    node = nodeinfo.name
636

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

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

    
649
    if local_version != remote_version:
650
      feedback_fn("  - ERROR: sw version mismatch: master %s, node(%s) %s" %
651
                      (local_version, node, remote_version))
652
      return True
653

    
654
    # checks vg existance and size > 20G
655

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

    
669
    # checks config file checksum
670

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

    
698
    # checks ssh to any
699

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

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

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

    
729
  def _VerifyInstance(self, instance, instanceconfig, node_vol_is,
730
                      node_instance, feedback_fn, n_offline):
731
    """Verify an instance.
732

733
    This function checks to see if the required block devices are
734
    available on the instance's node.
735

736
    """
737
    bad = False
738

    
739
    node_current = instanceconfig.primary_node
740

    
741
    node_vol_should = {}
742
    instanceconfig.MapLVsByNode(node_vol_should)
743

    
744
    for node in node_vol_should:
745
      if node in n_offline:
746
        # ignore missing volumes on offline nodes
747
        continue
748
      for volume in node_vol_should[node]:
749
        if node not in node_vol_is or volume not in node_vol_is[node]:
750
          feedback_fn("  - ERROR: volume %s missing on node %s" %
751
                          (volume, node))
752
          bad = True
753

    
754
    if not instanceconfig.status == 'down':
755
      if ((node_current not in node_instance or
756
          not instance in node_instance[node_current]) and
757
          node_current not in n_offline):
758
        feedback_fn("  - ERROR: instance %s not running on node %s" %
759
                        (instance, node_current))
760
        bad = True
761

    
762
    for node in node_instance:
763
      if (not node == node_current):
764
        if instance in node_instance[node]:
765
          feedback_fn("  - ERROR: instance %s should not run on node %s" %
766
                          (instance, node))
767
          bad = True
768

    
769
    return bad
770

    
771
  def _VerifyOrphanVolumes(self, node_vol_should, node_vol_is, feedback_fn):
772
    """Verify if there are any unknown volumes in the cluster.
773

774
    The .os, .swap and backup volumes are ignored. All other volumes are
775
    reported as unknown.
776

777
    """
778
    bad = False
779

    
780
    for node in node_vol_is:
781
      for volume in node_vol_is[node]:
782
        if node not in node_vol_should or volume not in node_vol_should[node]:
783
          feedback_fn("  - ERROR: volume %s on node %s should not exist" %
784
                      (volume, node))
785
          bad = True
786
    return bad
787

    
788
  def _VerifyOrphanInstances(self, instancelist, node_instance, feedback_fn):
789
    """Verify the list of running instances.
790

791
    This checks what instances are running but unknown to the cluster.
792

793
    """
794
    bad = False
795
    for node in node_instance:
796
      for runninginstance in node_instance[node]:
797
        if runninginstance not in instancelist:
798
          feedback_fn("  - ERROR: instance %s on node %s should not exist" %
799
                          (runninginstance, node))
800
          bad = True
801
    return bad
802

    
803
  def _VerifyNPlusOneMemory(self, node_info, instance_cfg, feedback_fn):
804
    """Verify N+1 Memory Resilience.
805

806
    Check that if one single node dies we can still start all the instances it
807
    was primary for.
808

809
    """
810
    bad = False
811

    
812
    for node, nodeinfo in node_info.iteritems():
813
      # This code checks that every node which is now listed as secondary has
814
      # enough memory to host all instances it is supposed to should a single
815
      # other node in the cluster fail.
816
      # FIXME: not ready for failover to an arbitrary node
817
      # FIXME: does not support file-backed instances
818
      # WARNING: we currently take into account down instances as well as up
819
      # ones, considering that even if they're down someone might want to start
820
      # them even in the event of a node failure.
821
      for prinode, instances in nodeinfo['sinst-by-pnode'].iteritems():
822
        needed_mem = 0
823
        for instance in instances:
824
          bep = self.cfg.GetClusterInfo().FillBE(instance_cfg[instance])
825
          if bep[constants.BE_AUTO_BALANCE]:
826
            needed_mem += bep[constants.BE_MEMORY]
827
        if nodeinfo['mfree'] < needed_mem:
828
          feedback_fn("  - ERROR: not enough memory on node %s to accomodate"
829
                      " failovers should node %s fail" % (node, prinode))
830
          bad = True
831
    return bad
832

    
833
  def CheckPrereq(self):
834
    """Check prerequisites.
835

836
    Transform the list of checks we're going to skip into a set and check that
837
    all its members are valid.
838

839
    """
840
    self.skip_set = frozenset(self.op.skip_checks)
841
    if not constants.VERIFY_OPTIONAL_CHECKS.issuperset(self.skip_set):
842
      raise errors.OpPrereqError("Invalid checks to be skipped specified")
843

    
844
  def BuildHooksEnv(self):
845
    """Build hooks env.
846

847
    Cluster-Verify hooks just rone in the post phase and their failure makes
848
    the output be logged in the verify output and the verification to fail.
849

850
    """
851
    all_nodes = self.cfg.GetNodeList()
852
    # TODO: populate the environment with useful information for verify hooks
853
    env = {}
854
    return env, [], all_nodes
855

    
856
  def Exec(self, feedback_fn):
857
    """Verify integrity of cluster, performing various test on nodes.
858

859
    """
860
    bad = False
861
    feedback_fn("* Verifying global settings")
862
    for msg in self.cfg.VerifyConfig():
863
      feedback_fn("  - ERROR: %s" % msg)
864

    
865
    vg_name = self.cfg.GetVGName()
866
    hypervisors = self.cfg.GetClusterInfo().enabled_hypervisors
867
    nodelist = utils.NiceSort(self.cfg.GetNodeList())
868
    nodeinfo = [self.cfg.GetNodeInfo(nname) for nname in nodelist]
869
    instancelist = utils.NiceSort(self.cfg.GetInstanceList())
870
    i_non_redundant = [] # Non redundant instances
871
    i_non_a_balanced = [] # Non auto-balanced instances
872
    n_offline = [] # List of offline nodes
873
    node_volume = {}
874
    node_instance = {}
875
    node_info = {}
876
    instance_cfg = {}
877

    
878
    # FIXME: verify OS list
879
    # do local checksums
880
    master_files = [constants.CLUSTER_CONF_FILE]
881

    
882
    file_names = ssconf.SimpleStore().GetFileList()
883
    file_names.append(constants.SSL_CERT_FILE)
884
    file_names.append(constants.RAPI_CERT_FILE)
885
    file_names.extend(master_files)
886

    
887
    local_checksums = utils.FingerprintFiles(file_names)
888

    
889
    feedback_fn("* Gathering data (%d nodes)" % len(nodelist))
890
    node_verify_param = {
891
      constants.NV_FILELIST: file_names,
892
      constants.NV_NODELIST: [node.name for node in nodeinfo
893
                              if not node.offline],
894
      constants.NV_HYPERVISOR: hypervisors,
895
      constants.NV_NODENETTEST: [(node.name, node.primary_ip,
896
                                  node.secondary_ip) for node in nodeinfo
897
                                 if not node.offline],
898
      constants.NV_LVLIST: vg_name,
899
      constants.NV_INSTANCELIST: hypervisors,
900
      constants.NV_VGLIST: None,
901
      constants.NV_VERSION: None,
902
      constants.NV_HVINFO: self.cfg.GetHypervisorType(),
903
      }
904
    all_nvinfo = self.rpc.call_node_verify(nodelist, node_verify_param,
905
                                           self.cfg.GetClusterName())
906

    
907
    cluster = self.cfg.GetClusterInfo()
908
    master_node = self.cfg.GetMasterNode()
909
    for node_i in nodeinfo:
910
      node = node_i.name
911
      nresult = all_nvinfo[node].data
912

    
913
      if node_i.offline:
914
        feedback_fn("* Skipping offline node %s" % (node,))
915
        n_offline.append(node)
916
        continue
917

    
918
      if node == master_node:
919
        ntype = "master"
920
      elif node_i.master_candidate:
921
        ntype = "master candidate"
922
      else:
923
        ntype = "regular"
924
      feedback_fn("* Verifying node %s (%s)" % (node, ntype))
925

    
926
      if all_nvinfo[node].failed or not isinstance(nresult, dict):
927
        feedback_fn("  - ERROR: connection to %s failed" % (node,))
928
        bad = True
929
        continue
930

    
931
      result = self._VerifyNode(node_i, file_names, local_checksums,
932
                                nresult, feedback_fn, master_files)
933
      bad = bad or result
934

    
935
      lvdata = nresult.get(constants.NV_LVLIST, "Missing LV data")
936
      if isinstance(lvdata, basestring):
937
        feedback_fn("  - ERROR: LVM problem on node %s: %s" %
938
                    (node, lvdata.encode('string_escape')))
939
        bad = True
940
        node_volume[node] = {}
941
      elif not isinstance(lvdata, dict):
942
        feedback_fn("  - ERROR: connection to %s failed (lvlist)" % (node,))
943
        bad = True
944
        continue
945
      else:
946
        node_volume[node] = lvdata
947

    
948
      # node_instance
949
      idata = nresult.get(constants.NV_INSTANCELIST, None)
950
      if not isinstance(idata, list):
951
        feedback_fn("  - ERROR: connection to %s failed (instancelist)" %
952
                    (node,))
953
        bad = True
954
        continue
955

    
956
      node_instance[node] = idata
957

    
958
      # node_info
959
      nodeinfo = nresult.get(constants.NV_HVINFO, None)
960
      if not isinstance(nodeinfo, dict):
961
        feedback_fn("  - ERROR: connection to %s failed (hvinfo)" % (node,))
962
        bad = True
963
        continue
964

    
965
      try:
966
        node_info[node] = {
967
          "mfree": int(nodeinfo['memory_free']),
968
          "dfree": int(nresult[constants.NV_VGLIST][vg_name]),
969
          "pinst": [],
970
          "sinst": [],
971
          # dictionary holding all instances this node is secondary for,
972
          # grouped by their primary node. Each key is a cluster node, and each
973
          # value is a list of instances which have the key as primary and the
974
          # current node as secondary.  this is handy to calculate N+1 memory
975
          # availability if you can only failover from a primary to its
976
          # secondary.
977
          "sinst-by-pnode": {},
978
        }
979
      except ValueError:
980
        feedback_fn("  - ERROR: invalid value returned from node %s" % (node,))
981
        bad = True
982
        continue
983

    
984
    node_vol_should = {}
985

    
986
    for instance in instancelist:
987
      feedback_fn("* Verifying instance %s" % instance)
988
      inst_config = self.cfg.GetInstanceInfo(instance)
989
      result =  self._VerifyInstance(instance, inst_config, node_volume,
990
                                     node_instance, feedback_fn, n_offline)
991
      bad = bad or result
992
      inst_nodes_offline = []
993

    
994
      inst_config.MapLVsByNode(node_vol_should)
995

    
996
      instance_cfg[instance] = inst_config
997

    
998
      pnode = inst_config.primary_node
999
      if pnode in node_info:
1000
        node_info[pnode]['pinst'].append(instance)
1001
      elif pnode not in n_offline:
1002
        feedback_fn("  - ERROR: instance %s, connection to primary node"
1003
                    " %s failed" % (instance, pnode))
1004
        bad = True
1005

    
1006
      if pnode in n_offline:
1007
        inst_nodes_offline.append(pnode)
1008

    
1009
      # If the instance is non-redundant we cannot survive losing its primary
1010
      # node, so we are not N+1 compliant. On the other hand we have no disk
1011
      # templates with more than one secondary so that situation is not well
1012
      # supported either.
1013
      # FIXME: does not support file-backed instances
1014
      if len(inst_config.secondary_nodes) == 0:
1015
        i_non_redundant.append(instance)
1016
      elif len(inst_config.secondary_nodes) > 1:
1017
        feedback_fn("  - WARNING: multiple secondaries for instance %s"
1018
                    % instance)
1019

    
1020
      if not cluster.FillBE(inst_config)[constants.BE_AUTO_BALANCE]:
1021
        i_non_a_balanced.append(instance)
1022

    
1023
      for snode in inst_config.secondary_nodes:
1024
        if snode in node_info:
1025
          node_info[snode]['sinst'].append(instance)
1026
          if pnode not in node_info[snode]['sinst-by-pnode']:
1027
            node_info[snode]['sinst-by-pnode'][pnode] = []
1028
          node_info[snode]['sinst-by-pnode'][pnode].append(instance)
1029
        elif snode not in n_offline:
1030
          feedback_fn("  - ERROR: instance %s, connection to secondary node"
1031
                      " %s failed" % (instance, snode))
1032
          bad = True
1033
        if snode in n_offline:
1034
          inst_nodes_offline.append(snode)
1035

    
1036
      if inst_nodes_offline:
1037
        # warn that the instance lives on offline nodes, and set bad=True
1038
        feedback_fn("  - ERROR: instance lives on offline node(s) %s" %
1039
                    ", ".join(inst_nodes_offline))
1040
        bad = True
1041

    
1042
    feedback_fn("* Verifying orphan volumes")
1043
    result = self._VerifyOrphanVolumes(node_vol_should, node_volume,
1044
                                       feedback_fn)
1045
    bad = bad or result
1046

    
1047
    feedback_fn("* Verifying remaining instances")
1048
    result = self._VerifyOrphanInstances(instancelist, node_instance,
1049
                                         feedback_fn)
1050
    bad = bad or result
1051

    
1052
    if constants.VERIFY_NPLUSONE_MEM not in self.skip_set:
1053
      feedback_fn("* Verifying N+1 Memory redundancy")
1054
      result = self._VerifyNPlusOneMemory(node_info, instance_cfg, feedback_fn)
1055
      bad = bad or result
1056

    
1057
    feedback_fn("* Other Notes")
1058
    if i_non_redundant:
1059
      feedback_fn("  - NOTICE: %d non-redundant instance(s) found."
1060
                  % len(i_non_redundant))
1061

    
1062
    if i_non_a_balanced:
1063
      feedback_fn("  - NOTICE: %d non-auto-balanced instance(s) found."
1064
                  % len(i_non_a_balanced))
1065

    
1066
    if n_offline:
1067
      feedback_fn("  - NOTICE: %d offline node(s) found." % len(n_offline))
1068

    
1069
    return not bad
1070

    
1071
  def HooksCallBack(self, phase, hooks_results, feedback_fn, lu_result):
1072
    """Analize the post-hooks' result
1073

1074
    This method analyses the hook result, handles it, and sends some
1075
    nicely-formatted feedback back to the user.
1076

1077
    @param phase: one of L{constants.HOOKS_PHASE_POST} or
1078
        L{constants.HOOKS_PHASE_PRE}; it denotes the hooks phase
1079
    @param hooks_results: the results of the multi-node hooks rpc call
1080
    @param feedback_fn: function used send feedback back to the caller
1081
    @param lu_result: previous Exec result
1082
    @return: the new Exec result, based on the previous result
1083
        and hook results
1084

1085
    """
1086
    # We only really run POST phase hooks, and are only interested in
1087
    # their results
1088
    if phase == constants.HOOKS_PHASE_POST:
1089
      # Used to change hooks' output to proper indentation
1090
      indent_re = re.compile('^', re.M)
1091
      feedback_fn("* Hooks Results")
1092
      if not hooks_results:
1093
        feedback_fn("  - ERROR: general communication failure")
1094
        lu_result = 1
1095
      else:
1096
        for node_name in hooks_results:
1097
          show_node_header = True
1098
          res = hooks_results[node_name]
1099
          if res.failed or res.data is False or not isinstance(res.data, list):
1100
            if res.offline:
1101
              # no need to warn or set fail return value
1102
              continue
1103
            feedback_fn("    Communication failure in hooks execution")
1104
            lu_result = 1
1105
            continue
1106
          for script, hkr, output in res.data:
1107
            if hkr == constants.HKR_FAIL:
1108
              # The node header is only shown once, if there are
1109
              # failing hooks on that node
1110
              if show_node_header:
1111
                feedback_fn("  Node %s:" % node_name)
1112
                show_node_header = False
1113
              feedback_fn("    ERROR: Script %s failed, output:" % script)
1114
              output = indent_re.sub('      ', output)
1115
              feedback_fn("%s" % output)
1116
              lu_result = 1
1117

    
1118
      return lu_result
1119

    
1120

    
1121
class LUVerifyDisks(NoHooksLU):
1122
  """Verifies the cluster disks status.
1123

1124
  """
1125
  _OP_REQP = []
1126
  REQ_BGL = False
1127

    
1128
  def ExpandNames(self):
1129
    self.needed_locks = {
1130
      locking.LEVEL_NODE: locking.ALL_SET,
1131
      locking.LEVEL_INSTANCE: locking.ALL_SET,
1132
    }
1133
    self.share_locks = dict(((i, 1) for i in locking.LEVELS))
1134

    
1135
  def CheckPrereq(self):
1136
    """Check prerequisites.
1137

1138
    This has no prerequisites.
1139

1140
    """
1141
    pass
1142

    
1143
  def Exec(self, feedback_fn):
1144
    """Verify integrity of cluster disks.
1145

1146
    """
1147
    result = res_nodes, res_nlvm, res_instances, res_missing = [], {}, [], {}
1148

    
1149
    vg_name = self.cfg.GetVGName()
1150
    nodes = utils.NiceSort(self.cfg.GetNodeList())
1151
    instances = [self.cfg.GetInstanceInfo(name)
1152
                 for name in self.cfg.GetInstanceList()]
1153

    
1154
    nv_dict = {}
1155
    for inst in instances:
1156
      inst_lvs = {}
1157
      if (inst.status != "up" or
1158
          inst.disk_template not in constants.DTS_NET_MIRROR):
1159
        continue
1160
      inst.MapLVsByNode(inst_lvs)
1161
      # transform { iname: {node: [vol,],},} to {(node, vol): iname}
1162
      for node, vol_list in inst_lvs.iteritems():
1163
        for vol in vol_list:
1164
          nv_dict[(node, vol)] = inst
1165

    
1166
    if not nv_dict:
1167
      return result
1168

    
1169
    node_lvs = self.rpc.call_volume_list(nodes, vg_name)
1170

    
1171
    to_act = set()
1172
    for node in nodes:
1173
      # node_volume
1174
      lvs = node_lvs[node]
1175
      if lvs.failed:
1176
        if not lvs.offline:
1177
          self.LogWarning("Connection to node %s failed: %s" %
1178
                          (node, lvs.data))
1179
        continue
1180
      lvs = lvs.data
1181
      if isinstance(lvs, basestring):
1182
        logging.warning("Error enumerating LVs on node %s: %s", node, lvs)
1183
        res_nlvm[node] = lvs
1184
      elif not isinstance(lvs, dict):
1185
        logging.warning("Connection to node %s failed or invalid data"
1186
                        " returned", node)
1187
        res_nodes.append(node)
1188
        continue
1189

    
1190
      for lv_name, (_, lv_inactive, lv_online) in lvs.iteritems():
1191
        inst = nv_dict.pop((node, lv_name), None)
1192
        if (not lv_online and inst is not None
1193
            and inst.name not in res_instances):
1194
          res_instances.append(inst.name)
1195

    
1196
    # any leftover items in nv_dict are missing LVs, let's arrange the
1197
    # data better
1198
    for key, inst in nv_dict.iteritems():
1199
      if inst.name not in res_missing:
1200
        res_missing[inst.name] = []
1201
      res_missing[inst.name].append(key)
1202

    
1203
    return result
1204

    
1205

    
1206
class LURenameCluster(LogicalUnit):
1207
  """Rename the cluster.
1208

1209
  """
1210
  HPATH = "cluster-rename"
1211
  HTYPE = constants.HTYPE_CLUSTER
1212
  _OP_REQP = ["name"]
1213

    
1214
  def BuildHooksEnv(self):
1215
    """Build hooks env.
1216

1217
    """
1218
    env = {
1219
      "OP_TARGET": self.cfg.GetClusterName(),
1220
      "NEW_NAME": self.op.name,
1221
      }
1222
    mn = self.cfg.GetMasterNode()
1223
    return env, [mn], [mn]
1224

    
1225
  def CheckPrereq(self):
1226
    """Verify that the passed name is a valid one.
1227

1228
    """
1229
    hostname = utils.HostInfo(self.op.name)
1230

    
1231
    new_name = hostname.name
1232
    self.ip = new_ip = hostname.ip
1233
    old_name = self.cfg.GetClusterName()
1234
    old_ip = self.cfg.GetMasterIP()
1235
    if new_name == old_name and new_ip == old_ip:
1236
      raise errors.OpPrereqError("Neither the name nor the IP address of the"
1237
                                 " cluster has changed")
1238
    if new_ip != old_ip:
1239
      if utils.TcpPing(new_ip, constants.DEFAULT_NODED_PORT):
1240
        raise errors.OpPrereqError("The given cluster IP address (%s) is"
1241
                                   " reachable on the network. Aborting." %
1242
                                   new_ip)
1243

    
1244
    self.op.name = new_name
1245

    
1246
  def Exec(self, feedback_fn):
1247
    """Rename the cluster.
1248

1249
    """
1250
    clustername = self.op.name
1251
    ip = self.ip
1252

    
1253
    # shutdown the master IP
1254
    master = self.cfg.GetMasterNode()
1255
    result = self.rpc.call_node_stop_master(master, False)
1256
    if result.failed or not result.data:
1257
      raise errors.OpExecError("Could not disable the master role")
1258

    
1259
    try:
1260
      cluster = self.cfg.GetClusterInfo()
1261
      cluster.cluster_name = clustername
1262
      cluster.master_ip = ip
1263
      self.cfg.Update(cluster)
1264

    
1265
      # update the known hosts file
1266
      ssh.WriteKnownHostsFile(self.cfg, constants.SSH_KNOWN_HOSTS_FILE)
1267
      node_list = self.cfg.GetNodeList()
1268
      try:
1269
        node_list.remove(master)
1270
      except ValueError:
1271
        pass
1272
      result = self.rpc.call_upload_file(node_list,
1273
                                         constants.SSH_KNOWN_HOSTS_FILE)
1274
      for to_node, to_result in result.iteritems():
1275
        if to_result.failed or not to_result.data:
1276
          logging.error("Copy of file %s to node %s failed",
1277
                        constants.SSH_KNOWN_HOSTS_FILE, to_node)
1278

    
1279
    finally:
1280
      result = self.rpc.call_node_start_master(master, False)
1281
      if result.failed or not result.data:
1282
        self.LogWarning("Could not re-enable the master role on"
1283
                        " the master, please restart manually.")
1284

    
1285

    
1286
def _RecursiveCheckIfLVMBased(disk):
1287
  """Check if the given disk or its children are lvm-based.
1288

1289
  @type disk: L{objects.Disk}
1290
  @param disk: the disk to check
1291
  @rtype: booleean
1292
  @return: boolean indicating whether a LD_LV dev_type was found or not
1293

1294
  """
1295
  if disk.children:
1296
    for chdisk in disk.children:
1297
      if _RecursiveCheckIfLVMBased(chdisk):
1298
        return True
1299
  return disk.dev_type == constants.LD_LV
1300

    
1301

    
1302
class LUSetClusterParams(LogicalUnit):
1303
  """Change the parameters of the cluster.
1304

1305
  """
1306
  HPATH = "cluster-modify"
1307
  HTYPE = constants.HTYPE_CLUSTER
1308
  _OP_REQP = []
1309
  REQ_BGL = False
1310

    
1311
  def CheckParameters(self):
1312
    """Check parameters
1313

1314
    """
1315
    if not hasattr(self.op, "candidate_pool_size"):
1316
      self.op.candidate_pool_size = None
1317
    if self.op.candidate_pool_size is not None:
1318
      try:
1319
        self.op.candidate_pool_size = int(self.op.candidate_pool_size)
1320
      except ValueError, err:
1321
        raise errors.OpPrereqError("Invalid candidate_pool_size value: %s" %
1322
                                   str(err))
1323
      if self.op.candidate_pool_size < 1:
1324
        raise errors.OpPrereqError("At least one master candidate needed")
1325

    
1326
  def ExpandNames(self):
1327
    # FIXME: in the future maybe other cluster params won't require checking on
1328
    # all nodes to be modified.
1329
    self.needed_locks = {
1330
      locking.LEVEL_NODE: locking.ALL_SET,
1331
    }
1332
    self.share_locks[locking.LEVEL_NODE] = 1
1333

    
1334
  def BuildHooksEnv(self):
1335
    """Build hooks env.
1336

1337
    """
1338
    env = {
1339
      "OP_TARGET": self.cfg.GetClusterName(),
1340
      "NEW_VG_NAME": self.op.vg_name,
1341
      }
1342
    mn = self.cfg.GetMasterNode()
1343
    return env, [mn], [mn]
1344

    
1345
  def CheckPrereq(self):
1346
    """Check prerequisites.
1347

1348
    This checks whether the given params don't conflict and
1349
    if the given volume group is valid.
1350

1351
    """
1352
    # FIXME: This only works because there is only one parameter that can be
1353
    # changed or removed.
1354
    if self.op.vg_name is not None and not self.op.vg_name:
1355
      instances = self.cfg.GetAllInstancesInfo().values()
1356
      for inst in instances:
1357
        for disk in inst.disks:
1358
          if _RecursiveCheckIfLVMBased(disk):
1359
            raise errors.OpPrereqError("Cannot disable lvm storage while"
1360
                                       " lvm-based instances exist")
1361

    
1362
    node_list = self.acquired_locks[locking.LEVEL_NODE]
1363

    
1364
    # if vg_name not None, checks given volume group on all nodes
1365
    if self.op.vg_name:
1366
      vglist = self.rpc.call_vg_list(node_list)
1367
      for node in node_list:
1368
        if vglist[node].failed:
1369
          # ignoring down node
1370
          self.LogWarning("Node %s unreachable/error, ignoring" % node)
1371
          continue
1372
        vgstatus = utils.CheckVolumeGroupSize(vglist[node].data,
1373
                                              self.op.vg_name,
1374
                                              constants.MIN_VG_SIZE)
1375
        if vgstatus:
1376
          raise errors.OpPrereqError("Error on node '%s': %s" %
1377
                                     (node, vgstatus))
1378

    
1379
    self.cluster = cluster = self.cfg.GetClusterInfo()
1380
    # validate beparams changes
1381
    if self.op.beparams:
1382
      utils.CheckBEParams(self.op.beparams)
1383
      self.new_beparams = cluster.FillDict(
1384
        cluster.beparams[constants.BEGR_DEFAULT], self.op.beparams)
1385

    
1386
    # hypervisor list/parameters
1387
    self.new_hvparams = cluster.FillDict(cluster.hvparams, {})
1388
    if self.op.hvparams:
1389
      if not isinstance(self.op.hvparams, dict):
1390
        raise errors.OpPrereqError("Invalid 'hvparams' parameter on input")
1391
      for hv_name, hv_dict in self.op.hvparams.items():
1392
        if hv_name not in self.new_hvparams:
1393
          self.new_hvparams[hv_name] = hv_dict
1394
        else:
1395
          self.new_hvparams[hv_name].update(hv_dict)
1396

    
1397
    if self.op.enabled_hypervisors is not None:
1398
      self.hv_list = self.op.enabled_hypervisors
1399
    else:
1400
      self.hv_list = cluster.enabled_hypervisors
1401

    
1402
    if self.op.hvparams or self.op.enabled_hypervisors is not None:
1403
      # either the enabled list has changed, or the parameters have, validate
1404
      for hv_name, hv_params in self.new_hvparams.items():
1405
        if ((self.op.hvparams and hv_name in self.op.hvparams) or
1406
            (self.op.enabled_hypervisors and
1407
             hv_name in self.op.enabled_hypervisors)):
1408
          # either this is a new hypervisor, or its parameters have changed
1409
          hv_class = hypervisor.GetHypervisor(hv_name)
1410
          hv_class.CheckParameterSyntax(hv_params)
1411
          _CheckHVParams(self, node_list, hv_name, hv_params)
1412

    
1413
  def Exec(self, feedback_fn):
1414
    """Change the parameters of the cluster.
1415

1416
    """
1417
    if self.op.vg_name is not None:
1418
      if self.op.vg_name != self.cfg.GetVGName():
1419
        self.cfg.SetVGName(self.op.vg_name)
1420
      else:
1421
        feedback_fn("Cluster LVM configuration already in desired"
1422
                    " state, not changing")
1423
    if self.op.hvparams:
1424
      self.cluster.hvparams = self.new_hvparams
1425
    if self.op.enabled_hypervisors is not None:
1426
      self.cluster.enabled_hypervisors = self.op.enabled_hypervisors
1427
    if self.op.beparams:
1428
      self.cluster.beparams[constants.BEGR_DEFAULT] = self.new_beparams
1429
    if self.op.candidate_pool_size is not None:
1430
      self.cluster.candidate_pool_size = self.op.candidate_pool_size
1431

    
1432
    self.cfg.Update(self.cluster)
1433

    
1434
    # we want to update nodes after the cluster so that if any errors
1435
    # happen, we have recorded and saved the cluster info
1436
    if self.op.candidate_pool_size is not None:
1437
      _AdjustCandidatePool(self)
1438

    
1439

    
1440
class LURedistributeConfig(NoHooksLU):
1441
  """Force the redistribution of cluster configuration.
1442

1443
  This is a very simple LU.
1444

1445
  """
1446
  _OP_REQP = []
1447
  REQ_BGL = False
1448

    
1449
  def ExpandNames(self):
1450
    self.needed_locks = {
1451
      locking.LEVEL_NODE: locking.ALL_SET,
1452
    }
1453
    self.share_locks[locking.LEVEL_NODE] = 1
1454

    
1455
  def CheckPrereq(self):
1456
    """Check prerequisites.
1457

1458
    """
1459

    
1460
  def Exec(self, feedback_fn):
1461
    """Redistribute the configuration.
1462

1463
    """
1464
    self.cfg.Update(self.cfg.GetClusterInfo())
1465

    
1466

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

1470
  """
1471
  if not instance.disks:
1472
    return True
1473

    
1474
  if not oneshot:
1475
    lu.proc.LogInfo("Waiting for instance %s to sync disks." % instance.name)
1476

    
1477
  node = instance.primary_node
1478

    
1479
  for dev in instance.disks:
1480
    lu.cfg.SetDiskID(dev, node)
1481

    
1482
  retries = 0
1483
  while True:
1484
    max_time = 0
1485
    done = True
1486
    cumul_degraded = False
1487
    rstats = lu.rpc.call_blockdev_getmirrorstatus(node, instance.disks)
1488
    if rstats.failed or not rstats.data:
1489
      lu.LogWarning("Can't get any data from node %s", node)
1490
      retries += 1
1491
      if retries >= 10:
1492
        raise errors.RemoteError("Can't contact node %s for mirror data,"
1493
                                 " aborting." % node)
1494
      time.sleep(6)
1495
      continue
1496
    rstats = rstats.data
1497
    retries = 0
1498
    for i in range(len(rstats)):
1499
      mstat = rstats[i]
1500
      if mstat is None:
1501
        lu.LogWarning("Can't compute data for node %s/%s",
1502
                           node, instance.disks[i].iv_name)
1503
        continue
1504
      # we ignore the ldisk parameter
1505
      perc_done, est_time, is_degraded, _ = mstat
1506
      cumul_degraded = cumul_degraded or (is_degraded and perc_done is None)
1507
      if perc_done is not None:
1508
        done = False
1509
        if est_time is not None:
1510
          rem_time = "%d estimated seconds remaining" % est_time
1511
          max_time = est_time
1512
        else:
1513
          rem_time = "no time estimate"
1514
        lu.proc.LogInfo("- device %s: %5.2f%% done, %s" %
1515
                        (instance.disks[i].iv_name, perc_done, rem_time))
1516
    if done or oneshot:
1517
      break
1518

    
1519
    time.sleep(min(60, max_time))
1520

    
1521
  if done:
1522
    lu.proc.LogInfo("Instance %s's disks are in sync." % instance.name)
1523
  return not cumul_degraded
1524

    
1525

    
1526
def _CheckDiskConsistency(lu, dev, node, on_primary, ldisk=False):
1527
  """Check that mirrors are not degraded.
1528

1529
  The ldisk parameter, if True, will change the test from the
1530
  is_degraded attribute (which represents overall non-ok status for
1531
  the device(s)) to the ldisk (representing the local storage status).
1532

1533
  """
1534
  lu.cfg.SetDiskID(dev, node)
1535
  if ldisk:
1536
    idx = 6
1537
  else:
1538
    idx = 5
1539

    
1540
  result = True
1541
  if on_primary or dev.AssembleOnSecondary():
1542
    rstats = lu.rpc.call_blockdev_find(node, dev)
1543
    if rstats.failed or not rstats.data:
1544
      logging.warning("Node %s: disk degraded, not found or node down", node)
1545
      result = False
1546
    else:
1547
      result = result and (not rstats.data[idx])
1548
  if dev.children:
1549
    for child in dev.children:
1550
      result = result and _CheckDiskConsistency(lu, child, node, on_primary)
1551

    
1552
  return result
1553

    
1554

    
1555
class LUDiagnoseOS(NoHooksLU):
1556
  """Logical unit for OS diagnose/query.
1557

1558
  """
1559
  _OP_REQP = ["output_fields", "names"]
1560
  REQ_BGL = False
1561
  _FIELDS_STATIC = utils.FieldSet()
1562
  _FIELDS_DYNAMIC = utils.FieldSet("name", "valid", "node_status")
1563

    
1564
  def ExpandNames(self):
1565
    if self.op.names:
1566
      raise errors.OpPrereqError("Selective OS query not supported")
1567

    
1568
    _CheckOutputFields(static=self._FIELDS_STATIC,
1569
                       dynamic=self._FIELDS_DYNAMIC,
1570
                       selected=self.op.output_fields)
1571

    
1572
    # Lock all nodes, in shared mode
1573
    self.needed_locks = {}
1574
    self.share_locks[locking.LEVEL_NODE] = 1
1575
    self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
1576

    
1577
  def CheckPrereq(self):
1578
    """Check prerequisites.
1579

1580
    """
1581

    
1582
  @staticmethod
1583
  def _DiagnoseByOS(node_list, rlist):
1584
    """Remaps a per-node return list into an a per-os per-node dictionary
1585

1586
    @param node_list: a list with the names of all nodes
1587
    @param rlist: a map with node names as keys and OS objects as values
1588

1589
    @rtype: dict
1590
    @returns: a dictionary with osnames as keys and as value another map, with
1591
        nodes as keys and list of OS objects as values, eg::
1592

1593
          {"debian-etch": {"node1": [<object>,...],
1594
                           "node2": [<object>,]}
1595
          }
1596

1597
    """
1598
    all_os = {}
1599
    for node_name, nr in rlist.iteritems():
1600
      if nr.failed or not nr.data:
1601
        continue
1602
      for os_obj in nr.data:
1603
        if os_obj.name not in all_os:
1604
          # build a list of nodes for this os containing empty lists
1605
          # for each node in node_list
1606
          all_os[os_obj.name] = {}
1607
          for nname in node_list:
1608
            all_os[os_obj.name][nname] = []
1609
        all_os[os_obj.name][node_name].append(os_obj)
1610
    return all_os
1611

    
1612
  def Exec(self, feedback_fn):
1613
    """Compute the list of OSes.
1614

1615
    """
1616
    node_list = self.acquired_locks[locking.LEVEL_NODE]
1617
    valid_nodes = [node for node in self.cfg.GetOnlineNodeList()
1618
                   if node in node_list]
1619
    node_data = self.rpc.call_os_diagnose(valid_nodes)
1620
    if node_data == False:
1621
      raise errors.OpExecError("Can't gather the list of OSes")
1622
    pol = self._DiagnoseByOS(valid_nodes, node_data)
1623
    output = []
1624
    for os_name, os_data in pol.iteritems():
1625
      row = []
1626
      for field in self.op.output_fields:
1627
        if field == "name":
1628
          val = os_name
1629
        elif field == "valid":
1630
          val = utils.all([osl and osl[0] for osl in os_data.values()])
1631
        elif field == "node_status":
1632
          val = {}
1633
          for node_name, nos_list in os_data.iteritems():
1634
            val[node_name] = [(v.status, v.path) for v in nos_list]
1635
        else:
1636
          raise errors.ParameterError(field)
1637
        row.append(val)
1638
      output.append(row)
1639

    
1640
    return output
1641

    
1642

    
1643
class LURemoveNode(LogicalUnit):
1644
  """Logical unit for removing a node.
1645

1646
  """
1647
  HPATH = "node-remove"
1648
  HTYPE = constants.HTYPE_NODE
1649
  _OP_REQP = ["node_name"]
1650

    
1651
  def BuildHooksEnv(self):
1652
    """Build hooks env.
1653

1654
    This doesn't run on the target node in the pre phase as a failed
1655
    node would then be impossible to remove.
1656

1657
    """
1658
    env = {
1659
      "OP_TARGET": self.op.node_name,
1660
      "NODE_NAME": self.op.node_name,
1661
      }
1662
    all_nodes = self.cfg.GetNodeList()
1663
    all_nodes.remove(self.op.node_name)
1664
    return env, all_nodes, all_nodes
1665

    
1666
  def CheckPrereq(self):
1667
    """Check prerequisites.
1668

1669
    This checks:
1670
     - the node exists in the configuration
1671
     - it does not have primary or secondary instances
1672
     - it's not the master
1673

1674
    Any errors are signalled by raising errors.OpPrereqError.
1675

1676
    """
1677
    node = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.node_name))
1678
    if node is None:
1679
      raise errors.OpPrereqError, ("Node '%s' is unknown." % self.op.node_name)
1680

    
1681
    instance_list = self.cfg.GetInstanceList()
1682

    
1683
    masternode = self.cfg.GetMasterNode()
1684
    if node.name == masternode:
1685
      raise errors.OpPrereqError("Node is the master node,"
1686
                                 " you need to failover first.")
1687

    
1688
    for instance_name in instance_list:
1689
      instance = self.cfg.GetInstanceInfo(instance_name)
1690
      if node.name in instance.all_nodes:
1691
        raise errors.OpPrereqError("Instance %s is still running on the node,"
1692
                                   " please remove first." % instance_name)
1693
    self.op.node_name = node.name
1694
    self.node = node
1695

    
1696
  def Exec(self, feedback_fn):
1697
    """Removes the node from the cluster.
1698

1699
    """
1700
    node = self.node
1701
    logging.info("Stopping the node daemon and removing configs from node %s",
1702
                 node.name)
1703

    
1704
    self.context.RemoveNode(node.name)
1705

    
1706
    self.rpc.call_node_leave_cluster(node.name)
1707

    
1708
    # Promote nodes to master candidate as needed
1709
    _AdjustCandidatePool(self)
1710

    
1711

    
1712
class LUQueryNodes(NoHooksLU):
1713
  """Logical unit for querying nodes.
1714

1715
  """
1716
  _OP_REQP = ["output_fields", "names"]
1717
  REQ_BGL = False
1718
  _FIELDS_DYNAMIC = utils.FieldSet(
1719
    "dtotal", "dfree",
1720
    "mtotal", "mnode", "mfree",
1721
    "bootid",
1722
    "ctotal",
1723
    )
1724

    
1725
  _FIELDS_STATIC = utils.FieldSet(
1726
    "name", "pinst_cnt", "sinst_cnt",
1727
    "pinst_list", "sinst_list",
1728
    "pip", "sip", "tags",
1729
    "serial_no",
1730
    "master_candidate",
1731
    "master",
1732
    "offline",
1733
    )
1734

    
1735
  def ExpandNames(self):
1736
    _CheckOutputFields(static=self._FIELDS_STATIC,
1737
                       dynamic=self._FIELDS_DYNAMIC,
1738
                       selected=self.op.output_fields)
1739

    
1740
    self.needed_locks = {}
1741
    self.share_locks[locking.LEVEL_NODE] = 1
1742

    
1743
    if self.op.names:
1744
      self.wanted = _GetWantedNodes(self, self.op.names)
1745
    else:
1746
      self.wanted = locking.ALL_SET
1747

    
1748
    self.do_locking = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
1749
    if self.do_locking:
1750
      # if we don't request only static fields, we need to lock the nodes
1751
      self.needed_locks[locking.LEVEL_NODE] = self.wanted
1752

    
1753

    
1754
  def CheckPrereq(self):
1755
    """Check prerequisites.
1756

1757
    """
1758
    # The validation of the node list is done in the _GetWantedNodes,
1759
    # if non empty, and if empty, there's no validation to do
1760
    pass
1761

    
1762
  def Exec(self, feedback_fn):
1763
    """Computes the list of nodes and their attributes.
1764

1765
    """
1766
    all_info = self.cfg.GetAllNodesInfo()
1767
    if self.do_locking:
1768
      nodenames = self.acquired_locks[locking.LEVEL_NODE]
1769
    elif self.wanted != locking.ALL_SET:
1770
      nodenames = self.wanted
1771
      missing = set(nodenames).difference(all_info.keys())
1772
      if missing:
1773
        raise errors.OpExecError(
1774
          "Some nodes were removed before retrieving their data: %s" % missing)
1775
    else:
1776
      nodenames = all_info.keys()
1777

    
1778
    nodenames = utils.NiceSort(nodenames)
1779
    nodelist = [all_info[name] for name in nodenames]
1780

    
1781
    # begin data gathering
1782

    
1783
    if self.do_locking:
1784
      live_data = {}
1785
      node_data = self.rpc.call_node_info(nodenames, self.cfg.GetVGName(),
1786
                                          self.cfg.GetHypervisorType())
1787
      for name in nodenames:
1788
        nodeinfo = node_data[name]
1789
        if not nodeinfo.failed and nodeinfo.data:
1790
          nodeinfo = nodeinfo.data
1791
          fn = utils.TryConvert
1792
          live_data[name] = {
1793
            "mtotal": fn(int, nodeinfo.get('memory_total', None)),
1794
            "mnode": fn(int, nodeinfo.get('memory_dom0', None)),
1795
            "mfree": fn(int, nodeinfo.get('memory_free', None)),
1796
            "dtotal": fn(int, nodeinfo.get('vg_size', None)),
1797
            "dfree": fn(int, nodeinfo.get('vg_free', None)),
1798
            "ctotal": fn(int, nodeinfo.get('cpu_total', None)),
1799
            "bootid": nodeinfo.get('bootid', None),
1800
            }
1801
        else:
1802
          live_data[name] = {}
1803
    else:
1804
      live_data = dict.fromkeys(nodenames, {})
1805

    
1806
    node_to_primary = dict([(name, set()) for name in nodenames])
1807
    node_to_secondary = dict([(name, set()) for name in nodenames])
1808

    
1809
    inst_fields = frozenset(("pinst_cnt", "pinst_list",
1810
                             "sinst_cnt", "sinst_list"))
1811
    if inst_fields & frozenset(self.op.output_fields):
1812
      instancelist = self.cfg.GetInstanceList()
1813

    
1814
      for instance_name in instancelist:
1815
        inst = self.cfg.GetInstanceInfo(instance_name)
1816
        if inst.primary_node in node_to_primary:
1817
          node_to_primary[inst.primary_node].add(inst.name)
1818
        for secnode in inst.secondary_nodes:
1819
          if secnode in node_to_secondary:
1820
            node_to_secondary[secnode].add(inst.name)
1821

    
1822
    master_node = self.cfg.GetMasterNode()
1823

    
1824
    # end data gathering
1825

    
1826
    output = []
1827
    for node in nodelist:
1828
      node_output = []
1829
      for field in self.op.output_fields:
1830
        if field == "name":
1831
          val = node.name
1832
        elif field == "pinst_list":
1833
          val = list(node_to_primary[node.name])
1834
        elif field == "sinst_list":
1835
          val = list(node_to_secondary[node.name])
1836
        elif field == "pinst_cnt":
1837
          val = len(node_to_primary[node.name])
1838
        elif field == "sinst_cnt":
1839
          val = len(node_to_secondary[node.name])
1840
        elif field == "pip":
1841
          val = node.primary_ip
1842
        elif field == "sip":
1843
          val = node.secondary_ip
1844
        elif field == "tags":
1845
          val = list(node.GetTags())
1846
        elif field == "serial_no":
1847
          val = node.serial_no
1848
        elif field == "master_candidate":
1849
          val = node.master_candidate
1850
        elif field == "master":
1851
          val = node.name == master_node
1852
        elif field == "offline":
1853
          val = node.offline
1854
        elif self._FIELDS_DYNAMIC.Matches(field):
1855
          val = live_data[node.name].get(field, None)
1856
        else:
1857
          raise errors.ParameterError(field)
1858
        node_output.append(val)
1859
      output.append(node_output)
1860

    
1861
    return output
1862

    
1863

    
1864
class LUQueryNodeVolumes(NoHooksLU):
1865
  """Logical unit for getting volumes on node(s).
1866

1867
  """
1868
  _OP_REQP = ["nodes", "output_fields"]
1869
  REQ_BGL = False
1870
  _FIELDS_DYNAMIC = utils.FieldSet("phys", "vg", "name", "size", "instance")
1871
  _FIELDS_STATIC = utils.FieldSet("node")
1872

    
1873
  def ExpandNames(self):
1874
    _CheckOutputFields(static=self._FIELDS_STATIC,
1875
                       dynamic=self._FIELDS_DYNAMIC,
1876
                       selected=self.op.output_fields)
1877

    
1878
    self.needed_locks = {}
1879
    self.share_locks[locking.LEVEL_NODE] = 1
1880
    if not self.op.nodes:
1881
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
1882
    else:
1883
      self.needed_locks[locking.LEVEL_NODE] = \
1884
        _GetWantedNodes(self, self.op.nodes)
1885

    
1886
  def CheckPrereq(self):
1887
    """Check prerequisites.
1888

1889
    This checks that the fields required are valid output fields.
1890

1891
    """
1892
    self.nodes = self.acquired_locks[locking.LEVEL_NODE]
1893

    
1894
  def Exec(self, feedback_fn):
1895
    """Computes the list of nodes and their attributes.
1896

1897
    """
1898
    nodenames = self.nodes
1899
    volumes = self.rpc.call_node_volumes(nodenames)
1900

    
1901
    ilist = [self.cfg.GetInstanceInfo(iname) for iname
1902
             in self.cfg.GetInstanceList()]
1903

    
1904
    lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
1905

    
1906
    output = []
1907
    for node in nodenames:
1908
      if node not in volumes or volumes[node].failed or not volumes[node].data:
1909
        continue
1910

    
1911
      node_vols = volumes[node].data[:]
1912
      node_vols.sort(key=lambda vol: vol['dev'])
1913

    
1914
      for vol in node_vols:
1915
        node_output = []
1916
        for field in self.op.output_fields:
1917
          if field == "node":
1918
            val = node
1919
          elif field == "phys":
1920
            val = vol['dev']
1921
          elif field == "vg":
1922
            val = vol['vg']
1923
          elif field == "name":
1924
            val = vol['name']
1925
          elif field == "size":
1926
            val = int(float(vol['size']))
1927
          elif field == "instance":
1928
            for inst in ilist:
1929
              if node not in lv_by_node[inst]:
1930
                continue
1931
              if vol['name'] in lv_by_node[inst][node]:
1932
                val = inst.name
1933
                break
1934
            else:
1935
              val = '-'
1936
          else:
1937
            raise errors.ParameterError(field)
1938
          node_output.append(str(val))
1939

    
1940
        output.append(node_output)
1941

    
1942
    return output
1943

    
1944

    
1945
class LUAddNode(LogicalUnit):
1946
  """Logical unit for adding node to the cluster.
1947

1948
  """
1949
  HPATH = "node-add"
1950
  HTYPE = constants.HTYPE_NODE
1951
  _OP_REQP = ["node_name"]
1952

    
1953
  def BuildHooksEnv(self):
1954
    """Build hooks env.
1955

1956
    This will run on all nodes before, and on all nodes + the new node after.
1957

1958
    """
1959
    env = {
1960
      "OP_TARGET": self.op.node_name,
1961
      "NODE_NAME": self.op.node_name,
1962
      "NODE_PIP": self.op.primary_ip,
1963
      "NODE_SIP": self.op.secondary_ip,
1964
      }
1965
    nodes_0 = self.cfg.GetNodeList()
1966
    nodes_1 = nodes_0 + [self.op.node_name, ]
1967
    return env, nodes_0, nodes_1
1968

    
1969
  def CheckPrereq(self):
1970
    """Check prerequisites.
1971

1972
    This checks:
1973
     - the new node is not already in the config
1974
     - it is resolvable
1975
     - its parameters (single/dual homed) matches the cluster
1976

1977
    Any errors are signalled by raising errors.OpPrereqError.
1978

1979
    """
1980
    node_name = self.op.node_name
1981
    cfg = self.cfg
1982

    
1983
    dns_data = utils.HostInfo(node_name)
1984

    
1985
    node = dns_data.name
1986
    primary_ip = self.op.primary_ip = dns_data.ip
1987
    secondary_ip = getattr(self.op, "secondary_ip", None)
1988
    if secondary_ip is None:
1989
      secondary_ip = primary_ip
1990
    if not utils.IsValidIP(secondary_ip):
1991
      raise errors.OpPrereqError("Invalid secondary IP given")
1992
    self.op.secondary_ip = secondary_ip
1993

    
1994
    node_list = cfg.GetNodeList()
1995
    if not self.op.readd and node in node_list:
1996
      raise errors.OpPrereqError("Node %s is already in the configuration" %
1997
                                 node)
1998
    elif self.op.readd and node not in node_list:
1999
      raise errors.OpPrereqError("Node %s is not in the configuration" % node)
2000

    
2001
    for existing_node_name in node_list:
2002
      existing_node = cfg.GetNodeInfo(existing_node_name)
2003

    
2004
      if self.op.readd and node == existing_node_name:
2005
        if (existing_node.primary_ip != primary_ip or
2006
            existing_node.secondary_ip != secondary_ip):
2007
          raise errors.OpPrereqError("Readded node doesn't have the same IP"
2008
                                     " address configuration as before")
2009
        continue
2010

    
2011
      if (existing_node.primary_ip == primary_ip or
2012
          existing_node.secondary_ip == primary_ip or
2013
          existing_node.primary_ip == secondary_ip or
2014
          existing_node.secondary_ip == secondary_ip):
2015
        raise errors.OpPrereqError("New node ip address(es) conflict with"
2016
                                   " existing node %s" % existing_node.name)
2017

    
2018
    # check that the type of the node (single versus dual homed) is the
2019
    # same as for the master
2020
    myself = cfg.GetNodeInfo(self.cfg.GetMasterNode())
2021
    master_singlehomed = myself.secondary_ip == myself.primary_ip
2022
    newbie_singlehomed = secondary_ip == primary_ip
2023
    if master_singlehomed != newbie_singlehomed:
2024
      if master_singlehomed:
2025
        raise errors.OpPrereqError("The master has no private ip but the"
2026
                                   " new node has one")
2027
      else:
2028
        raise errors.OpPrereqError("The master has a private ip but the"
2029
                                   " new node doesn't have one")
2030

    
2031
    # checks reachablity
2032
    if not utils.TcpPing(primary_ip, constants.DEFAULT_NODED_PORT):
2033
      raise errors.OpPrereqError("Node not reachable by ping")
2034

    
2035
    if not newbie_singlehomed:
2036
      # check reachability from my secondary ip to newbie's secondary ip
2037
      if not utils.TcpPing(secondary_ip, constants.DEFAULT_NODED_PORT,
2038
                           source=myself.secondary_ip):
2039
        raise errors.OpPrereqError("Node secondary ip not reachable by TCP"
2040
                                   " based ping to noded port")
2041

    
2042
    cp_size = self.cfg.GetClusterInfo().candidate_pool_size
2043
    mc_now, _ = self.cfg.GetMasterCandidateStats()
2044
    master_candidate = mc_now < cp_size
2045

    
2046
    self.new_node = objects.Node(name=node,
2047
                                 primary_ip=primary_ip,
2048
                                 secondary_ip=secondary_ip,
2049
                                 master_candidate=master_candidate,
2050
                                 offline=False)
2051

    
2052
  def Exec(self, feedback_fn):
2053
    """Adds the new node to the cluster.
2054

2055
    """
2056
    new_node = self.new_node
2057
    node = new_node.name
2058

    
2059
    # check connectivity
2060
    result = self.rpc.call_version([node])[node]
2061
    result.Raise()
2062
    if result.data:
2063
      if constants.PROTOCOL_VERSION == result.data:
2064
        logging.info("Communication to node %s fine, sw version %s match",
2065
                     node, result.data)
2066
      else:
2067
        raise errors.OpExecError("Version mismatch master version %s,"
2068
                                 " node version %s" %
2069
                                 (constants.PROTOCOL_VERSION, result.data))
2070
    else:
2071
      raise errors.OpExecError("Cannot get version from the new node")
2072

    
2073
    # setup ssh on node
2074
    logging.info("Copy ssh key to node %s", node)
2075
    priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
2076
    keyarray = []
2077
    keyfiles = [constants.SSH_HOST_DSA_PRIV, constants.SSH_HOST_DSA_PUB,
2078
                constants.SSH_HOST_RSA_PRIV, constants.SSH_HOST_RSA_PUB,
2079
                priv_key, pub_key]
2080

    
2081
    for i in keyfiles:
2082
      f = open(i, 'r')
2083
      try:
2084
        keyarray.append(f.read())
2085
      finally:
2086
        f.close()
2087

    
2088
    result = self.rpc.call_node_add(node, keyarray[0], keyarray[1],
2089
                                    keyarray[2],
2090
                                    keyarray[3], keyarray[4], keyarray[5])
2091

    
2092
    if result.failed or not result.data:
2093
      raise errors.OpExecError("Cannot transfer ssh keys to the new node")
2094

    
2095
    # Add node to our /etc/hosts, and add key to known_hosts
2096
    utils.AddHostToEtcHosts(new_node.name)
2097

    
2098
    if new_node.secondary_ip != new_node.primary_ip:
2099
      result = self.rpc.call_node_has_ip_address(new_node.name,
2100
                                                 new_node.secondary_ip)
2101
      if result.failed or not result.data:
2102
        raise errors.OpExecError("Node claims it doesn't have the secondary ip"
2103
                                 " you gave (%s). Please fix and re-run this"
2104
                                 " command." % new_node.secondary_ip)
2105

    
2106
    node_verify_list = [self.cfg.GetMasterNode()]
2107
    node_verify_param = {
2108
      'nodelist': [node],
2109
      # TODO: do a node-net-test as well?
2110
    }
2111

    
2112
    result = self.rpc.call_node_verify(node_verify_list, node_verify_param,
2113
                                       self.cfg.GetClusterName())
2114
    for verifier in node_verify_list:
2115
      if result[verifier].failed or not result[verifier].data:
2116
        raise errors.OpExecError("Cannot communicate with %s's node daemon"
2117
                                 " for remote verification" % verifier)
2118
      if result[verifier].data['nodelist']:
2119
        for failed in result[verifier].data['nodelist']:
2120
          feedback_fn("ssh/hostname verification failed %s -> %s" %
2121
                      (verifier, result[verifier]['nodelist'][failed]))
2122
        raise errors.OpExecError("ssh/hostname verification failed.")
2123

    
2124
    # Distribute updated /etc/hosts and known_hosts to all nodes,
2125
    # including the node just added
2126
    myself = self.cfg.GetNodeInfo(self.cfg.GetMasterNode())
2127
    dist_nodes = self.cfg.GetNodeList()
2128
    if not self.op.readd:
2129
      dist_nodes.append(node)
2130
    if myself.name in dist_nodes:
2131
      dist_nodes.remove(myself.name)
2132

    
2133
    logging.debug("Copying hosts and known_hosts to all nodes")
2134
    for fname in (constants.ETC_HOSTS, constants.SSH_KNOWN_HOSTS_FILE):
2135
      result = self.rpc.call_upload_file(dist_nodes, fname)
2136
      for to_node, to_result in result.iteritems():
2137
        if to_result.failed or not to_result.data:
2138
          logging.error("Copy of file %s to node %s failed", fname, to_node)
2139

    
2140
    to_copy = []
2141
    if constants.HT_XEN_HVM in self.cfg.GetClusterInfo().enabled_hypervisors:
2142
      to_copy.append(constants.VNC_PASSWORD_FILE)
2143
    for fname in to_copy:
2144
      result = self.rpc.call_upload_file([node], fname)
2145
      if result[node].failed or not result[node]:
2146
        logging.error("Could not copy file %s to node %s", fname, node)
2147

    
2148
    if self.op.readd:
2149
      self.context.ReaddNode(new_node)
2150
    else:
2151
      self.context.AddNode(new_node)
2152

    
2153

    
2154
class LUSetNodeParams(LogicalUnit):
2155
  """Modifies the parameters of a node.
2156

2157
  """
2158
  HPATH = "node-modify"
2159
  HTYPE = constants.HTYPE_NODE
2160
  _OP_REQP = ["node_name"]
2161
  REQ_BGL = False
2162

    
2163
  def CheckArguments(self):
2164
    node_name = self.cfg.ExpandNodeName(self.op.node_name)
2165
    if node_name is None:
2166
      raise errors.OpPrereqError("Invalid node name '%s'" % self.op.node_name)
2167
    self.op.node_name = node_name
2168
    _CheckBooleanOpField(self.op, 'master_candidate')
2169
    _CheckBooleanOpField(self.op, 'offline')
2170
    if self.op.master_candidate is None and self.op.offline is None:
2171
      raise errors.OpPrereqError("Please pass at least one modification")
2172
    if self.op.offline == True and self.op.master_candidate == True:
2173
      raise errors.OpPrereqError("Can't set the node into offline and"
2174
                                 " master_candidate at the same time")
2175

    
2176
  def ExpandNames(self):
2177
    self.needed_locks = {locking.LEVEL_NODE: self.op.node_name}
2178

    
2179
  def BuildHooksEnv(self):
2180
    """Build hooks env.
2181

2182
    This runs on the master node.
2183

2184
    """
2185
    env = {
2186
      "OP_TARGET": self.op.node_name,
2187
      "MASTER_CANDIDATE": str(self.op.master_candidate),
2188
      "OFFLINE": str(self.op.offline),
2189
      }
2190
    nl = [self.cfg.GetMasterNode(),
2191
          self.op.node_name]
2192
    return env, nl, nl
2193

    
2194
  def CheckPrereq(self):
2195
    """Check prerequisites.
2196

2197
    This only checks the instance list against the existing names.
2198

2199
    """
2200
    node = self.node = self.cfg.GetNodeInfo(self.op.node_name)
2201

    
2202
    if ((self.op.master_candidate == False or self.op.offline == True)
2203
        and node.master_candidate):
2204
      # we will demote the node from master_candidate
2205
      if self.op.node_name == self.cfg.GetMasterNode():
2206
        raise errors.OpPrereqError("The master node has to be a"
2207
                                   " master candidate and online")
2208
      cp_size = self.cfg.GetClusterInfo().candidate_pool_size
2209
      num_candidates, _ = self.cfg.GetMasterCandidateStats()
2210
      if num_candidates <= cp_size:
2211
        msg = ("Not enough master candidates (desired"
2212
               " %d, new value will be %d)" % (cp_size, num_candidates-1))
2213
        if self.op.force:
2214
          self.LogWarning(msg)
2215
        else:
2216
          raise errors.OpPrereqError(msg)
2217

    
2218
    if (self.op.master_candidate == True and node.offline and
2219
        not self.op.offline == False):
2220
      raise errors.OpPrereqError("Can't set an offline node to"
2221
                                 " master_candidate")
2222

    
2223
    return
2224

    
2225
  def Exec(self, feedback_fn):
2226
    """Modifies a node.
2227

2228
    """
2229
    node = self.node
2230

    
2231
    result = []
2232

    
2233
    if self.op.offline is not None:
2234
      node.offline = self.op.offline
2235
      result.append(("offline", str(self.op.offline)))
2236
      if self.op.offline == True and node.master_candidate:
2237
        node.master_candidate = False
2238
        result.append(("master_candidate", "auto-demotion due to offline"))
2239

    
2240
    if self.op.master_candidate is not None:
2241
      node.master_candidate = self.op.master_candidate
2242
      result.append(("master_candidate", str(self.op.master_candidate)))
2243
      if self.op.master_candidate == False:
2244
        rrc = self.rpc.call_node_demote_from_mc(node.name)
2245
        if (rrc.failed or not isinstance(rrc.data, (tuple, list))
2246
            or len(rrc.data) != 2):
2247
          self.LogWarning("Node rpc error: %s" % rrc.error)
2248
        elif not rrc.data[0]:
2249
          self.LogWarning("Node failed to demote itself: %s" % rrc.data[1])
2250

    
2251
    # this will trigger configuration file update, if needed
2252
    self.cfg.Update(node)
2253
    # this will trigger job queue propagation or cleanup
2254
    if self.op.node_name != self.cfg.GetMasterNode():
2255
      self.context.ReaddNode(node)
2256

    
2257
    return result
2258

    
2259

    
2260
class LUQueryClusterInfo(NoHooksLU):
2261
  """Query cluster configuration.
2262

2263
  """
2264
  _OP_REQP = []
2265
  REQ_BGL = False
2266

    
2267
  def ExpandNames(self):
2268
    self.needed_locks = {}
2269

    
2270
  def CheckPrereq(self):
2271
    """No prerequsites needed for this LU.
2272

2273
    """
2274
    pass
2275

    
2276
  def Exec(self, feedback_fn):
2277
    """Return cluster config.
2278

2279
    """
2280
    cluster = self.cfg.GetClusterInfo()
2281
    result = {
2282
      "software_version": constants.RELEASE_VERSION,
2283
      "protocol_version": constants.PROTOCOL_VERSION,
2284
      "config_version": constants.CONFIG_VERSION,
2285
      "os_api_version": constants.OS_API_VERSION,
2286
      "export_version": constants.EXPORT_VERSION,
2287
      "architecture": (platform.architecture()[0], platform.machine()),
2288
      "name": cluster.cluster_name,
2289
      "master": cluster.master_node,
2290
      "default_hypervisor": cluster.default_hypervisor,
2291
      "enabled_hypervisors": cluster.enabled_hypervisors,
2292
      "hvparams": cluster.hvparams,
2293
      "beparams": cluster.beparams,
2294
      "candidate_pool_size": cluster.candidate_pool_size,
2295
      }
2296

    
2297
    return result
2298

    
2299

    
2300
class LUQueryConfigValues(NoHooksLU):
2301
  """Return configuration values.
2302

2303
  """
2304
  _OP_REQP = []
2305
  REQ_BGL = False
2306
  _FIELDS_DYNAMIC = utils.FieldSet()
2307
  _FIELDS_STATIC = utils.FieldSet("cluster_name", "master_node", "drain_flag")
2308

    
2309
  def ExpandNames(self):
2310
    self.needed_locks = {}
2311

    
2312
    _CheckOutputFields(static=self._FIELDS_STATIC,
2313
                       dynamic=self._FIELDS_DYNAMIC,
2314
                       selected=self.op.output_fields)
2315

    
2316
  def CheckPrereq(self):
2317
    """No prerequisites.
2318

2319
    """
2320
    pass
2321

    
2322
  def Exec(self, feedback_fn):
2323
    """Dump a representation of the cluster config to the standard output.
2324

2325
    """
2326
    values = []
2327
    for field in self.op.output_fields:
2328
      if field == "cluster_name":
2329
        entry = self.cfg.GetClusterName()
2330
      elif field == "master_node":
2331
        entry = self.cfg.GetMasterNode()
2332
      elif field == "drain_flag":
2333
        entry = os.path.exists(constants.JOB_QUEUE_DRAIN_FILE)
2334
      else:
2335
        raise errors.ParameterError(field)
2336
      values.append(entry)
2337
    return values
2338

    
2339

    
2340
class LUActivateInstanceDisks(NoHooksLU):
2341
  """Bring up an instance's disks.
2342

2343
  """
2344
  _OP_REQP = ["instance_name"]
2345
  REQ_BGL = False
2346

    
2347
  def ExpandNames(self):
2348
    self._ExpandAndLockInstance()
2349
    self.needed_locks[locking.LEVEL_NODE] = []
2350
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2351

    
2352
  def DeclareLocks(self, level):
2353
    if level == locking.LEVEL_NODE:
2354
      self._LockInstancesNodes()
2355

    
2356
  def CheckPrereq(self):
2357
    """Check prerequisites.
2358

2359
    This checks that the instance is in the cluster.
2360

2361
    """
2362
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2363
    assert self.instance is not None, \
2364
      "Cannot retrieve locked instance %s" % self.op.instance_name
2365
    _CheckNodeOnline(self, self.instance.primary_node)
2366

    
2367
  def Exec(self, feedback_fn):
2368
    """Activate the disks.
2369

2370
    """
2371
    disks_ok, disks_info = _AssembleInstanceDisks(self, self.instance)
2372
    if not disks_ok:
2373
      raise errors.OpExecError("Cannot activate block devices")
2374

    
2375
    return disks_info
2376

    
2377

    
2378
def _AssembleInstanceDisks(lu, instance, ignore_secondaries=False):
2379
  """Prepare the block devices for an instance.
2380

2381
  This sets up the block devices on all nodes.
2382

2383
  @type lu: L{LogicalUnit}
2384
  @param lu: the logical unit on whose behalf we execute
2385
  @type instance: L{objects.Instance}
2386
  @param instance: the instance for whose disks we assemble
2387
  @type ignore_secondaries: boolean
2388
  @param ignore_secondaries: if true, errors on secondary nodes
2389
      won't result in an error return from the function
2390
  @return: False if the operation failed, otherwise a list of
2391
      (host, instance_visible_name, node_visible_name)
2392
      with the mapping from node devices to instance devices
2393

2394
  """
2395
  device_info = []
2396
  disks_ok = True
2397
  iname = instance.name
2398
  # With the two passes mechanism we try to reduce the window of
2399
  # opportunity for the race condition of switching DRBD to primary
2400
  # before handshaking occured, but we do not eliminate it
2401

    
2402
  # The proper fix would be to wait (with some limits) until the
2403
  # connection has been made and drbd transitions from WFConnection
2404
  # into any other network-connected state (Connected, SyncTarget,
2405
  # SyncSource, etc.)
2406

    
2407
  # 1st pass, assemble on all nodes in secondary mode
2408
  for inst_disk in instance.disks:
2409
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
2410
      lu.cfg.SetDiskID(node_disk, node)
2411
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, False)
2412
      if result.failed or not result:
2413
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
2414
                           " (is_primary=False, pass=1)",
2415
                           inst_disk.iv_name, node)
2416
        if not ignore_secondaries:
2417
          disks_ok = False
2418

    
2419
  # FIXME: race condition on drbd migration to primary
2420

    
2421
  # 2nd pass, do only the primary node
2422
  for inst_disk in instance.disks:
2423
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
2424
      if node != instance.primary_node:
2425
        continue
2426
      lu.cfg.SetDiskID(node_disk, node)
2427
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, True)
2428
      if result.failed or not result:
2429
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
2430
                           " (is_primary=True, pass=2)",
2431
                           inst_disk.iv_name, node)
2432
        disks_ok = False
2433
    device_info.append((instance.primary_node, inst_disk.iv_name, result.data))
2434

    
2435
  # leave the disks configured for the primary node
2436
  # this is a workaround that would be fixed better by
2437
  # improving the logical/physical id handling
2438
  for disk in instance.disks:
2439
    lu.cfg.SetDiskID(disk, instance.primary_node)
2440

    
2441
  return disks_ok, device_info
2442

    
2443

    
2444
def _StartInstanceDisks(lu, instance, force):
2445
  """Start the disks of an instance.
2446

2447
  """
2448
  disks_ok, dummy = _AssembleInstanceDisks(lu, instance,
2449
                                           ignore_secondaries=force)
2450
  if not disks_ok:
2451
    _ShutdownInstanceDisks(lu, instance)
2452
    if force is not None and not force:
2453
      lu.proc.LogWarning("", hint="If the message above refers to a"
2454
                         " secondary node,"
2455
                         " you can retry the operation using '--force'.")
2456
    raise errors.OpExecError("Disk consistency error")
2457

    
2458

    
2459
class LUDeactivateInstanceDisks(NoHooksLU):
2460
  """Shutdown an instance's disks.
2461

2462
  """
2463
  _OP_REQP = ["instance_name"]
2464
  REQ_BGL = False
2465

    
2466
  def ExpandNames(self):
2467
    self._ExpandAndLockInstance()
2468
    self.needed_locks[locking.LEVEL_NODE] = []
2469
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2470

    
2471
  def DeclareLocks(self, level):
2472
    if level == locking.LEVEL_NODE:
2473
      self._LockInstancesNodes()
2474

    
2475
  def CheckPrereq(self):
2476
    """Check prerequisites.
2477

2478
    This checks that the instance is in the cluster.
2479

2480
    """
2481
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2482
    assert self.instance is not None, \
2483
      "Cannot retrieve locked instance %s" % self.op.instance_name
2484

    
2485
  def Exec(self, feedback_fn):
2486
    """Deactivate the disks
2487

2488
    """
2489
    instance = self.instance
2490
    _SafeShutdownInstanceDisks(self, instance)
2491

    
2492

    
2493
def _SafeShutdownInstanceDisks(lu, instance):
2494
  """Shutdown block devices of an instance.
2495

2496
  This function checks if an instance is running, before calling
2497
  _ShutdownInstanceDisks.
2498

2499
  """
2500
  ins_l = lu.rpc.call_instance_list([instance.primary_node],
2501
                                      [instance.hypervisor])
2502
  ins_l = ins_l[instance.primary_node]
2503
  if ins_l.failed or not isinstance(ins_l.data, list):
2504
    raise errors.OpExecError("Can't contact node '%s'" %
2505
                             instance.primary_node)
2506

    
2507
  if instance.name in ins_l.data:
2508
    raise errors.OpExecError("Instance is running, can't shutdown"
2509
                             " block devices.")
2510

    
2511
  _ShutdownInstanceDisks(lu, instance)
2512

    
2513

    
2514
def _ShutdownInstanceDisks(lu, instance, ignore_primary=False):
2515
  """Shutdown block devices of an instance.
2516

2517
  This does the shutdown on all nodes of the instance.
2518

2519
  If the ignore_primary is false, errors on the primary node are
2520
  ignored.
2521

2522
  """
2523
  result = True
2524
  for disk in instance.disks:
2525
    for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
2526
      lu.cfg.SetDiskID(top_disk, node)
2527
      result = lu.rpc.call_blockdev_shutdown(node, top_disk)
2528
      if result.failed or not result.data:
2529
        logging.error("Could not shutdown block device %s on node %s",
2530
                      disk.iv_name, node)
2531
        if not ignore_primary or node != instance.primary_node:
2532
          result = False
2533
  return result
2534

    
2535

    
2536
def _CheckNodeFreeMemory(lu, node, reason, requested, hypervisor_name):
2537
  """Checks if a node has enough free memory.
2538

2539
  This function check if a given node has the needed amount of free
2540
  memory. In case the node has less memory or we cannot get the
2541
  information from the node, this function raise an OpPrereqError
2542
  exception.
2543

2544
  @type lu: C{LogicalUnit}
2545
  @param lu: a logical unit from which we get configuration data
2546
  @type node: C{str}
2547
  @param node: the node to check
2548
  @type reason: C{str}
2549
  @param reason: string to use in the error message
2550
  @type requested: C{int}
2551
  @param requested: the amount of memory in MiB to check for
2552
  @type hypervisor_name: C{str}
2553
  @param hypervisor_name: the hypervisor to ask for memory stats
2554
  @raise errors.OpPrereqError: if the node doesn't have enough memory, or
2555
      we cannot check the node
2556

2557
  """
2558
  nodeinfo = lu.rpc.call_node_info([node], lu.cfg.GetVGName(), hypervisor_name)
2559
  nodeinfo[node].Raise()
2560
  free_mem = nodeinfo[node].data.get('memory_free')
2561
  if not isinstance(free_mem, int):
2562
    raise errors.OpPrereqError("Can't compute free memory on node %s, result"
2563
                             " was '%s'" % (node, free_mem))
2564
  if requested > free_mem:
2565
    raise errors.OpPrereqError("Not enough memory on node %s for %s:"
2566
                             " needed %s MiB, available %s MiB" %
2567
                             (node, reason, requested, free_mem))
2568

    
2569

    
2570
class LUStartupInstance(LogicalUnit):
2571
  """Starts an instance.
2572

2573
  """
2574
  HPATH = "instance-start"
2575
  HTYPE = constants.HTYPE_INSTANCE
2576
  _OP_REQP = ["instance_name", "force"]
2577
  REQ_BGL = False
2578

    
2579
  def ExpandNames(self):
2580
    self._ExpandAndLockInstance()
2581

    
2582
  def BuildHooksEnv(self):
2583
    """Build hooks env.
2584

2585
    This runs on master, primary and secondary nodes of the instance.
2586

2587
    """
2588
    env = {
2589
      "FORCE": self.op.force,
2590
      }
2591
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
2592
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2593
    return env, nl, nl
2594

    
2595
  def CheckPrereq(self):
2596
    """Check prerequisites.
2597

2598
    This checks that the instance is in the cluster.
2599

2600
    """
2601
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2602
    assert self.instance is not None, \
2603
      "Cannot retrieve locked instance %s" % self.op.instance_name
2604

    
2605
    _CheckNodeOnline(self, instance.primary_node)
2606

    
2607
    bep = self.cfg.GetClusterInfo().FillBE(instance)
2608
    # check bridges existance
2609
    _CheckInstanceBridgesExist(self, instance)
2610

    
2611
    _CheckNodeFreeMemory(self, instance.primary_node,
2612
                         "starting instance %s" % instance.name,
2613
                         bep[constants.BE_MEMORY], instance.hypervisor)
2614

    
2615
  def Exec(self, feedback_fn):
2616
    """Start the instance.
2617

2618
    """
2619
    instance = self.instance
2620
    force = self.op.force
2621
    extra_args = getattr(self.op, "extra_args", "")
2622

    
2623
    self.cfg.MarkInstanceUp(instance.name)
2624

    
2625
    node_current = instance.primary_node
2626

    
2627
    _StartInstanceDisks(self, instance, force)
2628

    
2629
    result = self.rpc.call_instance_start(node_current, instance, extra_args)
2630
    if result.failed or not result.data:
2631
      _ShutdownInstanceDisks(self, instance)
2632
      raise errors.OpExecError("Could not start instance")
2633

    
2634

    
2635
class LURebootInstance(LogicalUnit):
2636
  """Reboot an instance.
2637

2638
  """
2639
  HPATH = "instance-reboot"
2640
  HTYPE = constants.HTYPE_INSTANCE
2641
  _OP_REQP = ["instance_name", "ignore_secondaries", "reboot_type"]
2642
  REQ_BGL = False
2643

    
2644
  def ExpandNames(self):
2645
    if self.op.reboot_type not in [constants.INSTANCE_REBOOT_SOFT,
2646
                                   constants.INSTANCE_REBOOT_HARD,
2647
                                   constants.INSTANCE_REBOOT_FULL]:
2648
      raise errors.ParameterError("reboot type not in [%s, %s, %s]" %
2649
                                  (constants.INSTANCE_REBOOT_SOFT,
2650
                                   constants.INSTANCE_REBOOT_HARD,
2651
                                   constants.INSTANCE_REBOOT_FULL))
2652
    self._ExpandAndLockInstance()
2653

    
2654
  def BuildHooksEnv(self):
2655
    """Build hooks env.
2656

2657
    This runs on master, primary and secondary nodes of the instance.
2658

2659
    """
2660
    env = {
2661
      "IGNORE_SECONDARIES": self.op.ignore_secondaries,
2662
      }
2663
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
2664
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2665
    return env, nl, nl
2666

    
2667
  def CheckPrereq(self):
2668
    """Check prerequisites.
2669

2670
    This checks that the instance is in the cluster.
2671

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

    
2677
    _CheckNodeOnline(self, instance.primary_node)
2678

    
2679
    # check bridges existance
2680
    _CheckInstanceBridgesExist(self, instance)
2681

    
2682
  def Exec(self, feedback_fn):
2683
    """Reboot the instance.
2684

2685
    """
2686
    instance = self.instance
2687
    ignore_secondaries = self.op.ignore_secondaries
2688
    reboot_type = self.op.reboot_type
2689
    extra_args = getattr(self.op, "extra_args", "")
2690

    
2691
    node_current = instance.primary_node
2692

    
2693
    if reboot_type in [constants.INSTANCE_REBOOT_SOFT,
2694
                       constants.INSTANCE_REBOOT_HARD]:
2695
      result = self.rpc.call_instance_reboot(node_current, instance,
2696
                                             reboot_type, extra_args)
2697
      if result.failed or not result.data:
2698
        raise errors.OpExecError("Could not reboot instance")
2699
    else:
2700
      if not self.rpc.call_instance_shutdown(node_current, instance):
2701
        raise errors.OpExecError("could not shutdown instance for full reboot")
2702
      _ShutdownInstanceDisks(self, instance)
2703
      _StartInstanceDisks(self, instance, ignore_secondaries)
2704
      result = self.rpc.call_instance_start(node_current, instance, extra_args)
2705
      if result.failed or not result.data:
2706
        _ShutdownInstanceDisks(self, instance)
2707
        raise errors.OpExecError("Could not start instance for full reboot")
2708

    
2709
    self.cfg.MarkInstanceUp(instance.name)
2710

    
2711

    
2712
class LUShutdownInstance(LogicalUnit):
2713
  """Shutdown an instance.
2714

2715
  """
2716
  HPATH = "instance-stop"
2717
  HTYPE = constants.HTYPE_INSTANCE
2718
  _OP_REQP = ["instance_name"]
2719
  REQ_BGL = False
2720

    
2721
  def ExpandNames(self):
2722
    self._ExpandAndLockInstance()
2723

    
2724
  def BuildHooksEnv(self):
2725
    """Build hooks env.
2726

2727
    This runs on master, primary and secondary nodes of the instance.
2728

2729
    """
2730
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2731
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2732
    return env, nl, nl
2733

    
2734
  def CheckPrereq(self):
2735
    """Check prerequisites.
2736

2737
    This checks that the instance is in the cluster.
2738

2739
    """
2740
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2741
    assert self.instance is not None, \
2742
      "Cannot retrieve locked instance %s" % self.op.instance_name
2743
    _CheckNodeOnline(self, self.instance.primary_node)
2744

    
2745
  def Exec(self, feedback_fn):
2746
    """Shutdown the instance.
2747

2748
    """
2749
    instance = self.instance
2750
    node_current = instance.primary_node
2751
    self.cfg.MarkInstanceDown(instance.name)
2752
    result = self.rpc.call_instance_shutdown(node_current, instance)
2753
    if result.failed or not result.data:
2754
      self.proc.LogWarning("Could not shutdown instance")
2755

    
2756
    _ShutdownInstanceDisks(self, instance)
2757

    
2758

    
2759
class LUReinstallInstance(LogicalUnit):
2760
  """Reinstall an instance.
2761

2762
  """
2763
  HPATH = "instance-reinstall"
2764
  HTYPE = constants.HTYPE_INSTANCE
2765
  _OP_REQP = ["instance_name"]
2766
  REQ_BGL = False
2767

    
2768
  def ExpandNames(self):
2769
    self._ExpandAndLockInstance()
2770

    
2771
  def BuildHooksEnv(self):
2772
    """Build hooks env.
2773

2774
    This runs on master, primary and secondary nodes of the instance.
2775

2776
    """
2777
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2778
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2779
    return env, nl, nl
2780

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

2784
    This checks that the instance is in the cluster and is not running.
2785

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

    
2792
    if instance.disk_template == constants.DT_DISKLESS:
2793
      raise errors.OpPrereqError("Instance '%s' has no disks" %
2794
                                 self.op.instance_name)
2795
    if instance.status != "down":
2796
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2797
                                 self.op.instance_name)
2798
    remote_info = self.rpc.call_instance_info(instance.primary_node,
2799
                                              instance.name,
2800
                                              instance.hypervisor)
2801
    if remote_info.failed or remote_info.data:
2802
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
2803
                                 (self.op.instance_name,
2804
                                  instance.primary_node))
2805

    
2806
    self.op.os_type = getattr(self.op, "os_type", None)
2807
    if self.op.os_type is not None:
2808
      # OS verification
2809
      pnode = self.cfg.GetNodeInfo(
2810
        self.cfg.ExpandNodeName(instance.primary_node))
2811
      if pnode is None:
2812
        raise errors.OpPrereqError("Primary node '%s' is unknown" %
2813
                                   self.op.pnode)
2814
      result = self.rpc.call_os_get(pnode.name, self.op.os_type)
2815
      result.Raise()
2816
      if not isinstance(result.data, objects.OS):
2817
        raise errors.OpPrereqError("OS '%s' not in supported OS list for"
2818
                                   " primary node"  % self.op.os_type)
2819

    
2820
    self.instance = instance
2821

    
2822
  def Exec(self, feedback_fn):
2823
    """Reinstall the instance.
2824

2825
    """
2826
    inst = self.instance
2827

    
2828
    if self.op.os_type is not None:
2829
      feedback_fn("Changing OS to '%s'..." % self.op.os_type)
2830
      inst.os = self.op.os_type
2831
      self.cfg.Update(inst)
2832

    
2833
    _StartInstanceDisks(self, inst, None)
2834
    try:
2835
      feedback_fn("Running the instance OS create scripts...")
2836
      result = self.rpc.call_instance_os_add(inst.primary_node, inst)
2837
      result.Raise()
2838
      if not result.data:
2839
        raise errors.OpExecError("Could not install OS for instance %s"
2840
                                 " on node %s" %
2841
                                 (inst.name, inst.primary_node))
2842
    finally:
2843
      _ShutdownInstanceDisks(self, inst)
2844

    
2845

    
2846
class LURenameInstance(LogicalUnit):
2847
  """Rename an instance.
2848

2849
  """
2850
  HPATH = "instance-rename"
2851
  HTYPE = constants.HTYPE_INSTANCE
2852
  _OP_REQP = ["instance_name", "new_name"]
2853

    
2854
  def BuildHooksEnv(self):
2855
    """Build hooks env.
2856

2857
    This runs on master, primary and secondary nodes of the instance.
2858

2859
    """
2860
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2861
    env["INSTANCE_NEW_NAME"] = self.op.new_name
2862
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2863
    return env, nl, nl
2864

    
2865
  def CheckPrereq(self):
2866
    """Check prerequisites.
2867

2868
    This checks that the instance is in the cluster and is not running.
2869

2870
    """
2871
    instance = self.cfg.GetInstanceInfo(
2872
      self.cfg.ExpandInstanceName(self.op.instance_name))
2873
    if instance is None:
2874
      raise errors.OpPrereqError("Instance '%s' not known" %
2875
                                 self.op.instance_name)
2876
    _CheckNodeOnline(self, instance.primary_node)
2877

    
2878
    if instance.status != "down":
2879
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2880
                                 self.op.instance_name)
2881
    remote_info = self.rpc.call_instance_info(instance.primary_node,
2882
                                              instance.name,
2883
                                              instance.hypervisor)
2884
    remote_info.Raise()
2885
    if remote_info.data:
2886
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
2887
                                 (self.op.instance_name,
2888
                                  instance.primary_node))
2889
    self.instance = instance
2890

    
2891
    # new name verification
2892
    name_info = utils.HostInfo(self.op.new_name)
2893

    
2894
    self.op.new_name = new_name = name_info.name
2895
    instance_list = self.cfg.GetInstanceList()
2896
    if new_name in instance_list:
2897
      raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
2898
                                 new_name)
2899

    
2900
    if not getattr(self.op, "ignore_ip", False):
2901
      if utils.TcpPing(name_info.ip, constants.DEFAULT_NODED_PORT):
2902
        raise errors.OpPrereqError("IP %s of instance %s already in use" %
2903
                                   (name_info.ip, new_name))
2904

    
2905

    
2906
  def Exec(self, feedback_fn):
2907
    """Reinstall the instance.
2908

2909
    """
2910
    inst = self.instance
2911
    old_name = inst.name
2912

    
2913
    if inst.disk_template == constants.DT_FILE:
2914
      old_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
2915

    
2916
    self.cfg.RenameInstance(inst.name, self.op.new_name)
2917
    # Change the instance lock. This is definitely safe while we hold the BGL
2918
    self.context.glm.remove(locking.LEVEL_INSTANCE, old_name)
2919
    self.context.glm.add(locking.LEVEL_INSTANCE, self.op.new_name)
2920

    
2921
    # re-read the instance from the configuration after rename
2922
    inst = self.cfg.GetInstanceInfo(self.op.new_name)
2923

    
2924
    if inst.disk_template == constants.DT_FILE:
2925
      new_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
2926
      result = self.rpc.call_file_storage_dir_rename(inst.primary_node,
2927
                                                     old_file_storage_dir,
2928
                                                     new_file_storage_dir)
2929
      result.Raise()
2930
      if not result.data:
2931
        raise errors.OpExecError("Could not connect to node '%s' to rename"
2932
                                 " directory '%s' to '%s' (but the instance"
2933
                                 " has been renamed in Ganeti)" % (
2934
                                 inst.primary_node, old_file_storage_dir,
2935
                                 new_file_storage_dir))
2936

    
2937
      if not result.data[0]:
2938
        raise errors.OpExecError("Could not rename directory '%s' to '%s'"
2939
                                 " (but the instance has been renamed in"
2940
                                 " Ganeti)" % (old_file_storage_dir,
2941
                                               new_file_storage_dir))
2942

    
2943
    _StartInstanceDisks(self, inst, None)
2944
    try:
2945
      result = self.rpc.call_instance_run_rename(inst.primary_node, inst,
2946
                                                 old_name)
2947
      if result.failed or not result.data:
2948
        msg = ("Could not run OS rename script for instance %s on node %s"
2949
               " (but the instance has been renamed in Ganeti)" %
2950
               (inst.name, inst.primary_node))
2951
        self.proc.LogWarning(msg)
2952
    finally:
2953
      _ShutdownInstanceDisks(self, inst)
2954

    
2955

    
2956
class LURemoveInstance(LogicalUnit):
2957
  """Remove an instance.
2958

2959
  """
2960
  HPATH = "instance-remove"
2961
  HTYPE = constants.HTYPE_INSTANCE
2962
  _OP_REQP = ["instance_name", "ignore_failures"]
2963
  REQ_BGL = False
2964

    
2965
  def ExpandNames(self):
2966
    self._ExpandAndLockInstance()
2967
    self.needed_locks[locking.LEVEL_NODE] = []
2968
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2969

    
2970
  def DeclareLocks(self, level):
2971
    if level == locking.LEVEL_NODE:
2972
      self._LockInstancesNodes()
2973

    
2974
  def BuildHooksEnv(self):
2975
    """Build hooks env.
2976

2977
    This runs on master, primary and secondary nodes of the instance.
2978

2979
    """
2980
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2981
    nl = [self.cfg.GetMasterNode()]
2982
    return env, nl, nl
2983

    
2984
  def CheckPrereq(self):
2985
    """Check prerequisites.
2986

2987
    This checks that the instance is in the cluster.
2988

2989
    """
2990
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2991
    assert self.instance is not None, \
2992
      "Cannot retrieve locked instance %s" % self.op.instance_name
2993

    
2994
  def Exec(self, feedback_fn):
2995
    """Remove the instance.
2996

2997
    """
2998
    instance = self.instance
2999
    logging.info("Shutting down instance %s on node %s",
3000
                 instance.name, instance.primary_node)
3001

    
3002
    result = self.rpc.call_instance_shutdown(instance.primary_node, instance)
3003
    if result.failed or not result.data:
3004
      if self.op.ignore_failures:
3005
        feedback_fn("Warning: can't shutdown instance")
3006
      else:
3007
        raise errors.OpExecError("Could not shutdown instance %s on node %s" %
3008
                                 (instance.name, instance.primary_node))
3009

    
3010
    logging.info("Removing block devices for instance %s", instance.name)
3011

    
3012
    if not _RemoveDisks(self, instance):
3013
      if self.op.ignore_failures:
3014
        feedback_fn("Warning: can't remove instance's disks")
3015
      else:
3016
        raise errors.OpExecError("Can't remove instance's disks")
3017

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

    
3020
    self.cfg.RemoveInstance(instance.name)
3021
    self.remove_locks[locking.LEVEL_INSTANCE] = instance.name
3022

    
3023

    
3024
class LUQueryInstances(NoHooksLU):
3025
  """Logical unit for querying instances.
3026

3027
  """
3028
  _OP_REQP = ["output_fields", "names"]
3029
  REQ_BGL = False
3030
  _FIELDS_STATIC = utils.FieldSet(*["name", "os", "pnode", "snodes",
3031
                                    "admin_state", "admin_ram",
3032
                                    "disk_template", "ip", "mac", "bridge",
3033
                                    "sda_size", "sdb_size", "vcpus", "tags",
3034
                                    "network_port", "beparams",
3035
                                    "(disk).(size)/([0-9]+)",
3036
                                    "(disk).(sizes)",
3037
                                    "(nic).(mac|ip|bridge)/([0-9]+)",
3038
                                    "(nic).(macs|ips|bridges)",
3039
                                    "(disk|nic).(count)",
3040
                                    "serial_no", "hypervisor", "hvparams",] +
3041
                                  ["hv/%s" % name
3042
                                   for name in constants.HVS_PARAMETERS] +
3043
                                  ["be/%s" % name
3044
                                   for name in constants.BES_PARAMETERS])
3045
  _FIELDS_DYNAMIC = utils.FieldSet("oper_state", "oper_ram", "status")
3046

    
3047

    
3048
  def ExpandNames(self):
3049
    _CheckOutputFields(static=self._FIELDS_STATIC,
3050
                       dynamic=self._FIELDS_DYNAMIC,
3051
                       selected=self.op.output_fields)
3052

    
3053
    self.needed_locks = {}
3054
    self.share_locks[locking.LEVEL_INSTANCE] = 1
3055
    self.share_locks[locking.LEVEL_NODE] = 1
3056

    
3057
    if self.op.names:
3058
      self.wanted = _GetWantedInstances(self, self.op.names)
3059
    else:
3060
      self.wanted = locking.ALL_SET
3061

    
3062
    self.do_locking = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
3063
    if self.do_locking:
3064
      self.needed_locks[locking.LEVEL_INSTANCE] = self.wanted
3065
      self.needed_locks[locking.LEVEL_NODE] = []
3066
      self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3067

    
3068
  def DeclareLocks(self, level):
3069
    if level == locking.LEVEL_NODE and self.do_locking:
3070
      self._LockInstancesNodes()
3071

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

3075
    """
3076
    pass
3077

    
3078
  def Exec(self, feedback_fn):
3079
    """Computes the list of nodes and their attributes.
3080

3081
    """
3082
    all_info = self.cfg.GetAllInstancesInfo()
3083
    if self.do_locking:
3084
      instance_names = self.acquired_locks[locking.LEVEL_INSTANCE]
3085
    elif self.wanted != locking.ALL_SET:
3086
      instance_names = self.wanted
3087
      missing = set(instance_names).difference(all_info.keys())
3088
      if missing:
3089
        raise errors.OpExecError(
3090
          "Some instances were removed before retrieving their data: %s"
3091
          % missing)
3092
    else:
3093
      instance_names = all_info.keys()
3094

    
3095
    instance_names = utils.NiceSort(instance_names)
3096
    instance_list = [all_info[iname] for iname in instance_names]
3097

    
3098
    # begin data gathering
3099

    
3100
    nodes = frozenset([inst.primary_node for inst in instance_list])
3101
    hv_list = list(set([inst.hypervisor for inst in instance_list]))
3102

    
3103
    bad_nodes = []
3104
    off_nodes = []
3105
    if self.do_locking:
3106
      live_data = {}
3107
      node_data = self.rpc.call_all_instances_info(nodes, hv_list)
3108
      for name in nodes:
3109
        result = node_data[name]
3110
        if result.offline:
3111
          # offline nodes will be in both lists
3112
          off_nodes.append(name)
3113
        if result.failed:
3114
          bad_nodes.append(name)
3115
        else:
3116
          if result.data:
3117
            live_data.update(result.data)
3118
            # else no instance is alive
3119
    else:
3120
      live_data = dict([(name, {}) for name in instance_names])
3121

    
3122
    # end data gathering
3123

    
3124
    HVPREFIX = "hv/"
3125
    BEPREFIX = "be/"
3126
    output = []
3127
    for instance in instance_list:
3128
      iout = []
3129
      i_hv = self.cfg.GetClusterInfo().FillHV(instance)
3130
      i_be = self.cfg.GetClusterInfo().FillBE(instance)
3131
      for field in self.op.output_fields:
3132
        st_match = self._FIELDS_STATIC.Matches(field)
3133
        if field == "name":
3134
          val = instance.name
3135
        elif field == "os":
3136
          val = instance.os
3137
        elif field == "pnode":
3138
          val = instance.primary_node
3139
        elif field == "snodes":
3140
          val = list(instance.secondary_nodes)
3141
        elif field == "admin_state":
3142
          val = (instance.status != "down")
3143
        elif field == "oper_state":
3144
          if instance.primary_node in bad_nodes:
3145
            val = None
3146
          else:
3147
            val = bool(live_data.get(instance.name))
3148
        elif field == "status":
3149
          if instance.primary_node in off_nodes:
3150
            val = "ERROR_nodeoffline"
3151
          elif instance.primary_node in bad_nodes:
3152
            val = "ERROR_nodedown"
3153
          else:
3154
            running = bool(live_data.get(instance.name))
3155
            if running:
3156
              if instance.status != "down":
3157
                val = "running"
3158
              else:
3159
                val = "ERROR_up"
3160
            else:
3161
              if instance.status != "down":
3162
                val = "ERROR_down"
3163
              else:
3164
                val = "ADMIN_down"
3165
        elif field == "oper_ram":
3166
          if instance.primary_node in bad_nodes:
3167
            val = None
3168
          elif instance.name in live_data:
3169
            val = live_data[instance.name].get("memory", "?")
3170
          else:
3171
            val = "-"
3172
        elif field == "disk_template":
3173
          val = instance.disk_template
3174
        elif field == "ip":
3175
          val = instance.nics[0].ip
3176
        elif field == "bridge":
3177
          val = instance.nics[0].bridge
3178
        elif field == "mac":
3179
          val = instance.nics[0].mac
3180
        elif field == "sda_size" or field == "sdb_size":
3181
          idx = ord(field[2]) - ord('a')
3182
          try:
3183
            val = instance.FindDisk(idx).size
3184
          except errors.OpPrereqError:
3185
            val = None
3186
        elif field == "tags":
3187
          val = list(instance.GetTags())
3188
        elif field == "serial_no":
3189
          val = instance.serial_no
3190
        elif field == "network_port":
3191
          val = instance.network_port
3192
        elif field == "hypervisor":
3193
          val = instance.hypervisor
3194
        elif field == "hvparams":
3195
          val = i_hv
3196
        elif (field.startswith(HVPREFIX) and
3197
              field[len(HVPREFIX):] in constants.HVS_PARAMETERS):
3198
          val = i_hv.get(field[len(HVPREFIX):], None)
3199
        elif field == "beparams":
3200
          val = i_be
3201
        elif (field.startswith(BEPREFIX) and
3202
              field[len(BEPREFIX):] in constants.BES_PARAMETERS):
3203
          val = i_be.get(field[len(BEPREFIX):], None)
3204
        elif st_match and st_match.groups():
3205
          # matches a variable list
3206
          st_groups = st_match.groups()
3207
          if st_groups and st_groups[0] == "disk":
3208
            if st_groups[1] == "count":
3209
              val = len(instance.disks)
3210
            elif st_groups[1] == "sizes":
3211
              val = [disk.size for disk in instance.disks]
3212
            elif st_groups[1] == "size":
3213
              try:
3214
                val = instance.FindDisk(st_groups[2]).size
3215
              except errors.OpPrereqError:
3216
                val = None
3217
            else:
3218
              assert False, "Unhandled disk parameter"
3219
          elif st_groups[0] == "nic":
3220
            if st_groups[1] == "count":
3221
              val = len(instance.nics)
3222
            elif st_groups[1] == "macs":
3223
              val = [nic.mac for nic in instance.nics]
3224
            elif st_groups[1] == "ips":
3225
              val = [nic.ip for nic in instance.nics]
3226
            elif st_groups[1] == "bridges":
3227
              val = [nic.bridge for nic in instance.nics]
3228
            else:
3229
              # index-based item
3230
              nic_idx = int(st_groups[2])
3231
              if nic_idx >= len(instance.nics):
3232
                val = None
3233
              else:
3234
                if st_groups[1] == "mac":
3235
                  val = instance.nics[nic_idx].mac
3236
                elif st_groups[1] == "ip":
3237
                  val = instance.nics[nic_idx].ip
3238
                elif st_groups[1] == "bridge":
3239
                  val = instance.nics[nic_idx].bridge
3240
                else:
3241
                  assert False, "Unhandled NIC parameter"
3242
          else:
3243
            assert False, "Unhandled variable parameter"
3244
        else:
3245
          raise errors.ParameterError(field)
3246
        iout.append(val)
3247
      output.append(iout)
3248

    
3249
    return output
3250

    
3251

    
3252
class LUFailoverInstance(LogicalUnit):
3253
  """Failover an instance.
3254

3255
  """
3256
  HPATH = "instance-failover"
3257
  HTYPE = constants.HTYPE_INSTANCE
3258
  _OP_REQP = ["instance_name", "ignore_consistency"]
3259
  REQ_BGL = False
3260

    
3261
  def ExpandNames(self):
3262
    self._ExpandAndLockInstance()
3263
    self.needed_locks[locking.LEVEL_NODE] = []
3264
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3265

    
3266
  def DeclareLocks(self, level):
3267
    if level == locking.LEVEL_NODE:
3268
      self._LockInstancesNodes()
3269

    
3270
  def BuildHooksEnv(self):
3271
    """Build hooks env.
3272

3273
    This runs on master, primary and secondary nodes of the instance.
3274

3275
    """
3276
    env = {
3277
      "IGNORE_CONSISTENCY": self.op.ignore_consistency,
3278
      }
3279
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3280
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3281
    return env, nl, nl
3282

    
3283
  def CheckPrereq(self):
3284
    """Check prerequisites.
3285

3286
    This checks that the instance is in the cluster.
3287

3288
    """
3289
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3290
    assert self.instance is not None, \
3291
      "Cannot retrieve locked instance %s" % self.op.instance_name
3292

    
3293
    bep = self.cfg.GetClusterInfo().FillBE(instance)
3294
    if instance.disk_template not in constants.DTS_NET_MIRROR:
3295
      raise errors.OpPrereqError("Instance's disk layout is not"
3296
                                 " network mirrored, cannot failover.")
3297

    
3298
    secondary_nodes = instance.secondary_nodes
3299
    if not secondary_nodes:
3300
      raise errors.ProgrammerError("no secondary node but using "
3301
                                   "a mirrored disk template")
3302

    
3303
    target_node = secondary_nodes[0]
3304
    _CheckNodeOnline(self, target_node)
3305
    # check memory requirements on the secondary node
3306
    _CheckNodeFreeMemory(self, target_node, "failing over instance %s" %
3307
                         instance.name, bep[constants.BE_MEMORY],
3308
                         instance.hypervisor)
3309

    
3310
    # check bridge existance
3311
    brlist = [nic.bridge for nic in instance.nics]
3312
    result = self.rpc.call_bridges_exist(target_node, brlist)
3313
    result.Raise()
3314
    if not result.data:
3315
      raise errors.OpPrereqError("One or more target bridges %s does not"
3316
                                 " exist on destination node '%s'" %
3317
                                 (brlist, target_node))
3318

    
3319
  def Exec(self, feedback_fn):
3320
    """Failover an instance.
3321

3322
    The failover is done by shutting it down on its present node and
3323
    starting it on the secondary.
3324

3325
    """
3326
    instance = self.instance
3327

    
3328
    source_node = instance.primary_node
3329
    target_node = instance.secondary_nodes[0]
3330

    
3331
    feedback_fn("* checking disk consistency between source and target")
3332
    for dev in instance.disks:
3333
      # for drbd, these are drbd over lvm
3334
      if not _CheckDiskConsistency(self, dev, target_node, False):
3335
        if instance.status == "up" and not self.op.ignore_consistency:
3336
          raise errors.OpExecError("Disk %s is degraded on target node,"
3337
                                   " aborting failover." % dev.iv_name)
3338

    
3339
    feedback_fn("* shutting down instance on source node")
3340
    logging.info("Shutting down instance %s on node %s",
3341
                 instance.name, source_node)
3342

    
3343
    result = self.rpc.call_instance_shutdown(source_node, instance)
3344
    if result.failed or not result.data:
3345
      if self.op.ignore_consistency:
3346
        self.proc.LogWarning("Could not shutdown instance %s on node %s."
3347
                             " Proceeding"
3348
                             " anyway. Please make sure node %s is down",
3349
                             instance.name, source_node, source_node)
3350
      else:
3351
        raise errors.OpExecError("Could not shutdown instance %s on node %s" %
3352
                                 (instance.name, source_node))
3353

    
3354
    feedback_fn("* deactivating the instance's disks on source node")
3355
    if not _ShutdownInstanceDisks(self, instance, ignore_primary=True):
3356
      raise errors.OpExecError("Can't shut down the instance's disks.")
3357

    
3358
    instance.primary_node = target_node
3359
    # distribute new instance config to the other nodes
3360
    self.cfg.Update(instance)
3361

    
3362
    # Only start the instance if it's marked as up
3363
    if instance.status == "up":
3364
      feedback_fn("* activating the instance's disks on target node")
3365
      logging.info("Starting instance %s on node %s",
3366
                   instance.name, target_node)
3367

    
3368
      disks_ok, dummy = _AssembleInstanceDisks(self, instance,
3369
                                               ignore_secondaries=True)
3370
      if not disks_ok:
3371
        _ShutdownInstanceDisks(self, instance)
3372
        raise errors.OpExecError("Can't activate the instance's disks")
3373

    
3374
      feedback_fn("* starting the instance on the target node")
3375
      result = self.rpc.call_instance_start(target_node, instance, None)
3376
      if result.failed or not result.data:
3377
        _ShutdownInstanceDisks(self, instance)
3378
        raise errors.OpExecError("Could not start instance %s on node %s." %
3379
                                 (instance.name, target_node))
3380

    
3381

    
3382
class LUMigrateInstance(LogicalUnit):
3383
  """Migrate an instance.
3384

3385
  This is migration without shutting down, compared to the failover,
3386
  which is done with shutdown.
3387

3388
  """
3389
  HPATH = "instance-migrate"
3390
  HTYPE = constants.HTYPE_INSTANCE
3391
  _OP_REQP = ["instance_name", "live", "cleanup"]
3392

    
3393
  REQ_BGL = False
3394

    
3395
  def ExpandNames(self):
3396
    self._ExpandAndLockInstance()
3397
    self.needed_locks[locking.LEVEL_NODE] = []
3398
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3399

    
3400
  def DeclareLocks(self, level):
3401
    if level == locking.LEVEL_NODE:
3402
      self._LockInstancesNodes()
3403

    
3404
  def BuildHooksEnv(self):
3405
    """Build hooks env.
3406

3407
    This runs on master, primary and secondary nodes of the instance.
3408

3409
    """
3410
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3411
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3412
    return env, nl, nl
3413

    
3414
  def CheckPrereq(self):
3415
    """Check prerequisites.
3416

3417
    This checks that the instance is in the cluster.
3418

3419
    """
3420
    instance = self.cfg.GetInstanceInfo(
3421
      self.cfg.ExpandInstanceName(self.op.instance_name))
3422
    if instance is None:
3423
      raise errors.OpPrereqError("Instance '%s' not known" %
3424
                                 self.op.instance_name)
3425

    
3426
    if instance.disk_template != constants.DT_DRBD8:
3427
      raise errors.OpPrereqError("Instance's disk layout is not"
3428
                                 " drbd8, cannot migrate.")
3429

    
3430
    secondary_nodes = instance.secondary_nodes
3431
    if not secondary_nodes:
3432
      raise errors.ProgrammerError("no secondary node but using "
3433
                                   "drbd8 disk template")
3434

    
3435
    i_be = self.cfg.GetClusterInfo().FillBE(instance)
3436

    
3437
    target_node = secondary_nodes[0]
3438
    # check memory requirements on the secondary node
3439
    _CheckNodeFreeMemory(self, target_node, "migrating instance %s" %
3440
                         instance.name, i_be[constants.BE_MEMORY],
3441
                         instance.hypervisor)
3442

    
3443
    # check bridge existance
3444
    brlist = [nic.bridge for nic in instance.nics]
3445
    result = self.rpc.call_bridges_exist(target_node, brlist)
3446
    if result.failed or not result.data:
3447
      raise errors.OpPrereqError("One or more target bridges %s does not"
3448
                                 " exist on destination node '%s'" %
3449
                                 (brlist, target_node))
3450

    
3451
    if not self.op.cleanup:
3452
      result = self.rpc.call_instance_migratable(instance.primary_node,
3453
                                                 instance)
3454
      msg = result.RemoteFailMsg()
3455
      if msg:
3456
        raise errors.OpPrereqError("Can't migrate: %s - please use failover" %
3457
                                   msg)
3458

    
3459
    self.instance = instance
3460

    
3461
  def _WaitUntilSync(self):
3462
    """Poll with custom rpc for disk sync.
3463

3464
    This uses our own step-based rpc call.
3465

3466
    """
3467
    self.feedback_fn("* wait until resync is done")
3468
    all_done = False
3469
    while not all_done:
3470
      all_done = True
3471
      result = self.rpc.call_drbd_wait_sync(self.all_nodes,
3472
                                            self.nodes_ip,
3473
                                            self.instance.disks)
3474
      min_percent = 100
3475
      for node, nres in result.items():
3476
        msg = nres.RemoteFailMsg()
3477
        if msg:
3478
          raise errors.OpExecError("Cannot resync disks on node %s: %s" %
3479
                                   (node, msg))
3480
        node_done, node_percent = nres.data[1]
3481
        all_done = all_done and node_done
3482
        if node_percent is not None:
3483
          min_percent = min(min_percent, node_percent)
3484
      if not all_done:
3485
        if min_percent < 100:
3486
          self.feedback_fn("   - progress: %.1f%%" % min_percent)
3487
        time.sleep(2)
3488

    
3489
  def _EnsureSecondary(self, node):
3490
    """Demote a node to secondary.
3491

3492
    """
3493
    self.feedback_fn("* switching node %s to secondary mode" % node)
3494

    
3495
    for dev in self.instance.disks:
3496
      self.cfg.SetDiskID(dev, node)
3497

    
3498
    result = self.rpc.call_blockdev_close(node, self.instance.name,
3499
                                          self.instance.disks)
3500
    msg = result.RemoteFailMsg()
3501
    if msg:
3502
      raise errors.OpExecError("Cannot change disk to secondary on node %s,"
3503
                               " error %s" % (node, msg))
3504

    
3505
  def _GoStandalone(self):
3506
    """Disconnect from the network.
3507

3508
    """
3509
    self.feedback_fn("* changing into standalone mode")
3510
    result = self.rpc.call_drbd_disconnect_net(self.all_nodes, self.nodes_ip,
3511
                                               self.instance.disks)
3512
    for node, nres in result.items():
3513
      msg = nres.RemoteFailMsg()
3514
      if msg:
3515
        raise errors.OpExecError("Cannot disconnect disks node %s,"
3516
                                 " error %s" % (node, msg))
3517

    
3518
  def _GoReconnect(self, multimaster):
3519
    """Reconnect to the network.
3520

3521
    """
3522
    if multimaster:
3523
      msg = "dual-master"
3524
    else:
3525
      msg = "single-master"
3526
    self.feedback_fn("* changing disks into %s mode" % msg)
3527
    result = self.rpc.call_drbd_attach_net(self.all_nodes, self.nodes_ip,
3528
                                           self.instance.disks,
3529
                                           self.instance.name, multimaster)
3530
    for node, nres in result.items():
3531
      msg = nres.RemoteFailMsg()
3532
      if msg:
3533
        raise errors.OpExecError("Cannot change disks config on node %s,"
3534
                                 " error: %s" % (node, msg))
3535

    
3536
  def _ExecCleanup(self):
3537
    """Try to cleanup after a failed migration.
3538

3539
    The cleanup is done by:
3540
      - check that the instance is running only on one node
3541
        (and update the config if needed)
3542
      - change disks on its secondary node to secondary
3543
      - wait until disks are fully synchronized
3544
      - disconnect from the network
3545
      - change disks into single-master mode
3546
      - wait again until disks are fully synchronized
3547

3548
    """
3549
    instance = self.instance
3550
    target_node = self.target_node
3551
    source_node = self.source_node
3552

    
3553
    # check running on only one node
3554
    self.feedback_fn("* checking where the instance actually runs"
3555
                     " (if this hangs, the hypervisor might be in"
3556
                     " a bad state)")
3557
    ins_l = self.rpc.call_instance_list(self.all_nodes, [instance.hypervisor])
3558
    for node, result in ins_l.items():
3559
      result.Raise()
3560
      if not isinstance(result.data, list):
3561
        raise errors.OpExecError("Can't contact node '%s'" % node)
3562

    
3563
    runningon_source = instance.name in ins_l[source_node].data
3564
    runningon_target = instance.name in ins_l[target_node].data
3565

    
3566
    if runningon_source and runningon_target:
3567
      raise errors.OpExecError("Instance seems to be running on two nodes,"
3568
                               " or the hypervisor is confused. You will have"
3569
                               " to ensure manually that it runs only on one"
3570
                               " and restart this operation.")
3571

    
3572
    if not (runningon_source or runningon_target):
3573
      raise errors.OpExecError("Instance does not seem to be running at all."
3574
                               " In this case, it's safer to repair by"
3575
                               " running 'gnt-instance stop' to ensure disk"
3576
                               " shutdown, and then restarting it.")
3577

    
3578
    if runningon_target:
3579
      # the migration has actually succeeded, we need to update the config
3580
      self.feedback_fn("* instance running on secondary node (%s),"
3581
                       " updating config" % target_node)
3582
      instance.primary_node = target_node
3583
      self.cfg.Update(instance)
3584
      demoted_node = source_node
3585
    else:
3586
      self.feedback_fn("* instance confirmed to be running on its"
3587
                       " primary node (%s)" % source_node)
3588
      demoted_node = target_node
3589

    
3590
    self._EnsureSecondary(demoted_node)
3591
    try:
3592
      self._WaitUntilSync()
3593
    except errors.OpExecError:
3594
      # we ignore here errors, since if the device is standalone, it
3595
      # won't be able to sync
3596
      pass
3597
    self._GoStandalone()
3598
    self._GoReconnect(False)
3599
    self._WaitUntilSync()
3600

    
3601
    self.feedback_fn("* done")
3602

    
3603
  def _ExecMigration(self):
3604
    """Migrate an instance.
3605

3606
    The migrate is done by:
3607
      - change the disks into dual-master mode
3608
      - wait until disks are fully synchronized again
3609
      - migrate the instance
3610
      - change disks on the new secondary node (the old primary) to secondary
3611
      - wait until disks are fully synchronized
3612
      - change disks into single-master mode
3613

3614
    """
3615
    instance = self.instance
3616
    target_node = self.target_node
3617
    source_node = self.source_node
3618

    
3619
    self.feedback_fn("* checking disk consistency between source and target")
3620
    for dev in instance.disks:
3621
      if not _CheckDiskConsistency(self, dev, target_node, False):
3622
        raise errors.OpExecError("Disk %s is degraded or not fully"
3623
                                 " synchronized on target node,"
3624
                                 " aborting migrate." % dev.iv_name)
3625

    
3626
    self._EnsureSecondary(target_node)
3627
    self._GoStandalone()
3628
    self._GoReconnect(True)
3629
    self._WaitUntilSync()
3630

    
3631
    self.feedback_fn("* migrating instance to %s" % target_node)
3632
    time.sleep(10)
3633
    result = self.rpc.call_instance_migrate(source_node, instance,
3634
                                            self.nodes_ip[target_node],
3635
                                            self.op.live)
3636
    msg = result.RemoteFailMsg()
3637
    if msg:
3638
      logging.error("Instance migration failed, trying to revert"
3639
                    " disk status: %s", msg)
3640
      try:
3641
        self._EnsureSecondary(target_node)
3642
        self._GoStandalone()
3643
        self._GoReconnect(False)
3644
        self._WaitUntilSync()
3645
      except errors.OpExecError, err:
3646
        self.LogWarning("Migration failed and I can't reconnect the"
3647
                        " drives: error '%s'\n"
3648
                        "Please look and recover the instance status" %
3649
                        str(err))
3650

    
3651
      raise errors.OpExecError("Could not migrate instance %s: %s" %
3652
                               (instance.name, msg))
3653
    time.sleep(10)
3654

    
3655
    instance.primary_node = target_node
3656
    # distribute new instance config to the other nodes
3657
    self.cfg.Update(instance)
3658

    
3659
    self._EnsureSecondary(source_node)
3660
    self._WaitUntilSync()
3661
    self._GoStandalone()
3662
    self._GoReconnect(False)
3663
    self._WaitUntilSync()
3664

    
3665
    self.feedback_fn("* done")
3666

    
3667
  def Exec(self, feedback_fn):
3668
    """Perform the migration.
3669

3670
    """
3671
    self.feedback_fn = feedback_fn
3672

    
3673
    self.source_node = self.instance.primary_node
3674
    self.target_node = self.instance.secondary_nodes[0]
3675
    self.all_nodes = [self.source_node, self.target_node]
3676
    self.nodes_ip = {
3677
      self.source_node: self.cfg.GetNodeInfo(self.source_node).secondary_ip,
3678
      self.target_node: self.cfg.GetNodeInfo(self.target_node).secondary_ip,
3679
      }
3680
    if self.op.cleanup:
3681
      return self._ExecCleanup()
3682
    else:
3683
      return self._ExecMigration()
3684

    
3685

    
3686
def _CreateBlockDev(lu, node, instance, device, force_create,
3687
                    info, force_open):
3688
  """Create a tree of block devices on a given node.
3689

3690
  If this device type has to be created on secondaries, create it and
3691
  all its children.
3692

3693
  If not, just recurse to children keeping the same 'force' value.
3694

3695
  @param lu: the lu on whose behalf we execute
3696
  @param node: the node on which to create the device
3697
  @type instance: L{objects.Instance}
3698
  @param instance: the instance which owns the device
3699
  @type device: L{objects.Disk}
3700
  @param device: the device to create
3701
  @type force_create: boolean
3702
  @param force_create: whether to force creation of this device; this
3703
      will be change to True whenever we find a device which has
3704
      CreateOnSecondary() attribute
3705
  @param info: the extra 'metadata' we should attach to the device
3706
      (this will be represented as a LVM tag)
3707
  @type force_open: boolean
3708
  @param force_open: this parameter will be passes to the
3709
      L{backend.CreateBlockDevice} function where it specifies
3710
      whether we run on primary or not, and it affects both
3711
      the child assembly and the device own Open() execution
3712

3713
  """
3714
  if device.CreateOnSecondary():
3715
    force_create = True
3716

    
3717
  if device.children:
3718
    for child in device.children:
3719
      _CreateBlockDev(lu, node, instance, child, force_create,
3720
                      info, force_open)
3721

    
3722
  if not force_create:
3723
    return
3724

    
3725
  _CreateSingleBlockDev(lu, node, instance, device, info, force_open)
3726

    
3727

    
3728
def _CreateSingleBlockDev(lu, node, instance, device, info, force_open):
3729
  """Create a single block device on a given node.
3730

3731
  This will not recurse over children of the device, so they must be
3732
  created in advance.
3733

3734
  @param lu: the lu on whose behalf we execute
3735
  @param node: the node on which to create the device
3736
  @type instance: L{objects.Instance}
3737
  @param instance: the instance which owns the device
3738
  @type device: L{objects.Disk}
3739
  @param device: the device to create
3740
  @param info: the extra 'metadata' we should attach to the device
3741
      (this will be represented as a LVM tag)
3742
  @type force_open: boolean
3743
  @param force_open: this parameter will be passes to the
3744
      L{backend.CreateBlockDevice} function where it specifies
3745
      whether we run on primary or not, and it affects both
3746
      the child assembly and the device own Open() execution
3747

3748
  """
3749
  lu.cfg.SetDiskID(device, node)
3750
  new_id = lu.rpc.call_blockdev_create(node, device, device.size,
3751
                                       instance.name, force_open, info)
3752
  if new_id.failed or not new_id.data:
3753
    raise errors.OpExecError("Can't create block device %s on"
3754
                             " node %s for instance %s" %
3755
                             (device, node, instance.name))
3756
  if device.physical_id is None:
3757
    device.physical_id = new_id
3758

    
3759

    
3760
def _GenerateUniqueNames(lu, exts):
3761
  """Generate a suitable LV name.
3762

3763
  This will generate a logical volume name for the given instance.
3764

3765
  """
3766
  results = []
3767
  for val in exts:
3768
    new_id = lu.cfg.GenerateUniqueID()
3769
    results.append("%s%s" % (new_id, val))
3770
  return results
3771

    
3772

    
3773
def _GenerateDRBD8Branch(lu, primary, secondary, size, names, iv_name,
3774
                         p_minor, s_minor):
3775
  """Generate a drbd8 device complete with its children.
3776

3777
  """
3778
  port = lu.cfg.AllocatePort()
3779
  vgname = lu.cfg.GetVGName()
3780
  shared_secret = lu.cfg.GenerateDRBDSecret()
3781
  dev_data = objects.Disk(dev_type=constants.LD_LV, size=size,
3782
                          logical_id=(vgname, names[0]))
3783
  dev_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
3784
                          logical_id=(vgname, names[1]))
3785
  drbd_dev = objects.Disk(dev_type=constants.LD_DRBD8, size=size,
3786
                          logical_id=(primary, secondary, port,
3787
                                      p_minor, s_minor,
3788
                                      shared_secret),
3789
                          children=[dev_data, dev_meta],
3790
                          iv_name=iv_name)
3791
  return drbd_dev
3792

    
3793

    
3794
def _GenerateDiskTemplate(lu, template_name,
3795
                          instance_name, primary_node,
3796
                          secondary_nodes, disk_info,
3797
                          file_storage_dir, file_driver,
3798
                          base_index):
3799
  """Generate the entire disk layout for a given template type.
3800

3801
  """
3802
  #TODO: compute space requirements
3803

    
3804
  vgname = lu.cfg.GetVGName()
3805
  disk_count = len(disk_info)
3806
  disks = []
3807
  if template_name == constants.DT_DISKLESS:
3808
    pass
3809
  elif template_name == constants.DT_PLAIN:
3810
    if len(secondary_nodes) != 0:
3811
      raise errors.ProgrammerError("Wrong template configuration")
3812

    
3813
    names = _GenerateUniqueNames(lu, [".disk%d" % i
3814
                                      for i in range(disk_count)])
3815
    for idx, disk in enumerate(disk_info):
3816
      disk_index = idx + base_index
3817
      disk_dev = objects.Disk(dev_type=constants.LD_LV, size=disk["size"],
3818
                              logical_id=(vgname, names[idx]),
3819
                              iv_name="disk/%d" % disk_index)
3820
      disks.append(disk_dev)
3821
  elif template_name == constants.DT_DRBD8:
3822
    if len(secondary_nodes) != 1:
3823
      raise errors.ProgrammerError("Wrong template configuration")
3824
    remote_node = secondary_nodes[0]
3825
    minors = lu.cfg.AllocateDRBDMinor(
3826
      [primary_node, remote_node] * len(disk_info), instance_name)
3827

    
3828
    names = []
3829
    for lv_prefix in _GenerateUniqueNames(lu, [".disk%d" % i
3830
                                               for i in range(disk_count)]):
3831
      names.append(lv_prefix + "_data")
3832
      names.append(lv_prefix + "_meta")
3833
    for idx, disk in enumerate(disk_info):
3834
      disk_index = idx + base_index
3835
      disk_dev = _GenerateDRBD8Branch(lu, primary_node, remote_node,
3836
                                      disk["size"], names[idx*2:idx*2+2],
3837
                                      "disk/%d" % disk_index,
3838
                                      minors[idx*2], minors[idx*2+1])
3839
      disks.append(disk_dev)
3840
  elif template_name == constants.DT_FILE:
3841
    if len(secondary_nodes) != 0:
3842
      raise errors.ProgrammerError("Wrong template configuration")
3843

    
3844
    for idx, disk in enumerate(disk_info):
3845
      disk_index = idx + base_index
3846
      disk_dev = objects.Disk(dev_type=constants.LD_FILE, size=disk["size"],
3847
                              iv_name="disk/%d" % disk_index,
3848
                              logical_id=(file_driver,
3849
                                          "%s/disk%d" % (file_storage_dir,
3850
                                                         idx)))
3851
      disks.append(disk_dev)
3852
  else:
3853
    raise errors.ProgrammerError("Invalid disk template '%s'" % template_name)
3854
  return disks
3855

    
3856

    
3857
def _GetInstanceInfoText(instance):
3858
  """Compute that text that should be added to the disk's metadata.
3859

3860
  """
3861
  return "originstname+%s" % instance.name
3862

    
3863

    
3864
def _CreateDisks(lu, instance):
3865
  """Create all disks for an instance.
3866

3867
  This abstracts away some work from AddInstance.
3868

3869
  @type lu: L{LogicalUnit}
3870
  @param lu: the logical unit on whose behalf we execute
3871
  @type instance: L{objects.Instance}
3872
  @param instance: the instance whose disks we should create
3873
  @rtype: boolean
3874
  @return: the success of the creation
3875

3876
  """
3877
  info = _GetInstanceInfoText(instance)
3878
  pnode = instance.primary_node
3879

    
3880
  if instance.disk_template == constants.DT_FILE:
3881
    file_storage_dir = os.path.dirname(instance.disks[0].logical_id[1])
3882
    result = lu.rpc.call_file_storage_dir_create(pnode, file_storage_dir)
3883

    
3884
    if result.failed or not result.data:
3885
      raise errors.OpExecError("Could not connect to node '%s'" % pnode)
3886

    
3887
    if not result.data[0]:
3888
      raise errors.OpExecError("Failed to create directory '%s'" %
3889
                               file_storage_dir)
3890

    
3891
  # Note: this needs to be kept in sync with adding of disks in
3892
  # LUSetInstanceParams
3893
  for device in instance.disks:
3894
    logging.info("Creating volume %s for instance %s",
3895
                 device.iv_name, instance.name)
3896
    #HARDCODE
3897
    for node in instance.all_nodes:
3898
      f_create = node == pnode
3899
      _CreateBlockDev(lu, node, instance, device, f_create, info, f_create)
3900

    
3901

    
3902
def _RemoveDisks(lu, instance):
3903
  """Remove all disks for an instance.
3904

3905
  This abstracts away some work from `AddInstance()` and
3906
  `RemoveInstance()`. Note that in case some of the devices couldn't
3907
  be removed, the removal will continue with the other ones (compare
3908
  with `_CreateDisks()`).
3909

3910
  @type lu: L{LogicalUnit}
3911
  @param lu: the logical unit on whose behalf we execute
3912
  @type instance: L{objects.Instance}
3913
  @param instance: the instance whose disks we should remove
3914
  @rtype: boolean
3915
  @return: the success of the removal
3916

3917
  """
3918
  logging.info("Removing block devices for instance %s", instance.name)
3919

    
3920
  result = True
3921
  for device in instance.disks:
3922
    for node, disk in device.ComputeNodeTree(instance.primary_node):
3923
      lu.cfg.SetDiskID(disk, node)
3924
      result = lu.rpc.call_blockdev_remove(node, disk)
3925
      if result.failed or not result.data:
3926
        lu.proc.LogWarning("Could not remove block device %s on node %s,"
3927
                           " continuing anyway", device.iv_name, node)
3928
        result = False
3929

    
3930
  if instance.disk_template == constants.DT_FILE:
3931
    file_storage_dir = os.path.dirname(instance.disks[0].logical_id[1])
3932
    result = lu.rpc.call_file_storage_dir_remove(instance.primary_node,
3933
                                                 file_storage_dir)
3934
    if result.failed or not result.data:
3935
      logging.error("Could not remove directory '%s'", file_storage_dir)
3936
      result = False
3937

    
3938
  return result
3939

    
3940

    
3941
def _ComputeDiskSize(disk_template, disks):
3942
  """Compute disk size requirements in the volume group
3943

3944
  """
3945
  # Required free disk space as a function of disk and swap space
3946
  req_size_dict = {
3947
    constants.DT_DISKLESS: None,
3948
    constants.DT_PLAIN: sum(d["size"] for d in disks),
3949
    # 128 MB are added for drbd metadata for each disk
3950
    constants.DT_DRBD8: sum(d["size"] + 128 for d in disks),
3951
    constants.DT_FILE: None,
3952
  }
3953

    
3954
  if disk_template not in req_size_dict:
3955
    raise errors.ProgrammerError("Disk template '%s' size requirement"
3956
                                 " is unknown" %  disk_template)
3957

    
3958
  return req_size_dict[disk_template]
3959

    
3960

    
3961
def _CheckHVParams(lu, nodenames, hvname, hvparams):
3962
  """Hypervisor parameter validation.
3963

3964
  This function abstract the hypervisor parameter validation to be
3965
  used in both instance create and instance modify.
3966

3967
  @type lu: L{LogicalUnit}
3968
  @param lu: the logical unit for which we check
3969
  @type nodenames: list
3970
  @param nodenames: the list of nodes on which we should check
3971
  @type hvname: string
3972
  @param hvname: the name of the hypervisor we should use
3973
  @type hvparams: dict
3974
  @param hvparams: the parameters which we need to check
3975
  @raise errors.OpPrereqError: if the parameters are not valid
3976

3977
  """
3978
  hvinfo = lu.rpc.call_hypervisor_validate_params(nodenames,
3979
                                                  hvname,
3980
                                                  hvparams)
3981
  for node in nodenames:
3982
    info = hvinfo[node]
3983
    info.Raise()
3984
    if<