Statistics
| Branch: | Tag: | Revision:

root / lib / cmdlib.py @ 94a02bb5

History | View | Annotate | Download (217.8 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
def _WaitForSync(lu, instance, oneshot=False, unlock=False):
1441
  """Sleep and poll for an instance's disk to sync.
1442

1443
  """
1444
  if not instance.disks:
1445
    return True
1446

    
1447
  if not oneshot:
1448
    lu.proc.LogInfo("Waiting for instance %s to sync disks." % instance.name)
1449

    
1450
  node = instance.primary_node
1451

    
1452
  for dev in instance.disks:
1453
    lu.cfg.SetDiskID(dev, node)
1454

    
1455
  retries = 0
1456
  while True:
1457
    max_time = 0
1458
    done = True
1459
    cumul_degraded = False
1460
    rstats = lu.rpc.call_blockdev_getmirrorstatus(node, instance.disks)
1461
    if rstats.failed or not rstats.data:
1462
      lu.LogWarning("Can't get any data from node %s", node)
1463
      retries += 1
1464
      if retries >= 10:
1465
        raise errors.RemoteError("Can't contact node %s for mirror data,"
1466
                                 " aborting." % node)
1467
      time.sleep(6)
1468
      continue
1469
    rstats = rstats.data
1470
    retries = 0
1471
    for i in range(len(rstats)):
1472
      mstat = rstats[i]
1473
      if mstat is None:
1474
        lu.LogWarning("Can't compute data for node %s/%s",
1475
                           node, instance.disks[i].iv_name)
1476
        continue
1477
      # we ignore the ldisk parameter
1478
      perc_done, est_time, is_degraded, _ = mstat
1479
      cumul_degraded = cumul_degraded or (is_degraded and perc_done is None)
1480
      if perc_done is not None:
1481
        done = False
1482
        if est_time is not None:
1483
          rem_time = "%d estimated seconds remaining" % est_time
1484
          max_time = est_time
1485
        else:
1486
          rem_time = "no time estimate"
1487
        lu.proc.LogInfo("- device %s: %5.2f%% done, %s" %
1488
                        (instance.disks[i].iv_name, perc_done, rem_time))
1489
    if done or oneshot:
1490
      break
1491

    
1492
    time.sleep(min(60, max_time))
1493

    
1494
  if done:
1495
    lu.proc.LogInfo("Instance %s's disks are in sync." % instance.name)
1496
  return not cumul_degraded
1497

    
1498

    
1499
def _CheckDiskConsistency(lu, dev, node, on_primary, ldisk=False):
1500
  """Check that mirrors are not degraded.
1501

1502
  The ldisk parameter, if True, will change the test from the
1503
  is_degraded attribute (which represents overall non-ok status for
1504
  the device(s)) to the ldisk (representing the local storage status).
1505

1506
  """
1507
  lu.cfg.SetDiskID(dev, node)
1508
  if ldisk:
1509
    idx = 6
1510
  else:
1511
    idx = 5
1512

    
1513
  result = True
1514
  if on_primary or dev.AssembleOnSecondary():
1515
    rstats = lu.rpc.call_blockdev_find(node, dev)
1516
    if rstats.failed or not rstats.data:
1517
      logging.warning("Node %s: disk degraded, not found or node down", node)
1518
      result = False
1519
    else:
1520
      result = result and (not rstats.data[idx])
1521
  if dev.children:
1522
    for child in dev.children:
1523
      result = result and _CheckDiskConsistency(lu, child, node, on_primary)
1524

    
1525
  return result
1526

    
1527

    
1528
class LUDiagnoseOS(NoHooksLU):
1529
  """Logical unit for OS diagnose/query.
1530

1531
  """
1532
  _OP_REQP = ["output_fields", "names"]
1533
  REQ_BGL = False
1534
  _FIELDS_STATIC = utils.FieldSet()
1535
  _FIELDS_DYNAMIC = utils.FieldSet("name", "valid", "node_status")
1536

    
1537
  def ExpandNames(self):
1538
    if self.op.names:
1539
      raise errors.OpPrereqError("Selective OS query not supported")
1540

    
1541
    _CheckOutputFields(static=self._FIELDS_STATIC,
1542
                       dynamic=self._FIELDS_DYNAMIC,
1543
                       selected=self.op.output_fields)
1544

    
1545
    # Lock all nodes, in shared mode
1546
    self.needed_locks = {}
1547
    self.share_locks[locking.LEVEL_NODE] = 1
1548
    self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
1549

    
1550
  def CheckPrereq(self):
1551
    """Check prerequisites.
1552

1553
    """
1554

    
1555
  @staticmethod
1556
  def _DiagnoseByOS(node_list, rlist):
1557
    """Remaps a per-node return list into an a per-os per-node dictionary
1558

1559
    @param node_list: a list with the names of all nodes
1560
    @param rlist: a map with node names as keys and OS objects as values
1561

1562
    @rtype: dict
1563
    @returns: a dictionary with osnames as keys and as value another map, with
1564
        nodes as keys and list of OS objects as values, eg::
1565

1566
          {"debian-etch": {"node1": [<object>,...],
1567
                           "node2": [<object>,]}
1568
          }
1569

1570
    """
1571
    all_os = {}
1572
    for node_name, nr in rlist.iteritems():
1573
      if nr.failed or not nr.data:
1574
        continue
1575
      for os_obj in nr.data:
1576
        if os_obj.name not in all_os:
1577
          # build a list of nodes for this os containing empty lists
1578
          # for each node in node_list
1579
          all_os[os_obj.name] = {}
1580
          for nname in node_list:
1581
            all_os[os_obj.name][nname] = []
1582
        all_os[os_obj.name][node_name].append(os_obj)
1583
    return all_os
1584

    
1585
  def Exec(self, feedback_fn):
1586
    """Compute the list of OSes.
1587

1588
    """
1589
    node_list = self.acquired_locks[locking.LEVEL_NODE]
1590
    valid_nodes = [node for node in self.cfg.GetOnlineNodeList()
1591
                   if node in node_list]
1592
    node_data = self.rpc.call_os_diagnose(valid_nodes)
1593
    if node_data == False:
1594
      raise errors.OpExecError("Can't gather the list of OSes")
1595
    pol = self._DiagnoseByOS(valid_nodes, node_data)
1596
    output = []
1597
    for os_name, os_data in pol.iteritems():
1598
      row = []
1599
      for field in self.op.output_fields:
1600
        if field == "name":
1601
          val = os_name
1602
        elif field == "valid":
1603
          val = utils.all([osl and osl[0] for osl in os_data.values()])
1604
        elif field == "node_status":
1605
          val = {}
1606
          for node_name, nos_list in os_data.iteritems():
1607
            val[node_name] = [(v.status, v.path) for v in nos_list]
1608
        else:
1609
          raise errors.ParameterError(field)
1610
        row.append(val)
1611
      output.append(row)
1612

    
1613
    return output
1614

    
1615

    
1616
class LURemoveNode(LogicalUnit):
1617
  """Logical unit for removing a node.
1618

1619
  """
1620
  HPATH = "node-remove"
1621
  HTYPE = constants.HTYPE_NODE
1622
  _OP_REQP = ["node_name"]
1623

    
1624
  def BuildHooksEnv(self):
1625
    """Build hooks env.
1626

1627
    This doesn't run on the target node in the pre phase as a failed
1628
    node would then be impossible to remove.
1629

1630
    """
1631
    env = {
1632
      "OP_TARGET": self.op.node_name,
1633
      "NODE_NAME": self.op.node_name,
1634
      }
1635
    all_nodes = self.cfg.GetNodeList()
1636
    all_nodes.remove(self.op.node_name)
1637
    return env, all_nodes, all_nodes
1638

    
1639
  def CheckPrereq(self):
1640
    """Check prerequisites.
1641

1642
    This checks:
1643
     - the node exists in the configuration
1644
     - it does not have primary or secondary instances
1645
     - it's not the master
1646

1647
    Any errors are signalled by raising errors.OpPrereqError.
1648

1649
    """
1650
    node = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.node_name))
1651
    if node is None:
1652
      raise errors.OpPrereqError, ("Node '%s' is unknown." % self.op.node_name)
1653

    
1654
    instance_list = self.cfg.GetInstanceList()
1655

    
1656
    masternode = self.cfg.GetMasterNode()
1657
    if node.name == masternode:
1658
      raise errors.OpPrereqError("Node is the master node,"
1659
                                 " you need to failover first.")
1660

    
1661
    for instance_name in instance_list:
1662
      instance = self.cfg.GetInstanceInfo(instance_name)
1663
      if node.name == instance.primary_node:
1664
        raise errors.OpPrereqError("Instance %s still running on the node,"
1665
                                   " please remove first." % instance_name)
1666
      if node.name in instance.secondary_nodes:
1667
        raise errors.OpPrereqError("Instance %s has node as a secondary,"
1668
                                   " please remove first." % instance_name)
1669
    self.op.node_name = node.name
1670
    self.node = node
1671

    
1672
  def Exec(self, feedback_fn):
1673
    """Removes the node from the cluster.
1674

1675
    """
1676
    node = self.node
1677
    logging.info("Stopping the node daemon and removing configs from node %s",
1678
                 node.name)
1679

    
1680
    self.context.RemoveNode(node.name)
1681

    
1682
    self.rpc.call_node_leave_cluster(node.name)
1683

    
1684
    # Promote nodes to master candidate as needed
1685
    _AdjustCandidatePool(self)
1686

    
1687

    
1688
class LUQueryNodes(NoHooksLU):
1689
  """Logical unit for querying nodes.
1690

1691
  """
1692
  _OP_REQP = ["output_fields", "names"]
1693
  REQ_BGL = False
1694
  _FIELDS_DYNAMIC = utils.FieldSet(
1695
    "dtotal", "dfree",
1696
    "mtotal", "mnode", "mfree",
1697
    "bootid",
1698
    "ctotal",
1699
    )
1700

    
1701
  _FIELDS_STATIC = utils.FieldSet(
1702
    "name", "pinst_cnt", "sinst_cnt",
1703
    "pinst_list", "sinst_list",
1704
    "pip", "sip", "tags",
1705
    "serial_no",
1706
    "master_candidate",
1707
    "master",
1708
    "offline",
1709
    )
1710

    
1711
  def ExpandNames(self):
1712
    _CheckOutputFields(static=self._FIELDS_STATIC,
1713
                       dynamic=self._FIELDS_DYNAMIC,
1714
                       selected=self.op.output_fields)
1715

    
1716
    self.needed_locks = {}
1717
    self.share_locks[locking.LEVEL_NODE] = 1
1718

    
1719
    if self.op.names:
1720
      self.wanted = _GetWantedNodes(self, self.op.names)
1721
    else:
1722
      self.wanted = locking.ALL_SET
1723

    
1724
    self.do_locking = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
1725
    if self.do_locking:
1726
      # if we don't request only static fields, we need to lock the nodes
1727
      self.needed_locks[locking.LEVEL_NODE] = self.wanted
1728

    
1729

    
1730
  def CheckPrereq(self):
1731
    """Check prerequisites.
1732

1733
    """
1734
    # The validation of the node list is done in the _GetWantedNodes,
1735
    # if non empty, and if empty, there's no validation to do
1736
    pass
1737

    
1738
  def Exec(self, feedback_fn):
1739
    """Computes the list of nodes and their attributes.
1740

1741
    """
1742
    all_info = self.cfg.GetAllNodesInfo()
1743
    if self.do_locking:
1744
      nodenames = self.acquired_locks[locking.LEVEL_NODE]
1745
    elif self.wanted != locking.ALL_SET:
1746
      nodenames = self.wanted
1747
      missing = set(nodenames).difference(all_info.keys())
1748
      if missing:
1749
        raise errors.OpExecError(
1750
          "Some nodes were removed before retrieving their data: %s" % missing)
1751
    else:
1752
      nodenames = all_info.keys()
1753

    
1754
    nodenames = utils.NiceSort(nodenames)
1755
    nodelist = [all_info[name] for name in nodenames]
1756

    
1757
    # begin data gathering
1758

    
1759
    if self.do_locking:
1760
      live_data = {}
1761
      node_data = self.rpc.call_node_info(nodenames, self.cfg.GetVGName(),
1762
                                          self.cfg.GetHypervisorType())
1763
      for name in nodenames:
1764
        nodeinfo = node_data[name]
1765
        if not nodeinfo.failed and nodeinfo.data:
1766
          nodeinfo = nodeinfo.data
1767
          fn = utils.TryConvert
1768
          live_data[name] = {
1769
            "mtotal": fn(int, nodeinfo.get('memory_total', None)),
1770
            "mnode": fn(int, nodeinfo.get('memory_dom0', None)),
1771
            "mfree": fn(int, nodeinfo.get('memory_free', None)),
1772
            "dtotal": fn(int, nodeinfo.get('vg_size', None)),
1773
            "dfree": fn(int, nodeinfo.get('vg_free', None)),
1774
            "ctotal": fn(int, nodeinfo.get('cpu_total', None)),
1775
            "bootid": nodeinfo.get('bootid', None),
1776
            }
1777
        else:
1778
          live_data[name] = {}
1779
    else:
1780
      live_data = dict.fromkeys(nodenames, {})
1781

    
1782
    node_to_primary = dict([(name, set()) for name in nodenames])
1783
    node_to_secondary = dict([(name, set()) for name in nodenames])
1784

    
1785
    inst_fields = frozenset(("pinst_cnt", "pinst_list",
1786
                             "sinst_cnt", "sinst_list"))
1787
    if inst_fields & frozenset(self.op.output_fields):
1788
      instancelist = self.cfg.GetInstanceList()
1789

    
1790
      for instance_name in instancelist:
1791
        inst = self.cfg.GetInstanceInfo(instance_name)
1792
        if inst.primary_node in node_to_primary:
1793
          node_to_primary[inst.primary_node].add(inst.name)
1794
        for secnode in inst.secondary_nodes:
1795
          if secnode in node_to_secondary:
1796
            node_to_secondary[secnode].add(inst.name)
1797

    
1798
    master_node = self.cfg.GetMasterNode()
1799

    
1800
    # end data gathering
1801

    
1802
    output = []
1803
    for node in nodelist:
1804
      node_output = []
1805
      for field in self.op.output_fields:
1806
        if field == "name":
1807
          val = node.name
1808
        elif field == "pinst_list":
1809
          val = list(node_to_primary[node.name])
1810
        elif field == "sinst_list":
1811
          val = list(node_to_secondary[node.name])
1812
        elif field == "pinst_cnt":
1813
          val = len(node_to_primary[node.name])
1814
        elif field == "sinst_cnt":
1815
          val = len(node_to_secondary[node.name])
1816
        elif field == "pip":
1817
          val = node.primary_ip
1818
        elif field == "sip":
1819
          val = node.secondary_ip
1820
        elif field == "tags":
1821
          val = list(node.GetTags())
1822
        elif field == "serial_no":
1823
          val = node.serial_no
1824
        elif field == "master_candidate":
1825
          val = node.master_candidate
1826
        elif field == "master":
1827
          val = node.name == master_node
1828
        elif field == "offline":
1829
          val = node.offline
1830
        elif self._FIELDS_DYNAMIC.Matches(field):
1831
          val = live_data[node.name].get(field, None)
1832
        else:
1833
          raise errors.ParameterError(field)
1834
        node_output.append(val)
1835
      output.append(node_output)
1836

    
1837
    return output
1838

    
1839

    
1840
class LUQueryNodeVolumes(NoHooksLU):
1841
  """Logical unit for getting volumes on node(s).
1842

1843
  """
1844
  _OP_REQP = ["nodes", "output_fields"]
1845
  REQ_BGL = False
1846
  _FIELDS_DYNAMIC = utils.FieldSet("phys", "vg", "name", "size", "instance")
1847
  _FIELDS_STATIC = utils.FieldSet("node")
1848

    
1849
  def ExpandNames(self):
1850
    _CheckOutputFields(static=self._FIELDS_STATIC,
1851
                       dynamic=self._FIELDS_DYNAMIC,
1852
                       selected=self.op.output_fields)
1853

    
1854
    self.needed_locks = {}
1855
    self.share_locks[locking.LEVEL_NODE] = 1
1856
    if not self.op.nodes:
1857
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
1858
    else:
1859
      self.needed_locks[locking.LEVEL_NODE] = \
1860
        _GetWantedNodes(self, self.op.nodes)
1861

    
1862
  def CheckPrereq(self):
1863
    """Check prerequisites.
1864

1865
    This checks that the fields required are valid output fields.
1866

1867
    """
1868
    self.nodes = self.acquired_locks[locking.LEVEL_NODE]
1869

    
1870
  def Exec(self, feedback_fn):
1871
    """Computes the list of nodes and their attributes.
1872

1873
    """
1874
    nodenames = self.nodes
1875
    volumes = self.rpc.call_node_volumes(nodenames)
1876

    
1877
    ilist = [self.cfg.GetInstanceInfo(iname) for iname
1878
             in self.cfg.GetInstanceList()]
1879

    
1880
    lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
1881

    
1882
    output = []
1883
    for node in nodenames:
1884
      if node not in volumes or volumes[node].failed or not volumes[node].data:
1885
        continue
1886

    
1887
      node_vols = volumes[node].data[:]
1888
      node_vols.sort(key=lambda vol: vol['dev'])
1889

    
1890
      for vol in node_vols:
1891
        node_output = []
1892
        for field in self.op.output_fields:
1893
          if field == "node":
1894
            val = node
1895
          elif field == "phys":
1896
            val = vol['dev']
1897
          elif field == "vg":
1898
            val = vol['vg']
1899
          elif field == "name":
1900
            val = vol['name']
1901
          elif field == "size":
1902
            val = int(float(vol['size']))
1903
          elif field == "instance":
1904
            for inst in ilist:
1905
              if node not in lv_by_node[inst]:
1906
                continue
1907
              if vol['name'] in lv_by_node[inst][node]:
1908
                val = inst.name
1909
                break
1910
            else:
1911
              val = '-'
1912
          else:
1913
            raise errors.ParameterError(field)
1914
          node_output.append(str(val))
1915

    
1916
        output.append(node_output)
1917

    
1918
    return output
1919

    
1920

    
1921
class LUAddNode(LogicalUnit):
1922
  """Logical unit for adding node to the cluster.
1923

1924
  """
1925
  HPATH = "node-add"
1926
  HTYPE = constants.HTYPE_NODE
1927
  _OP_REQP = ["node_name"]
1928

    
1929
  def BuildHooksEnv(self):
1930
    """Build hooks env.
1931

1932
    This will run on all nodes before, and on all nodes + the new node after.
1933

1934
    """
1935
    env = {
1936
      "OP_TARGET": self.op.node_name,
1937
      "NODE_NAME": self.op.node_name,
1938
      "NODE_PIP": self.op.primary_ip,
1939
      "NODE_SIP": self.op.secondary_ip,
1940
      }
1941
    nodes_0 = self.cfg.GetNodeList()
1942
    nodes_1 = nodes_0 + [self.op.node_name, ]
1943
    return env, nodes_0, nodes_1
1944

    
1945
  def CheckPrereq(self):
1946
    """Check prerequisites.
1947

1948
    This checks:
1949
     - the new node is not already in the config
1950
     - it is resolvable
1951
     - its parameters (single/dual homed) matches the cluster
1952

1953
    Any errors are signalled by raising errors.OpPrereqError.
1954

1955
    """
1956
    node_name = self.op.node_name
1957
    cfg = self.cfg
1958

    
1959
    dns_data = utils.HostInfo(node_name)
1960

    
1961
    node = dns_data.name
1962
    primary_ip = self.op.primary_ip = dns_data.ip
1963
    secondary_ip = getattr(self.op, "secondary_ip", None)
1964
    if secondary_ip is None:
1965
      secondary_ip = primary_ip
1966
    if not utils.IsValidIP(secondary_ip):
1967
      raise errors.OpPrereqError("Invalid secondary IP given")
1968
    self.op.secondary_ip = secondary_ip
1969

    
1970
    node_list = cfg.GetNodeList()
1971
    if not self.op.readd and node in node_list:
1972
      raise errors.OpPrereqError("Node %s is already in the configuration" %
1973
                                 node)
1974
    elif self.op.readd and node not in node_list:
1975
      raise errors.OpPrereqError("Node %s is not in the configuration" % node)
1976

    
1977
    for existing_node_name in node_list:
1978
      existing_node = cfg.GetNodeInfo(existing_node_name)
1979

    
1980
      if self.op.readd and node == existing_node_name:
1981
        if (existing_node.primary_ip != primary_ip or
1982
            existing_node.secondary_ip != secondary_ip):
1983
          raise errors.OpPrereqError("Readded node doesn't have the same IP"
1984
                                     " address configuration as before")
1985
        continue
1986

    
1987
      if (existing_node.primary_ip == primary_ip or
1988
          existing_node.secondary_ip == primary_ip or
1989
          existing_node.primary_ip == secondary_ip or
1990
          existing_node.secondary_ip == secondary_ip):
1991
        raise errors.OpPrereqError("New node ip address(es) conflict with"
1992
                                   " existing node %s" % existing_node.name)
1993

    
1994
    # check that the type of the node (single versus dual homed) is the
1995
    # same as for the master
1996
    myself = cfg.GetNodeInfo(self.cfg.GetMasterNode())
1997
    master_singlehomed = myself.secondary_ip == myself.primary_ip
1998
    newbie_singlehomed = secondary_ip == primary_ip
1999
    if master_singlehomed != newbie_singlehomed:
2000
      if master_singlehomed:
2001
        raise errors.OpPrereqError("The master has no private ip but the"
2002
                                   " new node has one")
2003
      else:
2004
        raise errors.OpPrereqError("The master has a private ip but the"
2005
                                   " new node doesn't have one")
2006

    
2007
    # checks reachablity
2008
    if not utils.TcpPing(primary_ip, constants.DEFAULT_NODED_PORT):
2009
      raise errors.OpPrereqError("Node not reachable by ping")
2010

    
2011
    if not newbie_singlehomed:
2012
      # check reachability from my secondary ip to newbie's secondary ip
2013
      if not utils.TcpPing(secondary_ip, constants.DEFAULT_NODED_PORT,
2014
                           source=myself.secondary_ip):
2015
        raise errors.OpPrereqError("Node secondary ip not reachable by TCP"
2016
                                   " based ping to noded port")
2017

    
2018
    cp_size = self.cfg.GetClusterInfo().candidate_pool_size
2019
    mc_now, _ = self.cfg.GetMasterCandidateStats()
2020
    master_candidate = mc_now < cp_size
2021

    
2022
    self.new_node = objects.Node(name=node,
2023
                                 primary_ip=primary_ip,
2024
                                 secondary_ip=secondary_ip,
2025
                                 master_candidate=master_candidate,
2026
                                 offline=False)
2027

    
2028
  def Exec(self, feedback_fn):
2029
    """Adds the new node to the cluster.
2030

2031
    """
2032
    new_node = self.new_node
2033
    node = new_node.name
2034

    
2035
    # check connectivity
2036
    result = self.rpc.call_version([node])[node]
2037
    result.Raise()
2038
    if result.data:
2039
      if constants.PROTOCOL_VERSION == result.data:
2040
        logging.info("Communication to node %s fine, sw version %s match",
2041
                     node, result.data)
2042
      else:
2043
        raise errors.OpExecError("Version mismatch master version %s,"
2044
                                 " node version %s" %
2045
                                 (constants.PROTOCOL_VERSION, result.data))
2046
    else:
2047
      raise errors.OpExecError("Cannot get version from the new node")
2048

    
2049
    # setup ssh on node
2050
    logging.info("Copy ssh key to node %s", node)
2051
    priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
2052
    keyarray = []
2053
    keyfiles = [constants.SSH_HOST_DSA_PRIV, constants.SSH_HOST_DSA_PUB,
2054
                constants.SSH_HOST_RSA_PRIV, constants.SSH_HOST_RSA_PUB,
2055
                priv_key, pub_key]
2056

    
2057
    for i in keyfiles:
2058
      f = open(i, 'r')
2059
      try:
2060
        keyarray.append(f.read())
2061
      finally:
2062
        f.close()
2063

    
2064
    result = self.rpc.call_node_add(node, keyarray[0], keyarray[1],
2065
                                    keyarray[2],
2066
                                    keyarray[3], keyarray[4], keyarray[5])
2067

    
2068
    if result.failed or not result.data:
2069
      raise errors.OpExecError("Cannot transfer ssh keys to the new node")
2070

    
2071
    # Add node to our /etc/hosts, and add key to known_hosts
2072
    utils.AddHostToEtcHosts(new_node.name)
2073

    
2074
    if new_node.secondary_ip != new_node.primary_ip:
2075
      result = self.rpc.call_node_has_ip_address(new_node.name,
2076
                                                 new_node.secondary_ip)
2077
      if result.failed or not result.data:
2078
        raise errors.OpExecError("Node claims it doesn't have the secondary ip"
2079
                                 " you gave (%s). Please fix and re-run this"
2080
                                 " command." % new_node.secondary_ip)
2081

    
2082
    node_verify_list = [self.cfg.GetMasterNode()]
2083
    node_verify_param = {
2084
      'nodelist': [node],
2085
      # TODO: do a node-net-test as well?
2086
    }
2087

    
2088
    result = self.rpc.call_node_verify(node_verify_list, node_verify_param,
2089
                                       self.cfg.GetClusterName())
2090
    for verifier in node_verify_list:
2091
      if result[verifier].failed or not result[verifier].data:
2092
        raise errors.OpExecError("Cannot communicate with %s's node daemon"
2093
                                 " for remote verification" % verifier)
2094
      if result[verifier].data['nodelist']:
2095
        for failed in result[verifier].data['nodelist']:
2096
          feedback_fn("ssh/hostname verification failed %s -> %s" %
2097
                      (verifier, result[verifier]['nodelist'][failed]))
2098
        raise errors.OpExecError("ssh/hostname verification failed.")
2099

    
2100
    # Distribute updated /etc/hosts and known_hosts to all nodes,
2101
    # including the node just added
2102
    myself = self.cfg.GetNodeInfo(self.cfg.GetMasterNode())
2103
    dist_nodes = self.cfg.GetNodeList()
2104
    if not self.op.readd:
2105
      dist_nodes.append(node)
2106
    if myself.name in dist_nodes:
2107
      dist_nodes.remove(myself.name)
2108

    
2109
    logging.debug("Copying hosts and known_hosts to all nodes")
2110
    for fname in (constants.ETC_HOSTS, constants.SSH_KNOWN_HOSTS_FILE):
2111
      result = self.rpc.call_upload_file(dist_nodes, fname)
2112
      for to_node, to_result in result.iteritems():
2113
        if to_result.failed or not to_result.data:
2114
          logging.error("Copy of file %s to node %s failed", fname, to_node)
2115

    
2116
    to_copy = []
2117
    if constants.HT_XEN_HVM in self.cfg.GetClusterInfo().enabled_hypervisors:
2118
      to_copy.append(constants.VNC_PASSWORD_FILE)
2119
    for fname in to_copy:
2120
      result = self.rpc.call_upload_file([node], fname)
2121
      if result[node].failed or not result[node]:
2122
        logging.error("Could not copy file %s to node %s", fname, node)
2123

    
2124
    if self.op.readd:
2125
      self.context.ReaddNode(new_node)
2126
    else:
2127
      self.context.AddNode(new_node)
2128

    
2129

    
2130
class LUSetNodeParams(LogicalUnit):
2131
  """Modifies the parameters of a node.
2132

2133
  """
2134
  HPATH = "node-modify"
2135
  HTYPE = constants.HTYPE_NODE
2136
  _OP_REQP = ["node_name"]
2137
  REQ_BGL = False
2138

    
2139
  def CheckArguments(self):
2140
    node_name = self.cfg.ExpandNodeName(self.op.node_name)
2141
    if node_name is None:
2142
      raise errors.OpPrereqError("Invalid node name '%s'" % self.op.node_name)
2143
    self.op.node_name = node_name
2144
    _CheckBooleanOpField(self.op, 'master_candidate')
2145
    _CheckBooleanOpField(self.op, 'offline')
2146
    if self.op.master_candidate is None and self.op.offline is None:
2147
      raise errors.OpPrereqError("Please pass at least one modification")
2148
    if self.op.offline == True and self.op.master_candidate == True:
2149
      raise errors.OpPrereqError("Can't set the node into offline and"
2150
                                 " master_candidate at the same time")
2151

    
2152
  def ExpandNames(self):
2153
    self.needed_locks = {locking.LEVEL_NODE: self.op.node_name}
2154

    
2155
  def BuildHooksEnv(self):
2156
    """Build hooks env.
2157

2158
    This runs on the master node.
2159

2160
    """
2161
    env = {
2162
      "OP_TARGET": self.op.node_name,
2163
      "MASTER_CANDIDATE": str(self.op.master_candidate),
2164
      "OFFLINE": str(self.op.offline),
2165
      }
2166
    nl = [self.cfg.GetMasterNode(),
2167
          self.op.node_name]
2168
    return env, nl, nl
2169

    
2170
  def CheckPrereq(self):
2171
    """Check prerequisites.
2172

2173
    This only checks the instance list against the existing names.
2174

2175
    """
2176
    node = self.node = self.cfg.GetNodeInfo(self.op.node_name)
2177

    
2178
    if ((self.op.master_candidate == False or self.op.offline == True)
2179
        and node.master_candidate):
2180
      # we will demote the node from master_candidate
2181
      if self.op.node_name == self.cfg.GetMasterNode():
2182
        raise errors.OpPrereqError("The master node has to be a"
2183
                                   " master candidate and online")
2184
      cp_size = self.cfg.GetClusterInfo().candidate_pool_size
2185
      num_candidates, _ = self.cfg.GetMasterCandidateStats()
2186
      if num_candidates <= cp_size:
2187
        msg = ("Not enough master candidates (desired"
2188
               " %d, new value will be %d)" % (cp_size, num_candidates-1))
2189
        if self.op.force:
2190
          self.LogWarning(msg)
2191
        else:
2192
          raise errors.OpPrereqError(msg)
2193

    
2194
    if (self.op.master_candidate == True and node.offline and
2195
        not self.op.offline == False):
2196
      raise errors.OpPrereqError("Can't set an offline node to"
2197
                                 " master_candidate")
2198

    
2199
    return
2200

    
2201
  def Exec(self, feedback_fn):
2202
    """Modifies a node.
2203

2204
    """
2205
    node = self.node
2206

    
2207
    result = []
2208

    
2209
    if self.op.offline is not None:
2210
      node.offline = self.op.offline
2211
      result.append(("offline", str(self.op.offline)))
2212
      if self.op.offline == True and node.master_candidate:
2213
        node.master_candidate = False
2214
        result.append(("master_candidate", "auto-demotion due to offline"))
2215

    
2216
    if self.op.master_candidate is not None:
2217
      node.master_candidate = self.op.master_candidate
2218
      result.append(("master_candidate", str(self.op.master_candidate)))
2219
      if self.op.master_candidate == False:
2220
        rrc = self.rpc.call_node_demote_from_mc(node.name)
2221
        if (rrc.failed or not isinstance(rrc.data, (tuple, list))
2222
            or len(rrc.data) != 2):
2223
          self.LogWarning("Node rpc error: %s" % rrc.error)
2224
        elif not rrc.data[0]:
2225
          self.LogWarning("Node failed to demote itself: %s" % rrc.data[1])
2226

    
2227
    # this will trigger configuration file update, if needed
2228
    self.cfg.Update(node)
2229
    # this will trigger job queue propagation or cleanup
2230
    if self.op.node_name != self.cfg.GetMasterNode():
2231
      self.context.ReaddNode(node)
2232

    
2233
    return result
2234

    
2235

    
2236
class LUQueryClusterInfo(NoHooksLU):
2237
  """Query cluster configuration.
2238

2239
  """
2240
  _OP_REQP = []
2241
  REQ_BGL = False
2242

    
2243
  def ExpandNames(self):
2244
    self.needed_locks = {}
2245

    
2246
  def CheckPrereq(self):
2247
    """No prerequsites needed for this LU.
2248

2249
    """
2250
    pass
2251

    
2252
  def Exec(self, feedback_fn):
2253
    """Return cluster config.
2254

2255
    """
2256
    cluster = self.cfg.GetClusterInfo()
2257
    result = {
2258
      "software_version": constants.RELEASE_VERSION,
2259
      "protocol_version": constants.PROTOCOL_VERSION,
2260
      "config_version": constants.CONFIG_VERSION,
2261
      "os_api_version": constants.OS_API_VERSION,
2262
      "export_version": constants.EXPORT_VERSION,
2263
      "architecture": (platform.architecture()[0], platform.machine()),
2264
      "name": cluster.cluster_name,
2265
      "master": cluster.master_node,
2266
      "default_hypervisor": cluster.default_hypervisor,
2267
      "enabled_hypervisors": cluster.enabled_hypervisors,
2268
      "hvparams": cluster.hvparams,
2269
      "beparams": cluster.beparams,
2270
      "candidate_pool_size": cluster.candidate_pool_size,
2271
      }
2272

    
2273
    return result
2274

    
2275

    
2276
class LUQueryConfigValues(NoHooksLU):
2277
  """Return configuration values.
2278

2279
  """
2280
  _OP_REQP = []
2281
  REQ_BGL = False
2282
  _FIELDS_DYNAMIC = utils.FieldSet()
2283
  _FIELDS_STATIC = utils.FieldSet("cluster_name", "master_node", "drain_flag")
2284

    
2285
  def ExpandNames(self):
2286
    self.needed_locks = {}
2287

    
2288
    _CheckOutputFields(static=self._FIELDS_STATIC,
2289
                       dynamic=self._FIELDS_DYNAMIC,
2290
                       selected=self.op.output_fields)
2291

    
2292
  def CheckPrereq(self):
2293
    """No prerequisites.
2294

2295
    """
2296
    pass
2297

    
2298
  def Exec(self, feedback_fn):
2299
    """Dump a representation of the cluster config to the standard output.
2300

2301
    """
2302
    values = []
2303
    for field in self.op.output_fields:
2304
      if field == "cluster_name":
2305
        entry = self.cfg.GetClusterName()
2306
      elif field == "master_node":
2307
        entry = self.cfg.GetMasterNode()
2308
      elif field == "drain_flag":
2309
        entry = os.path.exists(constants.JOB_QUEUE_DRAIN_FILE)
2310
      else:
2311
        raise errors.ParameterError(field)
2312
      values.append(entry)
2313
    return values
2314

    
2315

    
2316
class LUActivateInstanceDisks(NoHooksLU):
2317
  """Bring up an instance's disks.
2318

2319
  """
2320
  _OP_REQP = ["instance_name"]
2321
  REQ_BGL = False
2322

    
2323
  def ExpandNames(self):
2324
    self._ExpandAndLockInstance()
2325
    self.needed_locks[locking.LEVEL_NODE] = []
2326
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2327

    
2328
  def DeclareLocks(self, level):
2329
    if level == locking.LEVEL_NODE:
2330
      self._LockInstancesNodes()
2331

    
2332
  def CheckPrereq(self):
2333
    """Check prerequisites.
2334

2335
    This checks that the instance is in the cluster.
2336

2337
    """
2338
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2339
    assert self.instance is not None, \
2340
      "Cannot retrieve locked instance %s" % self.op.instance_name
2341
    _CheckNodeOnline(self, self.instance.primary_node)
2342

    
2343
  def Exec(self, feedback_fn):
2344
    """Activate the disks.
2345

2346
    """
2347
    disks_ok, disks_info = _AssembleInstanceDisks(self, self.instance)
2348
    if not disks_ok:
2349
      raise errors.OpExecError("Cannot activate block devices")
2350

    
2351
    return disks_info
2352

    
2353

    
2354
def _AssembleInstanceDisks(lu, instance, ignore_secondaries=False):
2355
  """Prepare the block devices for an instance.
2356

2357
  This sets up the block devices on all nodes.
2358

2359
  @type lu: L{LogicalUnit}
2360
  @param lu: the logical unit on whose behalf we execute
2361
  @type instance: L{objects.Instance}
2362
  @param instance: the instance for whose disks we assemble
2363
  @type ignore_secondaries: boolean
2364
  @param ignore_secondaries: if true, errors on secondary nodes
2365
      won't result in an error return from the function
2366
  @return: False if the operation failed, otherwise a list of
2367
      (host, instance_visible_name, node_visible_name)
2368
      with the mapping from node devices to instance devices
2369

2370
  """
2371
  device_info = []
2372
  disks_ok = True
2373
  iname = instance.name
2374
  # With the two passes mechanism we try to reduce the window of
2375
  # opportunity for the race condition of switching DRBD to primary
2376
  # before handshaking occured, but we do not eliminate it
2377

    
2378
  # The proper fix would be to wait (with some limits) until the
2379
  # connection has been made and drbd transitions from WFConnection
2380
  # into any other network-connected state (Connected, SyncTarget,
2381
  # SyncSource, etc.)
2382

    
2383
  # 1st pass, assemble on all nodes in secondary mode
2384
  for inst_disk in instance.disks:
2385
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
2386
      lu.cfg.SetDiskID(node_disk, node)
2387
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, False)
2388
      if result.failed or not result:
2389
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
2390
                           " (is_primary=False, pass=1)",
2391
                           inst_disk.iv_name, node)
2392
        if not ignore_secondaries:
2393
          disks_ok = False
2394

    
2395
  # FIXME: race condition on drbd migration to primary
2396

    
2397
  # 2nd pass, do only the primary node
2398
  for inst_disk in instance.disks:
2399
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
2400
      if node != instance.primary_node:
2401
        continue
2402
      lu.cfg.SetDiskID(node_disk, node)
2403
      result = lu.rpc.call_blockdev_assemble(node, node_disk, iname, True)
2404
      if result.failed or not result:
2405
        lu.proc.LogWarning("Could not prepare block device %s on node %s"
2406
                           " (is_primary=True, pass=2)",
2407
                           inst_disk.iv_name, node)
2408
        disks_ok = False
2409
    device_info.append((instance.primary_node, inst_disk.iv_name, result.data))
2410

    
2411
  # leave the disks configured for the primary node
2412
  # this is a workaround that would be fixed better by
2413
  # improving the logical/physical id handling
2414
  for disk in instance.disks:
2415
    lu.cfg.SetDiskID(disk, instance.primary_node)
2416

    
2417
  return disks_ok, device_info
2418

    
2419

    
2420
def _StartInstanceDisks(lu, instance, force):
2421
  """Start the disks of an instance.
2422

2423
  """
2424
  disks_ok, dummy = _AssembleInstanceDisks(lu, instance,
2425
                                           ignore_secondaries=force)
2426
  if not disks_ok:
2427
    _ShutdownInstanceDisks(lu, instance)
2428
    if force is not None and not force:
2429
      lu.proc.LogWarning("", hint="If the message above refers to a"
2430
                         " secondary node,"
2431
                         " you can retry the operation using '--force'.")
2432
    raise errors.OpExecError("Disk consistency error")
2433

    
2434

    
2435
class LUDeactivateInstanceDisks(NoHooksLU):
2436
  """Shutdown an instance's disks.
2437

2438
  """
2439
  _OP_REQP = ["instance_name"]
2440
  REQ_BGL = False
2441

    
2442
  def ExpandNames(self):
2443
    self._ExpandAndLockInstance()
2444
    self.needed_locks[locking.LEVEL_NODE] = []
2445
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2446

    
2447
  def DeclareLocks(self, level):
2448
    if level == locking.LEVEL_NODE:
2449
      self._LockInstancesNodes()
2450

    
2451
  def CheckPrereq(self):
2452
    """Check prerequisites.
2453

2454
    This checks that the instance is in the cluster.
2455

2456
    """
2457
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2458
    assert self.instance is not None, \
2459
      "Cannot retrieve locked instance %s" % self.op.instance_name
2460

    
2461
  def Exec(self, feedback_fn):
2462
    """Deactivate the disks
2463

2464
    """
2465
    instance = self.instance
2466
    _SafeShutdownInstanceDisks(self, instance)
2467

    
2468

    
2469
def _SafeShutdownInstanceDisks(lu, instance):
2470
  """Shutdown block devices of an instance.
2471

2472
  This function checks if an instance is running, before calling
2473
  _ShutdownInstanceDisks.
2474

2475
  """
2476
  ins_l = lu.rpc.call_instance_list([instance.primary_node],
2477
                                      [instance.hypervisor])
2478
  ins_l = ins_l[instance.primary_node]
2479
  if ins_l.failed or not isinstance(ins_l.data, list):
2480
    raise errors.OpExecError("Can't contact node '%s'" %
2481
                             instance.primary_node)
2482

    
2483
  if instance.name in ins_l.data:
2484
    raise errors.OpExecError("Instance is running, can't shutdown"
2485
                             " block devices.")
2486

    
2487
  _ShutdownInstanceDisks(lu, instance)
2488

    
2489

    
2490
def _ShutdownInstanceDisks(lu, instance, ignore_primary=False):
2491
  """Shutdown block devices of an instance.
2492

2493
  This does the shutdown on all nodes of the instance.
2494

2495
  If the ignore_primary is false, errors on the primary node are
2496
  ignored.
2497

2498
  """
2499
  result = True
2500
  for disk in instance.disks:
2501
    for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
2502
      lu.cfg.SetDiskID(top_disk, node)
2503
      result = lu.rpc.call_blockdev_shutdown(node, top_disk)
2504
      if result.failed or not result.data:
2505
        logging.error("Could not shutdown block device %s on node %s",
2506
                      disk.iv_name, node)
2507
        if not ignore_primary or node != instance.primary_node:
2508
          result = False
2509
  return result
2510

    
2511

    
2512
def _CheckNodeFreeMemory(lu, node, reason, requested, hypervisor_name):
2513
  """Checks if a node has enough free memory.
2514

2515
  This function check if a given node has the needed amount of free
2516
  memory. In case the node has less memory or we cannot get the
2517
  information from the node, this function raise an OpPrereqError
2518
  exception.
2519

2520
  @type lu: C{LogicalUnit}
2521
  @param lu: a logical unit from which we get configuration data
2522
  @type node: C{str}
2523
  @param node: the node to check
2524
  @type reason: C{str}
2525
  @param reason: string to use in the error message
2526
  @type requested: C{int}
2527
  @param requested: the amount of memory in MiB to check for
2528
  @type hypervisor_name: C{str}
2529
  @param hypervisor_name: the hypervisor to ask for memory stats
2530
  @raise errors.OpPrereqError: if the node doesn't have enough memory, or
2531
      we cannot check the node
2532

2533
  """
2534
  nodeinfo = lu.rpc.call_node_info([node], lu.cfg.GetVGName(), hypervisor_name)
2535
  nodeinfo[node].Raise()
2536
  free_mem = nodeinfo[node].data.get('memory_free')
2537
  if not isinstance(free_mem, int):
2538
    raise errors.OpPrereqError("Can't compute free memory on node %s, result"
2539
                             " was '%s'" % (node, free_mem))
2540
  if requested > free_mem:
2541
    raise errors.OpPrereqError("Not enough memory on node %s for %s:"
2542
                             " needed %s MiB, available %s MiB" %
2543
                             (node, reason, requested, free_mem))
2544

    
2545

    
2546
class LUStartupInstance(LogicalUnit):
2547
  """Starts an instance.
2548

2549
  """
2550
  HPATH = "instance-start"
2551
  HTYPE = constants.HTYPE_INSTANCE
2552
  _OP_REQP = ["instance_name", "force"]
2553
  REQ_BGL = False
2554

    
2555
  def ExpandNames(self):
2556
    self._ExpandAndLockInstance()
2557

    
2558
  def BuildHooksEnv(self):
2559
    """Build hooks env.
2560

2561
    This runs on master, primary and secondary nodes of the instance.
2562

2563
    """
2564
    env = {
2565
      "FORCE": self.op.force,
2566
      }
2567
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
2568
    nl = ([self.cfg.GetMasterNode(), self.instance.primary_node] +
2569
          list(self.instance.secondary_nodes))
2570
    return env, nl, nl
2571

    
2572
  def CheckPrereq(self):
2573
    """Check prerequisites.
2574

2575
    This checks that the instance is in the cluster.
2576

2577
    """
2578
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2579
    assert self.instance is not None, \
2580
      "Cannot retrieve locked instance %s" % self.op.instance_name
2581

    
2582
    _CheckNodeOnline(self, instance.primary_node)
2583

    
2584
    bep = self.cfg.GetClusterInfo().FillBE(instance)
2585
    # check bridges existance
2586
    _CheckInstanceBridgesExist(self, instance)
2587

    
2588
    _CheckNodeFreeMemory(self, instance.primary_node,
2589
                         "starting instance %s" % instance.name,
2590
                         bep[constants.BE_MEMORY], instance.hypervisor)
2591

    
2592
  def Exec(self, feedback_fn):
2593
    """Start the instance.
2594

2595
    """
2596
    instance = self.instance
2597
    force = self.op.force
2598
    extra_args = getattr(self.op, "extra_args", "")
2599

    
2600
    self.cfg.MarkInstanceUp(instance.name)
2601

    
2602
    node_current = instance.primary_node
2603

    
2604
    _StartInstanceDisks(self, instance, force)
2605

    
2606
    result = self.rpc.call_instance_start(node_current, instance, extra_args)
2607
    if result.failed or not result.data:
2608
      _ShutdownInstanceDisks(self, instance)
2609
      raise errors.OpExecError("Could not start instance")
2610

    
2611

    
2612
class LURebootInstance(LogicalUnit):
2613
  """Reboot an instance.
2614

2615
  """
2616
  HPATH = "instance-reboot"
2617
  HTYPE = constants.HTYPE_INSTANCE
2618
  _OP_REQP = ["instance_name", "ignore_secondaries", "reboot_type"]
2619
  REQ_BGL = False
2620

    
2621
  def ExpandNames(self):
2622
    if self.op.reboot_type not in [constants.INSTANCE_REBOOT_SOFT,
2623
                                   constants.INSTANCE_REBOOT_HARD,
2624
                                   constants.INSTANCE_REBOOT_FULL]:
2625
      raise errors.ParameterError("reboot type not in [%s, %s, %s]" %
2626
                                  (constants.INSTANCE_REBOOT_SOFT,
2627
                                   constants.INSTANCE_REBOOT_HARD,
2628
                                   constants.INSTANCE_REBOOT_FULL))
2629
    self._ExpandAndLockInstance()
2630

    
2631
  def BuildHooksEnv(self):
2632
    """Build hooks env.
2633

2634
    This runs on master, primary and secondary nodes of the instance.
2635

2636
    """
2637
    env = {
2638
      "IGNORE_SECONDARIES": self.op.ignore_secondaries,
2639
      }
2640
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
2641
    nl = ([self.cfg.GetMasterNode(), self.instance.primary_node] +
2642
          list(self.instance.secondary_nodes))
2643
    return env, nl, nl
2644

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

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

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

    
2655
    _CheckNodeOnline(self, instance.primary_node)
2656

    
2657
    # check bridges existance
2658
    _CheckInstanceBridgesExist(self, instance)
2659

    
2660
  def Exec(self, feedback_fn):
2661
    """Reboot the instance.
2662

2663
    """
2664
    instance = self.instance
2665
    ignore_secondaries = self.op.ignore_secondaries
2666
    reboot_type = self.op.reboot_type
2667
    extra_args = getattr(self.op, "extra_args", "")
2668

    
2669
    node_current = instance.primary_node
2670

    
2671
    if reboot_type in [constants.INSTANCE_REBOOT_SOFT,
2672
                       constants.INSTANCE_REBOOT_HARD]:
2673
      result = self.rpc.call_instance_reboot(node_current, instance,
2674
                                             reboot_type, extra_args)
2675
      if result.failed or not result.data:
2676
        raise errors.OpExecError("Could not reboot instance")
2677
    else:
2678
      if not self.rpc.call_instance_shutdown(node_current, instance):
2679
        raise errors.OpExecError("could not shutdown instance for full reboot")
2680
      _ShutdownInstanceDisks(self, instance)
2681
      _StartInstanceDisks(self, instance, ignore_secondaries)
2682
      result = self.rpc.call_instance_start(node_current, instance, extra_args)
2683
      if result.failed or not result.data:
2684
        _ShutdownInstanceDisks(self, instance)
2685
        raise errors.OpExecError("Could not start instance for full reboot")
2686

    
2687
    self.cfg.MarkInstanceUp(instance.name)
2688

    
2689

    
2690
class LUShutdownInstance(LogicalUnit):
2691
  """Shutdown an instance.
2692

2693
  """
2694
  HPATH = "instance-stop"
2695
  HTYPE = constants.HTYPE_INSTANCE
2696
  _OP_REQP = ["instance_name"]
2697
  REQ_BGL = False
2698

    
2699
  def ExpandNames(self):
2700
    self._ExpandAndLockInstance()
2701

    
2702
  def BuildHooksEnv(self):
2703
    """Build hooks env.
2704

2705
    This runs on master, primary and secondary nodes of the instance.
2706

2707
    """
2708
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2709
    nl = ([self.cfg.GetMasterNode(), self.instance.primary_node] +
2710
          list(self.instance.secondary_nodes))
2711
    return env, nl, nl
2712

    
2713
  def CheckPrereq(self):
2714
    """Check prerequisites.
2715

2716
    This checks that the instance is in the cluster.
2717

2718
    """
2719
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2720
    assert self.instance is not None, \
2721
      "Cannot retrieve locked instance %s" % self.op.instance_name
2722
    _CheckNodeOnline(self, self.instance.primary_node)
2723

    
2724
  def Exec(self, feedback_fn):
2725
    """Shutdown the instance.
2726

2727
    """
2728
    instance = self.instance
2729
    node_current = instance.primary_node
2730
    self.cfg.MarkInstanceDown(instance.name)
2731
    result = self.rpc.call_instance_shutdown(node_current, instance)
2732
    if result.failed or not result.data:
2733
      self.proc.LogWarning("Could not shutdown instance")
2734

    
2735
    _ShutdownInstanceDisks(self, instance)
2736

    
2737

    
2738
class LUReinstallInstance(LogicalUnit):
2739
  """Reinstall an instance.
2740

2741
  """
2742
  HPATH = "instance-reinstall"
2743
  HTYPE = constants.HTYPE_INSTANCE
2744
  _OP_REQP = ["instance_name"]
2745
  REQ_BGL = False
2746

    
2747
  def ExpandNames(self):
2748
    self._ExpandAndLockInstance()
2749

    
2750
  def BuildHooksEnv(self):
2751
    """Build hooks env.
2752

2753
    This runs on master, primary and secondary nodes of the instance.
2754

2755
    """
2756
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2757
    nl = ([self.cfg.GetMasterNode(), self.instance.primary_node] +
2758
          list(self.instance.secondary_nodes))
2759
    return env, nl, nl
2760

    
2761
  def CheckPrereq(self):
2762
    """Check prerequisites.
2763

2764
    This checks that the instance is in the cluster and is not running.
2765

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

    
2772
    if instance.disk_template == constants.DT_DISKLESS:
2773
      raise errors.OpPrereqError("Instance '%s' has no disks" %
2774
                                 self.op.instance_name)
2775
    if instance.status != "down":
2776
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2777
                                 self.op.instance_name)
2778
    remote_info = self.rpc.call_instance_info(instance.primary_node,
2779
                                              instance.name,
2780
                                              instance.hypervisor)
2781
    if remote_info.failed or remote_info.data:
2782
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
2783
                                 (self.op.instance_name,
2784
                                  instance.primary_node))
2785

    
2786
    self.op.os_type = getattr(self.op, "os_type", None)
2787
    if self.op.os_type is not None:
2788
      # OS verification
2789
      pnode = self.cfg.GetNodeInfo(
2790
        self.cfg.ExpandNodeName(instance.primary_node))
2791
      if pnode is None:
2792
        raise errors.OpPrereqError("Primary node '%s' is unknown" %
2793
                                   self.op.pnode)
2794
      result = self.rpc.call_os_get(pnode.name, self.op.os_type)
2795
      result.Raise()
2796
      if not isinstance(result.data, objects.OS):
2797
        raise errors.OpPrereqError("OS '%s' not in supported OS list for"
2798
                                   " primary node"  % self.op.os_type)
2799

    
2800
    self.instance = instance
2801

    
2802
  def Exec(self, feedback_fn):
2803
    """Reinstall the instance.
2804

2805
    """
2806
    inst = self.instance
2807

    
2808
    if self.op.os_type is not None:
2809
      feedback_fn("Changing OS to '%s'..." % self.op.os_type)
2810
      inst.os = self.op.os_type
2811
      self.cfg.Update(inst)
2812

    
2813
    _StartInstanceDisks(self, inst, None)
2814
    try:
2815
      feedback_fn("Running the instance OS create scripts...")
2816
      result = self.rpc.call_instance_os_add(inst.primary_node, inst)
2817
      result.Raise()
2818
      if not result.data:
2819
        raise errors.OpExecError("Could not install OS for instance %s"
2820
                                 " on node %s" %
2821
                                 (inst.name, inst.primary_node))
2822
    finally:
2823
      _ShutdownInstanceDisks(self, inst)
2824

    
2825

    
2826
class LURenameInstance(LogicalUnit):
2827
  """Rename an instance.
2828

2829
  """
2830
  HPATH = "instance-rename"
2831
  HTYPE = constants.HTYPE_INSTANCE
2832
  _OP_REQP = ["instance_name", "new_name"]
2833

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

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

2839
    """
2840
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2841
    env["INSTANCE_NEW_NAME"] = self.op.new_name
2842
    nl = ([self.cfg.GetMasterNode(), self.instance.primary_node] +
2843
          list(self.instance.secondary_nodes))
2844
    return env, nl, nl
2845

    
2846
  def CheckPrereq(self):
2847
    """Check prerequisites.
2848

2849
    This checks that the instance is in the cluster and is not running.
2850

2851
    """
2852
    instance = self.cfg.GetInstanceInfo(
2853
      self.cfg.ExpandInstanceName(self.op.instance_name))
2854
    if instance is None:
2855
      raise errors.OpPrereqError("Instance '%s' not known" %
2856
                                 self.op.instance_name)
2857
    _CheckNodeOnline(self, instance.primary_node)
2858

    
2859
    if instance.status != "down":
2860
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2861
                                 self.op.instance_name)
2862
    remote_info = self.rpc.call_instance_info(instance.primary_node,
2863
                                              instance.name,
2864
                                              instance.hypervisor)
2865
    remote_info.Raise()
2866
    if remote_info.data:
2867
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
2868
                                 (self.op.instance_name,
2869
                                  instance.primary_node))
2870
    self.instance = instance
2871

    
2872
    # new name verification
2873
    name_info = utils.HostInfo(self.op.new_name)
2874

    
2875
    self.op.new_name = new_name = name_info.name
2876
    instance_list = self.cfg.GetInstanceList()
2877
    if new_name in instance_list:
2878
      raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
2879
                                 new_name)
2880

    
2881
    if not getattr(self.op, "ignore_ip", False):
2882
      if utils.TcpPing(name_info.ip, constants.DEFAULT_NODED_PORT):
2883
        raise errors.OpPrereqError("IP %s of instance %s already in use" %
2884
                                   (name_info.ip, new_name))
2885

    
2886

    
2887
  def Exec(self, feedback_fn):
2888
    """Reinstall the instance.
2889

2890
    """
2891
    inst = self.instance
2892
    old_name = inst.name
2893

    
2894
    if inst.disk_template == constants.DT_FILE:
2895
      old_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
2896

    
2897
    self.cfg.RenameInstance(inst.name, self.op.new_name)
2898
    # Change the instance lock. This is definitely safe while we hold the BGL
2899
    self.context.glm.remove(locking.LEVEL_INSTANCE, old_name)
2900
    self.context.glm.add(locking.LEVEL_INSTANCE, self.op.new_name)
2901

    
2902
    # re-read the instance from the configuration after rename
2903
    inst = self.cfg.GetInstanceInfo(self.op.new_name)
2904

    
2905
    if inst.disk_template == constants.DT_FILE:
2906
      new_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
2907
      result = self.rpc.call_file_storage_dir_rename(inst.primary_node,
2908
                                                     old_file_storage_dir,
2909
                                                     new_file_storage_dir)
2910
      result.Raise()
2911
      if not result.data:
2912
        raise errors.OpExecError("Could not connect to node '%s' to rename"
2913
                                 " directory '%s' to '%s' (but the instance"
2914
                                 " has been renamed in Ganeti)" % (
2915
                                 inst.primary_node, old_file_storage_dir,
2916
                                 new_file_storage_dir))
2917

    
2918
      if not result.data[0]:
2919
        raise errors.OpExecError("Could not rename directory '%s' to '%s'"
2920
                                 " (but the instance has been renamed in"
2921
                                 " Ganeti)" % (old_file_storage_dir,
2922
                                               new_file_storage_dir))
2923

    
2924
    _StartInstanceDisks(self, inst, None)
2925
    try:
2926
      result = self.rpc.call_instance_run_rename(inst.primary_node, inst,
2927
                                                 old_name)
2928
      if result.failed or not result.data:
2929
        msg = ("Could not run OS rename script for instance %s on node %s"
2930
               " (but the instance has been renamed in Ganeti)" %
2931
               (inst.name, inst.primary_node))
2932
        self.proc.LogWarning(msg)
2933
    finally:
2934
      _ShutdownInstanceDisks(self, inst)
2935

    
2936

    
2937
class LURemoveInstance(LogicalUnit):
2938
  """Remove an instance.
2939

2940
  """
2941
  HPATH = "instance-remove"
2942
  HTYPE = constants.HTYPE_INSTANCE
2943
  _OP_REQP = ["instance_name", "ignore_failures"]
2944
  REQ_BGL = False
2945

    
2946
  def ExpandNames(self):
2947
    self._ExpandAndLockInstance()
2948
    self.needed_locks[locking.LEVEL_NODE] = []
2949
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2950

    
2951
  def DeclareLocks(self, level):
2952
    if level == locking.LEVEL_NODE:
2953
      self._LockInstancesNodes()
2954

    
2955
  def BuildHooksEnv(self):
2956
    """Build hooks env.
2957

2958
    This runs on master, primary and secondary nodes of the instance.
2959

2960
    """
2961
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2962
    nl = [self.cfg.GetMasterNode()]
2963
    return env, nl, nl
2964

    
2965
  def CheckPrereq(self):
2966
    """Check prerequisites.
2967

2968
    This checks that the instance is in the cluster.
2969

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

    
2975
  def Exec(self, feedback_fn):
2976
    """Remove the instance.
2977

2978
    """
2979
    instance = self.instance
2980
    logging.info("Shutting down instance %s on node %s",
2981
                 instance.name, instance.primary_node)
2982

    
2983
    result = self.rpc.call_instance_shutdown(instance.primary_node, instance)
2984
    if result.failed or not result.data:
2985
      if self.op.ignore_failures:
2986
        feedback_fn("Warning: can't shutdown instance")
2987
      else:
2988
        raise errors.OpExecError("Could not shutdown instance %s on node %s" %
2989
                                 (instance.name, instance.primary_node))
2990

    
2991
    logging.info("Removing block devices for instance %s", instance.name)
2992

    
2993
    if not _RemoveDisks(self, instance):
2994
      if self.op.ignore_failures:
2995
        feedback_fn("Warning: can't remove instance's disks")
2996
      else:
2997
        raise errors.OpExecError("Can't remove instance's disks")
2998

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

    
3001
    self.cfg.RemoveInstance(instance.name)
3002
    self.remove_locks[locking.LEVEL_INSTANCE] = instance.name
3003

    
3004

    
3005
class LUQueryInstances(NoHooksLU):
3006
  """Logical unit for querying instances.
3007

3008
  """
3009
  _OP_REQP = ["output_fields", "names"]
3010
  REQ_BGL = False
3011
  _FIELDS_STATIC = utils.FieldSet(*["name", "os", "pnode", "snodes",
3012
                                    "admin_state", "admin_ram",
3013
                                    "disk_template", "ip", "mac", "bridge",
3014
                                    "sda_size", "sdb_size", "vcpus", "tags",
3015
                                    "network_port", "beparams",
3016
                                    "(disk).(size)/([0-9]+)",
3017
                                    "(disk).(sizes)",
3018
                                    "(nic).(mac|ip|bridge)/([0-9]+)",
3019
                                    "(nic).(macs|ips|bridges)",
3020
                                    "(disk|nic).(count)",
3021
                                    "serial_no", "hypervisor", "hvparams",] +
3022
                                  ["hv/%s" % name
3023
                                   for name in constants.HVS_PARAMETERS] +
3024
                                  ["be/%s" % name
3025
                                   for name in constants.BES_PARAMETERS])
3026
  _FIELDS_DYNAMIC = utils.FieldSet("oper_state", "oper_ram", "status")
3027

    
3028

    
3029
  def ExpandNames(self):
3030
    _CheckOutputFields(static=self._FIELDS_STATIC,
3031
                       dynamic=self._FIELDS_DYNAMIC,
3032
                       selected=self.op.output_fields)
3033

    
3034
    self.needed_locks = {}
3035
    self.share_locks[locking.LEVEL_INSTANCE] = 1
3036
    self.share_locks[locking.LEVEL_NODE] = 1
3037

    
3038
    if self.op.names:
3039
      self.wanted = _GetWantedInstances(self, self.op.names)
3040
    else:
3041
      self.wanted = locking.ALL_SET
3042

    
3043
    self.do_locking = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
3044
    if self.do_locking:
3045
      self.needed_locks[locking.LEVEL_INSTANCE] = self.wanted
3046
      self.needed_locks[locking.LEVEL_NODE] = []
3047
      self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3048

    
3049
  def DeclareLocks(self, level):
3050
    if level == locking.LEVEL_NODE and self.do_locking:
3051
      self._LockInstancesNodes()
3052

    
3053
  def CheckPrereq(self):
3054
    """Check prerequisites.
3055

3056
    """
3057
    pass
3058

    
3059
  def Exec(self, feedback_fn):
3060
    """Computes the list of nodes and their attributes.
3061

3062
    """
3063
    all_info = self.cfg.GetAllInstancesInfo()
3064
    if self.do_locking:
3065
      instance_names = self.acquired_locks[locking.LEVEL_INSTANCE]
3066
    elif self.wanted != locking.ALL_SET:
3067
      instance_names = self.wanted
3068
      missing = set(instance_names).difference(all_info.keys())
3069
      if missing:
3070
        raise errors.OpExecError(
3071
          "Some instances were removed before retrieving their data: %s"
3072
          % missing)
3073
    else:
3074
      instance_names = all_info.keys()
3075

    
3076
    instance_names = utils.NiceSort(instance_names)
3077
    instance_list = [all_info[iname] for iname in instance_names]
3078

    
3079
    # begin data gathering
3080

    
3081
    nodes = frozenset([inst.primary_node for inst in instance_list])
3082
    hv_list = list(set([inst.hypervisor for inst in instance_list]))
3083

    
3084
    bad_nodes = []
3085
    off_nodes = []
3086
    if self.do_locking:
3087
      live_data = {}
3088
      node_data = self.rpc.call_all_instances_info(nodes, hv_list)
3089
      for name in nodes:
3090
        result = node_data[name]
3091
        if result.offline:
3092
          # offline nodes will be in both lists
3093
          off_nodes.append(name)
3094
        if result.failed:
3095
          bad_nodes.append(name)
3096
        else:
3097
          if result.data:
3098
            live_data.update(result.data)
3099
            # else no instance is alive
3100
    else:
3101
      live_data = dict([(name, {}) for name in instance_names])
3102

    
3103
    # end data gathering
3104

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

    
3230
    return output
3231

    
3232

    
3233
class LUFailoverInstance(LogicalUnit):
3234
  """Failover an instance.
3235

3236
  """
3237
  HPATH = "instance-failover"
3238
  HTYPE = constants.HTYPE_INSTANCE
3239
  _OP_REQP = ["instance_name", "ignore_consistency"]
3240
  REQ_BGL = False
3241

    
3242
  def ExpandNames(self):
3243
    self._ExpandAndLockInstance()
3244
    self.needed_locks[locking.LEVEL_NODE] = []
3245
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3246

    
3247
  def DeclareLocks(self, level):
3248
    if level == locking.LEVEL_NODE:
3249
      self._LockInstancesNodes()
3250

    
3251
  def BuildHooksEnv(self):
3252
    """Build hooks env.
3253

3254
    This runs on master, primary and secondary nodes of the instance.
3255

3256
    """
3257
    env = {
3258
      "IGNORE_CONSISTENCY": self.op.ignore_consistency,
3259
      }
3260
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3261
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3262
    return env, nl, nl
3263

    
3264
  def CheckPrereq(self):
3265
    """Check prerequisites.
3266

3267
    This checks that the instance is in the cluster.
3268

3269
    """
3270
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3271
    assert self.instance is not None, \
3272
      "Cannot retrieve locked instance %s" % self.op.instance_name
3273

    
3274
    bep = self.cfg.GetClusterInfo().FillBE(instance)
3275
    if instance.disk_template not in constants.DTS_NET_MIRROR:
3276
      raise errors.OpPrereqError("Instance's disk layout is not"
3277
                                 " network mirrored, cannot failover.")
3278

    
3279
    secondary_nodes = instance.secondary_nodes
3280
    if not secondary_nodes:
3281
      raise errors.ProgrammerError("no secondary node but using "
3282
                                   "a mirrored disk template")
3283

    
3284
    target_node = secondary_nodes[0]
3285
    _CheckNodeOnline(self, target_node)
3286
    # check memory requirements on the secondary node
3287
    _CheckNodeFreeMemory(self, target_node, "failing over instance %s" %
3288
                         instance.name, bep[constants.BE_MEMORY],
3289
                         instance.hypervisor)
3290

    
3291
    # check bridge existance
3292
    brlist = [nic.bridge for nic in instance.nics]
3293
    result = self.rpc.call_bridges_exist(target_node, brlist)
3294
    result.Raise()
3295
    if not result.data:
3296
      raise errors.OpPrereqError("One or more target bridges %s does not"
3297
                                 " exist on destination node '%s'" %
3298
                                 (brlist, target_node))
3299

    
3300
  def Exec(self, feedback_fn):
3301
    """Failover an instance.
3302

3303
    The failover is done by shutting it down on its present node and
3304
    starting it on the secondary.
3305

3306
    """
3307
    instance = self.instance
3308

    
3309
    source_node = instance.primary_node
3310
    target_node = instance.secondary_nodes[0]
3311

    
3312
    feedback_fn("* checking disk consistency between source and target")
3313
    for dev in instance.disks:
3314
      # for drbd, these are drbd over lvm
3315
      if not _CheckDiskConsistency(self, dev, target_node, False):
3316
        if instance.status == "up" and not self.op.ignore_consistency:
3317
          raise errors.OpExecError("Disk %s is degraded on target node,"
3318
                                   " aborting failover." % dev.iv_name)
3319

    
3320
    feedback_fn("* shutting down instance on source node")
3321
    logging.info("Shutting down instance %s on node %s",
3322
                 instance.name, source_node)
3323

    
3324
    result = self.rpc.call_instance_shutdown(source_node, instance)
3325
    if result.failed or not result.data:
3326
      if self.op.ignore_consistency:
3327
        self.proc.LogWarning("Could not shutdown instance %s on node %s."
3328
                             " Proceeding"
3329
                             " anyway. Please make sure node %s is down",
3330
                             instance.name, source_node, source_node)
3331
      else:
3332
        raise errors.OpExecError("Could not shutdown instance %s on node %s" %
3333
                                 (instance.name, source_node))
3334

    
3335
    feedback_fn("* deactivating the instance's disks on source node")
3336
    if not _ShutdownInstanceDisks(self, instance, ignore_primary=True):
3337
      raise errors.OpExecError("Can't shut down the instance's disks.")
3338

    
3339
    instance.primary_node = target_node
3340
    # distribute new instance config to the other nodes
3341
    self.cfg.Update(instance)
3342

    
3343
    # Only start the instance if it's marked as up
3344
    if instance.status == "up":
3345
      feedback_fn("* activating the instance's disks on target node")
3346
      logging.info("Starting instance %s on node %s",
3347
                   instance.name, target_node)
3348

    
3349
      disks_ok, dummy = _AssembleInstanceDisks(self, instance,
3350
                                               ignore_secondaries=True)
3351
      if not disks_ok:
3352
        _ShutdownInstanceDisks(self, instance)
3353
        raise errors.OpExecError("Can't activate the instance's disks")
3354

    
3355
      feedback_fn("* starting the instance on the target node")
3356
      result = self.rpc.call_instance_start(target_node, instance, None)
3357
      if result.failed or not result.data:
3358
        _ShutdownInstanceDisks(self, instance)
3359
        raise errors.OpExecError("Could not start instance %s on node %s." %
3360
                                 (instance.name, target_node))
3361

    
3362

    
3363
def _CreateBlockDevOnPrimary(lu, node, instance, device, info):
3364
  """Create a tree of block devices on the primary node.
3365

3366
  This always creates all devices.
3367

3368
  """
3369
  if device.children:
3370
    for child in device.children:
3371
      if not _CreateBlockDevOnPrimary(lu, node, instance, child, info):
3372
        return False
3373

    
3374
  lu.cfg.SetDiskID(device, node)
3375
  new_id = lu.rpc.call_blockdev_create(node, device, device.size,
3376
                                       instance.name, True, info)
3377
  if new_id.failed or not new_id.data:
3378
    return False
3379
  if device.physical_id is None:
3380
    device.physical_id = new_id
3381
  return True
3382

    
3383

    
3384
def _CreateBlockDevOnSecondary(lu, node, instance, device, force, info):
3385
  """Create a tree of block devices on a secondary node.
3386

3387
  If this device type has to be created on secondaries, create it and
3388
  all its children.
3389

3390
  If not, just recurse to children keeping the same 'force' value.
3391

3392
  """
3393
  if device.CreateOnSecondary():
3394
    force = True
3395
  if device.children:
3396
    for child in device.children:
3397
      if not _CreateBlockDevOnSecondary(lu, node, instance,
3398
                                        child, force, info):
3399
        return False
3400

    
3401
  if not force:
3402
    return True
3403
  lu.cfg.SetDiskID(device, node)
3404
  new_id = lu.rpc.call_blockdev_create(node, device, device.size,
3405
                                       instance.name, False, info)
3406
  if new_id.failed or not new_id.data:
3407
    return False
3408
  if device.physical_id is None:
3409
    device.physical_id = new_id
3410
  return True
3411

    
3412

    
3413
def _GenerateUniqueNames(lu, exts):
3414
  """Generate a suitable LV name.
3415

3416
  This will generate a logical volume name for the given instance.
3417

3418
  """
3419
  results = []
3420
  for val in exts:
3421
    new_id = lu.cfg.GenerateUniqueID()
3422
    results.append("%s%s" % (new_id, val))
3423
  return results
3424

    
3425

    
3426
def _GenerateDRBD8Branch(lu, primary, secondary, size, names, iv_name,
3427
                         p_minor, s_minor):
3428
  """Generate a drbd8 device complete with its children.
3429

3430
  """
3431
  port = lu.cfg.AllocatePort()
3432
  vgname = lu.cfg.GetVGName()
3433
  shared_secret = lu.cfg.GenerateDRBDSecret()
3434
  dev_data = objects.Disk(dev_type=constants.LD_LV, size=size,
3435
                          logical_id=(vgname, names[0]))
3436
  dev_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
3437
                          logical_id=(vgname, names[1]))
3438
  drbd_dev = objects.Disk(dev_type=constants.LD_DRBD8, size=size,
3439
                          logical_id=(primary, secondary, port,
3440
                                      p_minor, s_minor,
3441
                                      shared_secret),
3442
                          children=[dev_data, dev_meta],
3443
                          iv_name=iv_name)
3444
  return drbd_dev
3445

    
3446

    
3447
def _GenerateDiskTemplate(lu, template_name,
3448
                          instance_name, primary_node,
3449
                          secondary_nodes, disk_info,
3450
                          file_storage_dir, file_driver,
3451
                          base_index):
3452
  """Generate the entire disk layout for a given template type.
3453

3454
  """
3455
  #TODO: compute space requirements
3456

    
3457
  vgname = lu.cfg.GetVGName()
3458
  disk_count = len(disk_info)
3459
  disks = []
3460
  if template_name == constants.DT_DISKLESS:
3461
    pass
3462
  elif template_name == constants.DT_PLAIN:
3463
    if len(secondary_nodes) != 0:
3464
      raise errors.ProgrammerError("Wrong template configuration")
3465

    
3466
    names = _GenerateUniqueNames(lu, [".disk%d" % i
3467
                                      for i in range(disk_count)])
3468
    for idx, disk in enumerate(disk_info):
3469
      disk_index = idx + base_index
3470
      disk_dev = objects.Disk(dev_type=constants.LD_LV, size=disk["size"],
3471
                              logical_id=(vgname, names[idx]),
3472
                              iv_name="disk/%d" % disk_index)
3473
      disks.append(disk_dev)
3474
  elif template_name == constants.DT_DRBD8:
3475
    if len(secondary_nodes) != 1:
3476
      raise errors.ProgrammerError("Wrong template configuration")
3477
    remote_node = secondary_nodes[0]
3478
    minors = lu.cfg.AllocateDRBDMinor(
3479
      [primary_node, remote_node] * len(disk_info), instance_name)
3480

    
3481
    names = _GenerateUniqueNames(lu,
3482
                                 [".disk%d_%s" % (i, s)
3483
                                  for i in range(disk_count)
3484
                                  for s in ("data", "meta")
3485
                                  ])
3486
    for idx, disk in enumerate(disk_info):
3487
      disk_index = idx + base_index
3488
      disk_dev = _GenerateDRBD8Branch(lu, primary_node, remote_node,
3489
                                      disk["size"], names[idx*2:idx*2+2],
3490
                                      "disk/%d" % disk_index,
3491
                                      minors[idx*2], minors[idx*2+1])
3492
      disks.append(disk_dev)
3493
  elif template_name == constants.DT_FILE:
3494
    if len(secondary_nodes) != 0:
3495
      raise errors.ProgrammerError("Wrong template configuration")
3496

    
3497
    for idx, disk in enumerate(disk_info):
3498
      disk_index = idx + base_index
3499
      disk_dev = objects.Disk(dev_type=constants.LD_FILE, size=disk["size"],
3500
                              iv_name="disk/%d" % disk_index,
3501
                              logical_id=(file_driver,
3502
                                          "%s/disk%d" % (file_storage_dir,
3503
                                                         idx)))
3504
      disks.append(disk_dev)
3505
  else:
3506
    raise errors.ProgrammerError("Invalid disk template '%s'" % template_name)
3507
  return disks
3508

    
3509

    
3510
def _GetInstanceInfoText(instance):
3511
  """Compute that text that should be added to the disk's metadata.
3512

3513
  """
3514
  return "originstname+%s" % instance.name
3515

    
3516

    
3517
def _CreateDisks(lu, instance):
3518
  """Create all disks for an instance.
3519

3520
  This abstracts away some work from AddInstance.
3521

3522
  @type lu: L{LogicalUnit}
3523
  @param lu: the logical unit on whose behalf we execute
3524
  @type instance: L{objects.Instance}
3525
  @param instance: the instance whose disks we should create
3526
  @rtype: boolean
3527
  @return: the success of the creation
3528

3529
  """
3530
  info = _GetInstanceInfoText(instance)
3531

    
3532
  if instance.disk_template == constants.DT_FILE:
3533
    file_storage_dir = os.path.dirname(instance.disks[0].logical_id[1])
3534
    result = lu.rpc.call_file_storage_dir_create(instance.primary_node,
3535
                                                 file_storage_dir)
3536

    
3537
    if result.failed or not result.data:
3538
      logging.error("Could not connect to node '%s'", instance.primary_node)
3539
      return False
3540

    
3541
    if not result.data[0]:
3542
      logging.error("Failed to create directory '%s'", file_storage_dir)
3543
      return False
3544

    
3545
  # Note: this needs to be kept in sync with adding of disks in
3546
  # LUSetInstanceParams
3547
  for device in instance.disks:
3548
    logging.info("Creating volume %s for instance %s",
3549
                 device.iv_name, instance.name)
3550
    #HARDCODE
3551
    for secondary_node in instance.secondary_nodes:
3552
      if not _CreateBlockDevOnSecondary(lu, secondary_node, instance,
3553
                                        device, False, info):
3554
        logging.error("Failed to create volume %s (%s) on secondary node %s!",
3555
                      device.iv_name, device, secondary_node)
3556
        return False
3557
    #HARDCODE
3558
    if not _CreateBlockDevOnPrimary(lu, instance.primary_node,
3559
                                    instance, device, info):
3560
      logging.error("Failed to create volume %s on primary!", device.iv_name)
3561
      return False
3562

    
3563
  return True
3564

    
3565

    
3566
def _RemoveDisks(lu, instance):
3567
  """Remove all disks for an instance.
3568

3569
  This abstracts away some work from `AddInstance()` and
3570
  `RemoveInstance()`. Note that in case some of the devices couldn't
3571
  be removed, the removal will continue with the other ones (compare
3572
  with `_CreateDisks()`).
3573

3574
  @type lu: L{LogicalUnit}
3575
  @param lu: the logical unit on whose behalf we execute
3576
  @type instance: L{objects.Instance}
3577
  @param instance: the instance whose disks we should remove
3578
  @rtype: boolean
3579
  @return: the success of the removal
3580

3581
  """
3582
  logging.info("Removing block devices for instance %s", instance.name)
3583

    
3584
  result = True
3585
  for device in instance.disks:
3586
    for node, disk in device.ComputeNodeTree(instance.primary_node):
3587
      lu.cfg.SetDiskID(disk, node)
3588
      result = lu.rpc.call_blockdev_remove(node, disk)
3589
      if result.failed or not result.data:
3590
        lu.proc.LogWarning("Could not remove block device %s on node %s,"
3591
                           " continuing anyway", device.iv_name, node)
3592
        result = False
3593

    
3594
  if instance.disk_template == constants.DT_FILE:
3595
    file_storage_dir = os.path.dirname(instance.disks[0].logical_id[1])
3596
    result = lu.rpc.call_file_storage_dir_remove(instance.primary_node,
3597
                                                 file_storage_dir)
3598
    if result.failed or not result.data:
3599
      logging.error("Could not remove directory '%s'", file_storage_dir)
3600
      result = False
3601

    
3602
  return result
3603

    
3604

    
3605
def _ComputeDiskSize(disk_template, disks):
3606
  """Compute disk size requirements in the volume group
3607

3608
  """
3609
  # Required free disk space as a function of disk and swap space
3610
  req_size_dict = {
3611
    constants.DT_DISKLESS: None,
3612
    constants.DT_PLAIN: sum(d["size"] for d in disks),
3613
    # 128 MB are added for drbd metadata for each disk
3614
    constants.DT_DRBD8: sum(d["size"] + 128 for d in disks),
3615
    constants.DT_FILE: None,
3616
  }
3617

    
3618
  if disk_template not in req_size_dict:
3619
    raise errors.ProgrammerError("Disk template '%s' size requirement"
3620
                                 " is unknown" %  disk_template)
3621

    
3622
  return req_size_dict[disk_template]
3623

    
3624

    
3625
def _CheckHVParams(lu, nodenames, hvname, hvparams):
3626
  """Hypervisor parameter validation.
3627

3628
  This function abstract the hypervisor parameter validation to be
3629
  used in both instance create and instance modify.
3630

3631
  @type lu: L{LogicalUnit}
3632
  @param lu: the logical unit for which we check
3633
  @type nodenames: list
3634
  @param nodenames: the list of nodes on which we should check
3635
  @type hvname: string
3636
  @param hvname: the name of the hypervisor we should use
3637
  @type hvparams: dict
3638
  @param hvparams: the parameters which we need to check
3639
  @raise errors.OpPrereqError: if the parameters are not valid
3640

3641
  """
3642
  hvinfo = lu.rpc.call_hypervisor_validate_params(nodenames,
3643
                                                  hvname,
3644
                                                  hvparams)
3645
  for node in nodenames:
3646
    info = hvinfo[node]
3647
    info.Raise()
3648
    if not info.data or not isinstance(info.data, (tuple, list)):
3649
      raise errors.OpPrereqError("Cannot get current information"
3650
                                 " from node '%s' (%s)" % (node, info.data))
3651
    if not info.data[0]:
3652
      raise errors.OpPrereqError("Hypervisor parameter validation failed:"
3653
                                 " %s" % info.data[1])
3654

    
3655

    
3656
class LUCreateInstance(LogicalUnit):
3657
  """Create an instance.
3658

3659
  """
3660
  HPATH = "instance-add"
3661
  HTYPE = constants.HTYPE_INSTANCE
3662
  _OP_REQP = ["instance_name", "disks", "disk_template",
3663
              "mode", "start",
3664
              "wait_for_sync", "ip_check", "nics",
3665
              "hvparams", "beparams"]
3666
  REQ_BGL = False
3667

    
3668
  def _ExpandNode(self, node):
3669
    """Expands and checks one node name.
3670

3671
    """
3672
    node_full = self.cfg.ExpandNodeName(node)
3673
    if node_full is None:
3674
      raise errors.OpPrereqError("Unknown node %s" % node)
3675
    return node_full
3676

    
3677
  def ExpandNames(self):
3678
    """ExpandNames for CreateInstance.
3679

3680
    Figure out the right locks for instance creation.
3681

3682
    """
3683
    self.needed_locks = {}
3684

    
3685
    # set optional parameters to none if they don't exist
3686
    for attr in ["pnode", "snode", "iallocator", "hypervisor"]:
3687
      if not hasattr(self.op, attr):
3688
        setattr(self.op, attr, None)
3689

    
3690
    # cheap checks, mostly valid constants given
3691

    
3692
    # verify creation mode
3693
    if self.op.mode not in (constants.INSTANCE_CREATE,
3694
                            constants.INSTANCE_IMPORT):
3695
      raise errors.OpPrereqError("Invalid instance creation mode '%s'" %
3696
                                 self.op.mode)
3697

    
3698
    # disk template and mirror node verification
3699
    if self.op.disk_template not in constants.DISK_TEMPLATES:
3700
      raise errors.OpPrereqError("Invalid disk template name")
3701

    
3702
    if self.op.hypervisor is None:
3703
      self.op.hypervisor = self.cfg.GetHypervisorType()
3704

    
3705
    cluster = self.cfg.GetClusterInfo()
3706
    enabled_hvs = cluster.enabled_hypervisors
3707
    if self.op.hypervisor not in enabled_hvs:
3708
      raise errors.OpPrereqError("Selected hypervisor (%s) not enabled in the"
3709
                                 " cluster (%s)" % (self.op.hypervisor,
3710
                                  ",".join(enabled_hvs)))
3711

    
3712
    # check hypervisor parameter syntax (locally)
3713

    
3714
    filled_hvp = cluster.FillDict(cluster.hvparams[self.op.hypervisor],
3715
                                  self.op.hvparams)
3716
    hv_type = hypervisor.GetHypervisor(self.op.hypervisor)
3717
    hv_type.CheckParameterSyntax(filled_hvp)
3718

    
3719
    # fill and remember the beparams dict
3720
    utils.CheckBEParams(self.op.beparams)
3721
    self.be_full = cluster.FillDict(cluster.beparams[constants.BEGR_DEFAULT],
3722
                                    self.op.beparams)
3723

    
3724
    #### instance parameters check
3725

    
3726
    # instance name verification
3727
    hostname1 = utils.HostInfo(self.op.instance_name)
3728
    self.op.instance_name = instance_name = hostname1.name
3729

    
3730
    # this is just a preventive check, but someone might still add this
3731
    # instance in the meantime, and creation will fail at lock-add time
3732
    if instance_name in self.cfg.GetInstanceList():
3733
      raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
3734
                                 instance_name)
3735

    
3736
    self.add_locks[locking.LEVEL_INSTANCE] = instance_name
3737

    
3738
    # NIC buildup
3739
    self.nics = []
3740
    for nic in self.op.nics:
3741
      # ip validity checks
3742
      ip = nic.get("ip", None)
3743
      if ip is None or ip.lower() == "none":
3744
        nic_ip = None
3745
      elif ip.lower() == constants.VALUE_AUTO:
3746
        nic_ip = hostname1.ip
3747
      else:
3748
        if not utils.IsValidIP(ip):
3749
          raise errors.OpPrereqError("Given IP address '%s' doesn't look"
3750
                                     " like a valid IP" % ip)
3751
        nic_ip = ip
3752

    
3753
      # MAC address verification
3754
      mac = nic.get("mac", constants.VALUE_AUTO)
3755
      if mac not in (constants.VALUE_AUTO, constants.VALUE_GENERATE):
3756
        if not utils.IsValidMac(mac.lower()):
3757
          raise errors.OpPrereqError("Invalid MAC address specified: %s" %
3758
                                     mac)
3759
      # bridge verification
3760
      bridge = nic.get("bridge", self.cfg.GetDefBridge())
3761
      self.nics.append(objects.NIC(mac=mac, ip=nic_ip, bridge=bridge))
3762

    
3763
    # disk checks/pre-build
3764
    self.disks = []
3765
    for disk in self.op.disks:
3766
      mode = disk.get("mode", constants.DISK_RDWR)
3767
      if mode not in constants.DISK_ACCESS_SET:
3768
        raise errors.OpPrereqError("Invalid disk access mode '%s'" %
3769
                                   mode)
3770
      size = disk.get("size", None)
3771
      if size is None:
3772
        raise errors.OpPrereqError("Missing disk size")
3773
      try:
3774
        size = int(size)
3775
      except ValueError:
3776
        raise errors.OpPrereqError("Invalid disk size '%s'" % size)
3777
      self.disks.append({"size": size, "mode": mode})
3778

    
3779
    # used in CheckPrereq for ip ping check
3780
    self.check_ip = hostname1.ip
3781

    
3782
    # file storage checks
3783
    if (self.op.file_driver and
3784
        not self.op.file_driver in constants.FILE_DRIVER):
3785
      raise errors.OpPrereqError("Invalid file driver name '%s'" %
3786
                                 self.op.file_driver)
3787

    
3788
    if self.op.file_storage_dir and os.path.isabs(self.op.file_storage_dir):
3789
      raise errors.OpPrereqError("File storage directory path not absolute")
3790

    
3791
    ### Node/iallocator related checks
3792
    if [self.op.iallocator, self.op.pnode].count(None) != 1:
3793
      raise errors.OpPrereqError("One and only one of iallocator and primary"
3794
                                 " node must be given")
3795

    
3796
    if self.op.iallocator:
3797
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
3798
    else:
3799
      self.op.pnode = self._ExpandNode(self.op.pnode)
3800
      nodelist = [self.op.pnode]
3801
      if self.op.snode is not None:
3802
        self.op.snode = self._ExpandNode(self.op.snode)
3803
        nodelist.append(self.op.snode)
3804
      self.needed_locks[locking.LEVEL_NODE] = nodelist
3805

    
3806
    # in case of import lock the source node too
3807
    if self.op.mode == constants.INSTANCE_IMPORT:
3808
      src_node = getattr(self.op, "src_node", None)
3809
      src_path = getattr(self.op, "src_path", None)
3810

    
3811
      if src_path is None:
3812
        self.op.src_path = src_path = self.op.instance_name
3813

    
3814
      if src_node is None:
3815
        self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
3816
        self.op.src_node = None
3817
        if os.path.isabs(src_path):
3818
          raise errors.OpPrereqError("Importing an instance from an absolute"
3819
                                     " path requires a source node option.")
3820
      else:
3821
        self.op.src_node = src_node = self._ExpandNode(src_node)
3822
        if self.needed_locks[locking.LEVEL_NODE] is not locking.ALL_SET:
3823
          self.needed_locks[locking.LEVEL_NODE].append(src_node)
3824
        if not os.path.isabs(src_path):
3825
          self.op.src_path = src_path = \
3826
            os.path.join(constants.EXPORT_DIR, src_path)
3827

    
3828
    else: # INSTANCE_CREATE
3829
      if getattr(self.op, "os_type", None) is None:
3830
        raise errors.OpPrereqError("No guest OS specified")
3831

    
3832
  def _RunAllocator(self):
3833
    """Run the allocator based on input opcode.
3834

3835
    """
3836
    nics = [n.ToDict() for n in self.nics]
3837
    ial = IAllocator(self,
3838
                     mode=constants.IALLOCATOR_MODE_ALLOC,
3839
                     name=self.op.instance_name,
3840
                     disk_template=self.op.disk_template,
3841
                     tags=[],
3842
                     os=self.op.os_type,
3843
                     vcpus=self.be_full[constants.BE_VCPUS],
3844
                     mem_size=self.be_full[constants.BE_MEMORY],
3845
                     disks=self.disks,
3846
                     nics=nics,
3847
                     hypervisor=self.op.hypervisor,
3848
                     )
3849

    
3850
    ial.Run(self.op.iallocator)
3851

    
3852
    if not ial.success:
3853
      raise errors.OpPrereqError("Can't compute nodes using"
3854
                                 " iallocator '%s': %s" % (self.op.iallocator,
3855
                                                           ial.info))
3856
    if len(ial.nodes) != ial.required_nodes:
3857
      raise errors.OpPrereqError("iallocator '%s' returned invalid number"
3858
                                 " of nodes (%s), required %s" %
3859
                                 (self.op.iallocator, len(ial.nodes),
3860
                                  ial.required_nodes))
3861
    self.op.pnode = ial.nodes[0]
3862
    self.LogInfo("Selected nodes for instance %s via iallocator %s: %s",
3863
                 self.op.instance_name, self.op.iallocator,
3864
                 ", ".join(ial.nodes))
3865
    if ial.required_nodes == 2:
3866
      self.op.snode = ial.nodes[1]
3867

    
3868
  def BuildHooksEnv(self):
3869
    """Build hooks env.
3870

3871
    This runs on master, primary and secondary nodes of the instance.
3872

3873
    """
3874
    env = {
3875
      "INSTANCE_DISK_TEMPLATE": self.op.disk_template,
3876
      "INSTANCE_DISK_SIZE": ",".join(str(d["size"]) for d in self.disks),
3877
      "INSTANCE_ADD_MODE": self.op.mode,
3878
      }
3879
    if self.op.mode == constants.INSTANCE_IMPORT:
3880
      env["INSTANCE_SRC_NODE"] = self.op.src_node
3881
      env["INSTANCE_SRC_PATH"] = self.op.src_path
3882
      env["INSTANCE_SRC_IMAGES"] = self.src_images
3883

    
3884
    env.update(_BuildInstanceHookEnv(name=self.op.instance_name,
3885
      primary_node=self.op.pnode,
3886
      secondary_nodes=self.secondaries,
3887
      status=self.instance_status,
3888
      os_type=self.op.os_type,
3889
      memory=self.be_full[constants.BE_MEMORY],
3890
      vcpus=self.be_full[constants.BE_VCPUS],
3891
      nics=[(n.ip, n.bridge, n.mac) for n in self.nics],
3892
    ))
3893

    
3894
    nl = ([self.cfg.GetMasterNode(), self.op.pnode] +
3895
          self.secondaries)
3896
    return env, nl, nl
3897

    
3898

    
3899
  def CheckPrereq(self):
3900
    """Check prerequisites.
3901

3902
    """
3903
    if (not self.cfg.GetVGName() and
3904
        self.op.disk_template not in constants.DTS_NOT_LVM):
3905
      raise errors.OpPrereqError("Cluster does not support lvm-based"
3906
                                 " instances")
3907

    
3908

    
3909
    if self.op.mode == constants.INSTANCE_IMPORT:
3910
      src_node = self.op.src_node
3911
      src_path = self.op.src_path
3912

    
3913
      if src_node is None:
3914
        exp_list = self.rpc.call_export_list(
3915
          self.acquired_locks[locking.LEVEL_NODE])
3916
        found = False
3917
        for node in exp_list:
3918
          if not exp_list[node].failed and src_path in exp_list[node].data:
3919
            found = True
3920
            self.op.src_node = src_node = node
3921
            self.op.src_path = src_path = os.path.join(constants.EXPORT_DIR,
3922
                                                       src_path)
3923
            break
3924
        if not found:
3925
          raise errors.OpPrereqError("No export found for relative path %s" %
3926
                                      src_path)
3927

    
3928
      _CheckNodeOnline(self, src_node)
3929
      result = self.rpc.call_export_info(src_node, src_path)
3930
      result.Raise()
3931
      if not result.data:
3932
        raise errors.OpPrereqError("No export found in dir %s" % src_path)
3933

    
3934
      export_info = result.data
3935
      if not export_info.has_section(constants.INISECT_EXP):
3936
        raise errors.ProgrammerError("Corrupted export config")
3937

    
3938
      ei_version = export_info.get(constants.INISECT_EXP, 'version')
3939
      if (int(ei_version) != constants.EXPORT_VERSION):
3940
        raise errors.OpPrereqError("Wrong export version %s (wanted %d)" %
3941
                                   (ei_version, constants.EXPORT_VERSION))
3942

    
3943
      # Check that the new instance doesn't have less disks than the export
3944
      instance_disks = len(self.disks)
3945
      export_disks = export_info.getint(constants.INISECT_INS, 'disk_count')
3946
      if instance_disks < export_disks:
3947
        raise errors.OpPrereqError("Not enough disks to import."
3948
                                   " (instance: %d, export: %d)" %
3949
                                   (instance_disks, export_disks))
3950

    
3951
      self.op.os_type = export_info.get(constants.INISECT_EXP, 'os')
3952
      disk_images = []
3953
      for idx in range(export_disks):
3954
        option = 'disk%d_dump' % idx
3955
        if export_info.has_option(constants.INISECT_INS, option):
3956
          # FIXME: are the old os-es, disk sizes, etc. useful?
3957
          export_name = export_info.get(constants.INISECT_INS, option)
3958
          image = os.path.join(src_path, export_name)
3959
          disk_images.append(image)
3960
        else:
3961
          disk_images.append(False)
3962

    
3963
      self.src_images = disk_images
3964

    
3965
      old_name = export_info.get(constants.INISECT_INS, 'name')
3966
      # FIXME: int() here could throw a ValueError on broken exports
3967
      exp_nic_count = int(export_info.get(constants.INISECT_INS, 'nic_count'))
3968
      if self.op.instance_name == old_name:
3969
        for idx, nic in enumerate(self.nics):
3970
          if nic.mac == constants.VALUE_AUTO and exp_nic_count >= idx:
3971
            nic_mac_ini = 'nic%d_mac' % idx
3972
            nic.mac = export_info.get(constants.INISECT_INS, nic_mac_ini)
3973

    
3974
    # ip ping checks (we use the same ip that was resolved in ExpandNames)
3975
    if self.op.start and not self.op.ip_check:
3976
      raise errors.OpPrereqError("Cannot ignore IP address conflicts when"
3977
                                 " adding an instance in start mode")
3978

    
3979
    if self.op.ip_check:
3980
      if utils.TcpPing(self.check_ip, constants.DEFAULT_NODED_PORT):
3981
        raise errors.OpPrereqError(