Statistics
| Branch: | Tag: | Revision:

root / lib / cmdlib.py @ 998c712c

History | View | Annotate | Download (229 kB)

1
#
2
#
3

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

    
21

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

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

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

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

    
48

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

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

61
  Note that all commands require root permissions.
62

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

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

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

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

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

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

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

    
109
  ssh = property(fget=__GetSSH)
110

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

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

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

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

126
    """
127
    pass
128

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

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

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

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

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

150
    Examples::
151

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

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

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

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

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

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

189
    """
190

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

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

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

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

205
    """
206
    raise NotImplementedError
207

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

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

215
    """
216
    raise NotImplementedError
217

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

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

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

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

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

237
    """
238
    raise NotImplementedError
239

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

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

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

258
    """
259
    return lu_result
260

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

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

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

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

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

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

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

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

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

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

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

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

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

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

    
326
    del self.recalculate_locks[locking.LEVEL_NODE]
327

    
328

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

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

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

    
339

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

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

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

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

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

    
366
  return utils.NiceSort(wanted)
367

    
368

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

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

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

    
385
  if instances:
386
    wanted = []
387

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

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

    
398

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

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

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

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

    
417

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

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

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

    
431

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

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

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

    
443

    
444
def _BuildInstanceHookEnv(name, primary_node, secondary_nodes, os_type, status,
445
                          memory, vcpus, nics):
446
  """Builds instance related env variables for hooks
447

448
  This builds the hook environment from individual variables.
449

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

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

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

    
493
  env["INSTANCE_NIC_COUNT"] = nic_count
494

    
495
  return env
496

    
497

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

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

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

    
528

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

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

    
544

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

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

    
558

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

562
  """
563
  _OP_REQP = []
564

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

568
    This checks whether the cluster is empty.
569

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

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

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

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

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

    
598

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

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

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

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

619
    Test list:
620

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

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

634
    """
635
    node = nodeinfo.name
636

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

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

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

    
654
    # checks vg existance and size > 20G
655

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

    
669
    # checks config file checksum
670

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

    
698
    # checks ssh to any
699

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

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

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

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

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

736
    """
737
    bad = False
738

    
739
    node_current = instanceconfig.primary_node
740

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

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

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

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

    
769
    return bad
770

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

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

777
    """
778
    bad = False
779

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

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

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

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

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

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

809
    """
810
    bad = False
811

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

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

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

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

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

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

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

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

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

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

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

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

    
887
    local_checksums = utils.FingerprintFiles(file_names)
888

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

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

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

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

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

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

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

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

    
956
      node_instance[node] = idata
957

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

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

    
984
    node_vol_should = {}
985

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

    
994
      inst_config.MapLVsByNode(node_vol_should)
995

    
996
      instance_cfg[instance] = inst_config
997

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

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

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

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

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

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

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

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

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

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

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

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

    
1069
    return not bad
1070

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

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

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

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

    
1118
      return lu_result
1119

    
1120

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

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

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

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

1138
    This has no prerequisites.
1139

1140
    """
1141
    pass
1142

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

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

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

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

    
1166
    if not nv_dict:
1167
      return result
1168

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

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

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

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

    
1203
    return result
1204

    
1205

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

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

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

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

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

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

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

    
1244
    self.op.name = new_name
1245

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

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

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

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

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

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

    
1285

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

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

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

    
1301

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

    
1439

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

1443
  This is a very simple LU.
1444

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

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

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

1458
    """
1459

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

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

    
1466

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

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

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

    
1477
  node = instance.primary_node
1478

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

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

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

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

    
1525

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

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

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

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

    
1552
  return result
1553

    
1554

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

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

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

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

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

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

1580
    """
1581

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

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

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

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

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

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

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

    
1640
    return output
1641

    
1642

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

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

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

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

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

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

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

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

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

    
1681
    instance_list = self.cfg.GetInstanceList()
1682

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

    
1688
    for instance_name in instance_list:
1689
      instance = self.cfg.GetInstanceInfo(instance_name)
1690
      if node.name == instance.primary_node:
1691
        raise errors.OpPrereqError("Instance %s still running on the node,"
1692
                                   " please remove first." % instance_name)
1693
      if node.name in instance.secondary_nodes:
1694
        raise errors.OpPrereqError("Instance %s has node as a secondary,"
1695
                                   " please remove first." % instance_name)
1696
    self.op.node_name = node.name
1697
    self.node = node
1698

    
1699
  def Exec(self, feedback_fn):
1700
    """Removes the node from the cluster.
1701

1702
    """
1703
    node = self.node
1704
    logging.info("Stopping the node daemon and removing configs from node %s",
1705
                 node.name)
1706

    
1707
    self.context.RemoveNode(node.name)
1708

    
1709
    self.rpc.call_node_leave_cluster(node.name)
1710

    
1711
    # Promote nodes to master candidate as needed
1712
    _AdjustCandidatePool(self)
1713

    
1714

    
1715
class LUQueryNodes(NoHooksLU):
1716
  """Logical unit for querying nodes.
1717

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

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

    
1738
  def ExpandNames(self):
1739
    _CheckOutputFields(static=self._FIELDS_STATIC,
1740
                       dynamic=self._FIELDS_DYNAMIC,
1741
                       selected=self.op.output_fields)
1742

    
1743
    self.needed_locks = {}
1744
    self.share_locks[locking.LEVEL_NODE] = 1
1745

    
1746
    if self.op.names:
1747
      self.wanted = _GetWantedNodes(self, self.op.names)
1748
    else:
1749
      self.wanted = locking.ALL_SET
1750

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

    
1756

    
1757
  def CheckPrereq(self):
1758
    """Check prerequisites.
1759

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

    
1765
  def Exec(self, feedback_fn):
1766
    """Computes the list of nodes and their attributes.
1767

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

    
1781
    nodenames = utils.NiceSort(nodenames)
1782
    nodelist = [all_info[name] for name in nodenames]
1783

    
1784
    # begin data gathering
1785

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

    
1809
    node_to_primary = dict([(name, set()) for name in nodenames])
1810
    node_to_secondary = dict([(name, set()) for name in nodenames])
1811

    
1812
    inst_fields = frozenset(("pinst_cnt", "pinst_list",
1813
                             "sinst_cnt", "sinst_list"))
1814
    if inst_fields & frozenset(self.op.output_fields):
1815
      instancelist = self.cfg.GetInstanceList()
1816

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

    
1825
    master_node = self.cfg.GetMasterNode()
1826

    
1827
    # end data gathering
1828

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

    
1864
    return output
1865

    
1866

    
1867
class LUQueryNodeVolumes(NoHooksLU):
1868
  """Logical unit for getting volumes on node(s).
1869

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

    
1876
  def ExpandNames(self):
1877
    _CheckOutputFields(static=self._FIELDS_STATIC,
1878
                       dynamic=self._FIELDS_DYNAMIC,
1879
                       selected=self.op.output_fields)
1880

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

    
1889
  def CheckPrereq(self):
1890
    """Check prerequisites.
1891

1892
    This checks that the fields required are valid output fields.
1893

1894
    """
1895
    self.nodes = self.acquired_locks[locking.LEVEL_NODE]
1896

    
1897
  def Exec(self, feedback_fn):
1898
    """Computes the list of nodes and their attributes.
1899

1900
    """
1901
    nodenames = self.nodes
1902
    volumes = self.rpc.call_node_volumes(nodenames)
1903

    
1904
    ilist = [self.cfg.GetInstanceInfo(iname) for iname
1905
             in self.cfg.GetInstanceList()]
1906

    
1907
    lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
1908

    
1909
    output = []
1910
    for node in nodenames:
1911
      if node not in volumes or volumes[node].failed or not volumes[node].data:
1912
        continue
1913

    
1914
      node_vols = volumes[node].data[:]
1915
      node_vols.sort(key=lambda vol: vol['dev'])
1916

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

    
1943
        output.append(node_output)
1944

    
1945
    return output
1946

    
1947

    
1948
class LUAddNode(LogicalUnit):
1949
  """Logical unit for adding node to the cluster.
1950

1951
  """
1952
  HPATH = "node-add"
1953
  HTYPE = constants.HTYPE_NODE
1954
  _OP_REQP = ["node_name"]
1955

    
1956
  def BuildHooksEnv(self):
1957
    """Build hooks env.
1958

1959
    This will run on all nodes before, and on all nodes + the new node after.
1960

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

    
1972
  def CheckPrereq(self):
1973
    """Check prerequisites.
1974

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

1980
    Any errors are signalled by raising errors.OpPrereqError.
1981

1982
    """
1983
    node_name = self.op.node_name
1984
    cfg = self.cfg
1985

    
1986
    dns_data = utils.HostInfo(node_name)
1987

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

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

    
2004
    for existing_node_name in node_list:
2005
      existing_node = cfg.GetNodeInfo(existing_node_name)
2006

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

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

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

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

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

    
2045
    cp_size = self.cfg.GetClusterInfo().candidate_pool_size
2046
    mc_now, _ = self.cfg.GetMasterCandidateStats()
2047
    master_candidate = mc_now < cp_size
2048

    
2049
    self.new_node = objects.Node(name=node,
2050
                                 primary_ip=primary_ip,
2051
                                 secondary_ip=secondary_ip,
2052
                                 master_candidate=master_candidate,
2053
                                 offline=False)
2054

    
2055
  def Exec(self, feedback_fn):
2056
    """Adds the new node to the cluster.
2057

2058
    """
2059
    new_node = self.new_node
2060
    node = new_node.name
2061

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

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

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

    
2091
    result = self.rpc.call_node_add(node, keyarray[0], keyarray[1],
2092
                                    keyarray[2],
2093
                                    keyarray[3], keyarray[4], keyarray[5])
2094

    
2095
    if result.failed or not result.data:
2096
      raise errors.OpExecError("Cannot transfer ssh keys to the new node")
2097

    
2098
    # Add node to our /etc/hosts, and add key to known_hosts
2099
    utils.AddHostToEtcHosts(new_node.name)
2100

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

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

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

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

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

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

    
2151
    if self.op.readd:
2152
      self.context.ReaddNode(new_node)
2153
    else:
2154
      self.context.AddNode(new_node)
2155

    
2156

    
2157
class LUSetNodeParams(LogicalUnit):
2158
  """Modifies the parameters of a node.
2159

2160
  """
2161
  HPATH = "node-modify"
2162
  HTYPE = constants.HTYPE_NODE
2163
  _OP_REQP = ["node_name"]
2164
  REQ_BGL = False
2165

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

    
2179
  def ExpandNames(self):
2180
    self.needed_locks = {locking.LEVEL_NODE: self.op.node_name}
2181

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

2185
    This runs on the master node.
2186

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

    
2197
  def CheckPrereq(self):
2198
    """Check prerequisites.
2199

2200
    This only checks the instance list against the existing names.
2201

2202
    """
2203
    node = self.node = self.cfg.GetNodeInfo(self.op.node_name)
2204

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

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

    
2226
    return
2227

    
2228
  def Exec(self, feedback_fn):
2229
    """Modifies a node.
2230

2231
    """
2232
    node = self.node
2233

    
2234
    result = []
2235

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

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

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

    
2260
    return result
2261

    
2262

    
2263
class LUQueryClusterInfo(NoHooksLU):
2264
  """Query cluster configuration.
2265

2266
  """
2267
  _OP_REQP = []
2268
  REQ_BGL = False
2269

    
2270
  def ExpandNames(self):
2271
    self.needed_locks = {}
2272

    
2273
  def CheckPrereq(self):
2274
    """No prerequsites needed for this LU.
2275

2276
    """
2277
    pass
2278

    
2279
  def Exec(self, feedback_fn):
2280
    """Return cluster config.
2281

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

    
2300
    return result
2301

    
2302

    
2303
class LUQueryConfigValues(NoHooksLU):
2304
  """Return configuration values.
2305

2306
  """
2307
  _OP_REQP = []
2308
  REQ_BGL = False
2309
  _FIELDS_DYNAMIC = utils.FieldSet()
2310
  _FIELDS_STATIC = utils.FieldSet("cluster_name", "master_node", "drain_flag")
2311

    
2312
  def ExpandNames(self):
2313
    self.needed_locks = {}
2314

    
2315
    _CheckOutputFields(static=self._FIELDS_STATIC,
2316
                       dynamic=self._FIELDS_DYNAMIC,
2317
                       selected=self.op.output_fields)
2318

    
2319
  def CheckPrereq(self):
2320
    """No prerequisites.
2321

2322
    """
2323
    pass
2324

    
2325
  def Exec(self, feedback_fn):
2326
    """Dump a representation of the cluster config to the standard output.
2327

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

    
2342

    
2343
class LUActivateInstanceDisks(NoHooksLU):
2344
  """Bring up an instance's disks.
2345

2346
  """
2347
  _OP_REQP = ["instance_name"]
2348
  REQ_BGL = False
2349

    
2350
  def ExpandNames(self):
2351
    self._ExpandAndLockInstance()
2352
    self.needed_locks[locking.LEVEL_NODE] = []
2353
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2354

    
2355
  def DeclareLocks(self, level):
2356
    if level == locking.LEVEL_NODE:
2357
      self._LockInstancesNodes()
2358

    
2359
  def CheckPrereq(self):
2360
    """Check prerequisites.
2361

2362
    This checks that the instance is in the cluster.
2363

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

    
2370
  def Exec(self, feedback_fn):
2371
    """Activate the disks.
2372

2373
    """
2374
    disks_ok, disks_info = _AssembleInstanceDisks(self, self.instance)
2375
    if not disks_ok:
2376
      raise errors.OpExecError("Cannot activate block devices")
2377

    
2378
    return disks_info
2379

    
2380

    
2381
def _AssembleInstanceDisks(lu, instance, ignore_secondaries=False):
2382
  """Prepare the block devices for an instance.
2383

2384
  This sets up the block devices on all nodes.
2385

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

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

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

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

    
2422
  # FIXME: race condition on drbd migration to primary
2423

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

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

    
2444
  return disks_ok, device_info
2445

    
2446

    
2447
def _StartInstanceDisks(lu, instance, force):
2448
  """Start the disks of an instance.
2449

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

    
2461

    
2462
class LUDeactivateInstanceDisks(NoHooksLU):
2463
  """Shutdown an instance's disks.
2464

2465
  """
2466
  _OP_REQP = ["instance_name"]
2467
  REQ_BGL = False
2468

    
2469
  def ExpandNames(self):
2470
    self._ExpandAndLockInstance()
2471
    self.needed_locks[locking.LEVEL_NODE] = []
2472
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2473

    
2474
  def DeclareLocks(self, level):
2475
    if level == locking.LEVEL_NODE:
2476
      self._LockInstancesNodes()
2477

    
2478
  def CheckPrereq(self):
2479
    """Check prerequisites.
2480

2481
    This checks that the instance is in the cluster.
2482

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

    
2488
  def Exec(self, feedback_fn):
2489
    """Deactivate the disks
2490

2491
    """
2492
    instance = self.instance
2493
    _SafeShutdownInstanceDisks(self, instance)
2494

    
2495

    
2496
def _SafeShutdownInstanceDisks(lu, instance):
2497
  """Shutdown block devices of an instance.
2498

2499
  This function checks if an instance is running, before calling
2500
  _ShutdownInstanceDisks.
2501

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

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

    
2514
  _ShutdownInstanceDisks(lu, instance)
2515

    
2516

    
2517
def _ShutdownInstanceDisks(lu, instance, ignore_primary=False):
2518
  """Shutdown block devices of an instance.
2519

2520
  This does the shutdown on all nodes of the instance.
2521

2522
  If the ignore_primary is false, errors on the primary node are
2523
  ignored.
2524

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

    
2538

    
2539
def _CheckNodeFreeMemory(lu, node, reason, requested, hypervisor_name):
2540
  """Checks if a node has enough free memory.
2541

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

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

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

    
2572

    
2573
class LUStartupInstance(LogicalUnit):
2574
  """Starts an instance.
2575

2576
  """
2577
  HPATH = "instance-start"
2578
  HTYPE = constants.HTYPE_INSTANCE
2579
  _OP_REQP = ["instance_name", "force"]
2580
  REQ_BGL = False
2581

    
2582
  def ExpandNames(self):
2583
    self._ExpandAndLockInstance()
2584

    
2585
  def BuildHooksEnv(self):
2586
    """Build hooks env.
2587

2588
    This runs on master, primary and secondary nodes of the instance.
2589

2590
    """
2591
    env = {
2592
      "FORCE": self.op.force,
2593
      }
2594
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
2595
    nl = ([self.cfg.GetMasterNode(), self.instance.primary_node] +
2596
          list(self.instance.secondary_nodes))
2597
    return env, nl, nl
2598

    
2599
  def CheckPrereq(self):
2600
    """Check prerequisites.
2601

2602
    This checks that the instance is in the cluster.
2603

2604
    """
2605
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2606
    assert self.instance is not None, \
2607
      "Cannot retrieve locked instance %s" % self.op.instance_name
2608

    
2609
    _CheckNodeOnline(self, instance.primary_node)
2610

    
2611
    bep = self.cfg.GetClusterInfo().FillBE(instance)
2612
    # check bridges existance
2613
    _CheckInstanceBridgesExist(self, instance)
2614

    
2615
    _CheckNodeFreeMemory(self, instance.primary_node,
2616
                         "starting instance %s" % instance.name,
2617
                         bep[constants.BE_MEMORY], instance.hypervisor)
2618

    
2619
  def Exec(self, feedback_fn):
2620
    """Start the instance.
2621

2622
    """
2623
    instance = self.instance
2624
    force = self.op.force
2625
    extra_args = getattr(self.op, "extra_args", "")
2626

    
2627
    self.cfg.MarkInstanceUp(instance.name)
2628

    
2629
    node_current = instance.primary_node
2630

    
2631
    _StartInstanceDisks(self, instance, force)
2632

    
2633
    result = self.rpc.call_instance_start(node_current, instance, extra_args)
2634
    if result.failed or not result.data:
2635
      _ShutdownInstanceDisks(self, instance)
2636
      raise errors.OpExecError("Could not start instance")
2637

    
2638

    
2639
class LURebootInstance(LogicalUnit):
2640
  """Reboot an instance.
2641

2642
  """
2643
  HPATH = "instance-reboot"
2644
  HTYPE = constants.HTYPE_INSTANCE
2645
  _OP_REQP = ["instance_name", "ignore_secondaries", "reboot_type"]
2646
  REQ_BGL = False
2647

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

    
2658
  def BuildHooksEnv(self):
2659
    """Build hooks env.
2660

2661
    This runs on master, primary and secondary nodes of the instance.
2662

2663
    """
2664
    env = {
2665
      "IGNORE_SECONDARIES": self.op.ignore_secondaries,
2666
      }
2667
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
2668
    nl = ([self.cfg.GetMasterNode(), self.instance.primary_node] +
2669
          list(self.instance.secondary_nodes))
2670
    return env, nl, nl
2671

    
2672
  def CheckPrereq(self):
2673
    """Check prerequisites.
2674

2675
    This checks that the instance is in the cluster.
2676

2677
    """
2678
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2679
    assert self.instance is not None, \
2680
      "Cannot retrieve locked instance %s" % self.op.instance_name
2681

    
2682
    _CheckNodeOnline(self, instance.primary_node)
2683

    
2684
    # check bridges existance
2685
    _CheckInstanceBridgesExist(self, instance)
2686

    
2687
  def Exec(self, feedback_fn):
2688
    """Reboot the instance.
2689

2690
    """
2691
    instance = self.instance
2692
    ignore_secondaries = self.op.ignore_secondaries
2693
    reboot_type = self.op.reboot_type
2694
    extra_args = getattr(self.op, "extra_args", "")
2695

    
2696
    node_current = instance.primary_node
2697

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

    
2714
    self.cfg.MarkInstanceUp(instance.name)
2715

    
2716

    
2717
class LUShutdownInstance(LogicalUnit):
2718
  """Shutdown an instance.
2719

2720
  """
2721
  HPATH = "instance-stop"
2722
  HTYPE = constants.HTYPE_INSTANCE
2723
  _OP_REQP = ["instance_name"]
2724
  REQ_BGL = False
2725

    
2726
  def ExpandNames(self):
2727
    self._ExpandAndLockInstance()
2728

    
2729
  def BuildHooksEnv(self):
2730
    """Build hooks env.
2731

2732
    This runs on master, primary and secondary nodes of the instance.
2733

2734
    """
2735
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2736
    nl = ([self.cfg.GetMasterNode(), self.instance.primary_node] +
2737
          list(self.instance.secondary_nodes))
2738
    return env, nl, nl
2739

    
2740
  def CheckPrereq(self):
2741
    """Check prerequisites.
2742

2743
    This checks that the instance is in the cluster.
2744

2745
    """
2746
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2747
    assert self.instance is not None, \
2748
      "Cannot retrieve locked instance %s" % self.op.instance_name
2749
    _CheckNodeOnline(self, self.instance.primary_node)
2750

    
2751
  def Exec(self, feedback_fn):
2752
    """Shutdown the instance.
2753

2754
    """
2755
    instance = self.instance
2756
    node_current = instance.primary_node
2757
    self.cfg.MarkInstanceDown(instance.name)
2758
    result = self.rpc.call_instance_shutdown(node_current, instance)
2759
    if result.failed or not result.data:
2760
      self.proc.LogWarning("Could not shutdown instance")
2761

    
2762
    _ShutdownInstanceDisks(self, instance)
2763

    
2764

    
2765
class LUReinstallInstance(LogicalUnit):
2766
  """Reinstall an instance.
2767

2768
  """
2769
  HPATH = "instance-reinstall"
2770
  HTYPE = constants.HTYPE_INSTANCE
2771
  _OP_REQP = ["instance_name"]
2772
  REQ_BGL = False
2773

    
2774
  def ExpandNames(self):
2775
    self._ExpandAndLockInstance()
2776

    
2777
  def BuildHooksEnv(self):
2778
    """Build hooks env.
2779

2780
    This runs on master, primary and secondary nodes of the instance.
2781

2782
    """
2783
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2784
    nl = ([self.cfg.GetMasterNode(), self.instance.primary_node] +
2785
          list(self.instance.secondary_nodes))
2786
    return env, nl, nl
2787

    
2788
  def CheckPrereq(self):
2789
    """Check prerequisites.
2790

2791
    This checks that the instance is in the cluster and is not running.
2792

2793
    """
2794
    instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2795
    assert instance is not None, \
2796
      "Cannot retrieve locked instance %s" % self.op.instance_name
2797
    _CheckNodeOnline(self, instance.primary_node)
2798

    
2799
    if instance.disk_template == constants.DT_DISKLESS:
2800
      raise errors.OpPrereqError("Instance '%s' has no disks" %
2801
                                 self.op.instance_name)
2802
    if instance.status != "down":
2803
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2804
                                 self.op.instance_name)
2805
    remote_info = self.rpc.call_instance_info(instance.primary_node,
2806
                                              instance.name,
2807
                                              instance.hypervisor)
2808
    if remote_info.failed or remote_info.data:
2809
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
2810
                                 (self.op.instance_name,
2811
                                  instance.primary_node))
2812

    
2813
    self.op.os_type = getattr(self.op, "os_type", None)
2814
    if self.op.os_type is not None:
2815
      # OS verification
2816
      pnode = self.cfg.GetNodeInfo(
2817
        self.cfg.ExpandNodeName(instance.primary_node))
2818
      if pnode is None:
2819
        raise errors.OpPrereqError("Primary node '%s' is unknown" %
2820
                                   self.op.pnode)
2821
      result = self.rpc.call_os_get(pnode.name, self.op.os_type)
2822
      result.Raise()
2823
      if not isinstance(result.data, objects.OS):
2824
        raise errors.OpPrereqError("OS '%s' not in supported OS list for"
2825
                                   " primary node"  % self.op.os_type)
2826

    
2827
    self.instance = instance
2828

    
2829
  def Exec(self, feedback_fn):
2830
    """Reinstall the instance.
2831

2832
    """
2833
    inst = self.instance
2834

    
2835
    if self.op.os_type is not None:
2836
      feedback_fn("Changing OS to '%s'..." % self.op.os_type)
2837
      inst.os = self.op.os_type
2838
      self.cfg.Update(inst)
2839

    
2840
    _StartInstanceDisks(self, inst, None)
2841
    try:
2842
      feedback_fn("Running the instance OS create scripts...")
2843
      result = self.rpc.call_instance_os_add(inst.primary_node, inst)
2844
      result.Raise()
2845
      if not result.data:
2846
        raise errors.OpExecError("Could not install OS for instance %s"
2847
                                 " on node %s" %
2848
                                 (inst.name, inst.primary_node))
2849
    finally:
2850
      _ShutdownInstanceDisks(self, inst)
2851

    
2852

    
2853
class LURenameInstance(LogicalUnit):
2854
  """Rename an instance.
2855

2856
  """
2857
  HPATH = "instance-rename"
2858
  HTYPE = constants.HTYPE_INSTANCE
2859
  _OP_REQP = ["instance_name", "new_name"]
2860

    
2861
  def BuildHooksEnv(self):
2862
    """Build hooks env.
2863

2864
    This runs on master, primary and secondary nodes of the instance.
2865

2866
    """
2867
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2868
    env["INSTANCE_NEW_NAME"] = self.op.new_name
2869
    nl = ([self.cfg.GetMasterNode(), self.instance.primary_node] +
2870
          list(self.instance.secondary_nodes))
2871
    return env, nl, nl
2872

    
2873
  def CheckPrereq(self):
2874
    """Check prerequisites.
2875

2876
    This checks that the instance is in the cluster and is not running.
2877

2878
    """
2879
    instance = self.cfg.GetInstanceInfo(
2880
      self.cfg.ExpandInstanceName(self.op.instance_name))
2881
    if instance is None:
2882
      raise errors.OpPrereqError("Instance '%s' not known" %
2883
                                 self.op.instance_name)
2884
    _CheckNodeOnline(self, instance.primary_node)
2885

    
2886
    if instance.status != "down":
2887
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2888
                                 self.op.instance_name)
2889
    remote_info = self.rpc.call_instance_info(instance.primary_node,
2890
                                              instance.name,
2891
                                              instance.hypervisor)
2892
    remote_info.Raise()
2893
    if remote_info.data:
2894
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
2895
                                 (self.op.instance_name,
2896
                                  instance.primary_node))
2897
    self.instance = instance
2898

    
2899
    # new name verification
2900
    name_info = utils.HostInfo(self.op.new_name)
2901

    
2902
    self.op.new_name = new_name = name_info.name
2903
    instance_list = self.cfg.GetInstanceList()
2904
    if new_name in instance_list:
2905
      raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
2906
                                 new_name)
2907

    
2908
    if not getattr(self.op, "ignore_ip", False):
2909
      if utils.TcpPing(name_info.ip, constants.DEFAULT_NODED_PORT):
2910
        raise errors.OpPrereqError("IP %s of instance %s already in use" %
2911
                                   (name_info.ip, new_name))
2912

    
2913

    
2914
  def Exec(self, feedback_fn):
2915
    """Reinstall the instance.
2916

2917
    """
2918
    inst = self.instance
2919
    old_name = inst.name
2920

    
2921
    if inst.disk_template == constants.DT_FILE:
2922
      old_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
2923

    
2924
    self.cfg.RenameInstance(inst.name, self.op.new_name)
2925
    # Change the instance lock. This is definitely safe while we hold the BGL
2926
    self.context.glm.remove(locking.LEVEL_INSTANCE, old_name)
2927
    self.context.glm.add(locking.LEVEL_INSTANCE, self.op.new_name)
2928

    
2929
    # re-read the instance from the configuration after rename
2930
    inst = self.cfg.GetInstanceInfo(self.op.new_name)
2931

    
2932
    if inst.disk_template == constants.DT_FILE:
2933
      new_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
2934
      result = self.rpc.call_file_storage_dir_rename(inst.primary_node,
2935
                                                     old_file_storage_dir,
2936
                                                     new_file_storage_dir)
2937
      result.Raise()
2938
      if not result.data:
2939
        raise errors.OpExecError("Could not connect to node '%s' to rename"
2940
                                 " directory '%s' to '%s' (but the instance"
2941
                                 " has been renamed in Ganeti)" % (
2942
                                 inst.primary_node, old_file_storage_dir,
2943
                                 new_file_storage_dir))
2944

    
2945
      if not result.data[0]:
2946
        raise errors.OpExecError("Could not rename directory '%s' to '%s'"
2947
                                 " (but the instance has been renamed in"
2948
                                 " Ganeti)" % (old_file_storage_dir,
2949
                                               new_file_storage_dir))
2950

    
2951
    _StartInstanceDisks(self, inst, None)
2952
    try:
2953
      result = self.rpc.call_instance_run_rename(inst.primary_node, inst,
2954
                                                 old_name)
2955
      if result.failed or not result.data:
2956
        msg = ("Could not run OS rename script for instance %s on node %s"
2957
               " (but the instance has been renamed in Ganeti)" %
2958
               (inst.name, inst.primary_node))
2959
        self.proc.LogWarning(msg)
2960
    finally:
2961
      _ShutdownInstanceDisks(self, inst)
2962

    
2963

    
2964
class LURemoveInstance(LogicalUnit):
2965
  """Remove an instance.
2966

2967
  """
2968
  HPATH = "instance-remove"
2969
  HTYPE = constants.HTYPE_INSTANCE
2970
  _OP_REQP = ["instance_name", "ignore_failures"]
2971
  REQ_BGL = False
2972

    
2973
  def ExpandNames(self):
2974
    self._ExpandAndLockInstance()
2975
    self.needed_locks[locking.LEVEL_NODE] = []
2976
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2977

    
2978
  def DeclareLocks(self, level):
2979
    if level == locking.LEVEL_NODE:
2980
      self._LockInstancesNodes()
2981

    
2982
  def BuildHooksEnv(self):
2983
    """Build hooks env.
2984

2985
    This runs on master, primary and secondary nodes of the instance.
2986

2987
    """
2988
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2989
    nl = [self.cfg.GetMasterNode()]
2990
    return env, nl, nl
2991

    
2992
  def CheckPrereq(self):
2993
    """Check prerequisites.
2994

2995
    This checks that the instance is in the cluster.
2996

2997
    """
2998
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2999
    assert self.instance is not None, \
3000
      "Cannot retrieve locked instance %s" % self.op.instance_name
3001

    
3002
  def Exec(self, feedback_fn):
3003
    """Remove the instance.
3004

3005
    """
3006
    instance = self.instance
3007
    logging.info("Shutting down instance %s on node %s",
3008
                 instance.name, instance.primary_node)
3009

    
3010
    result = self.rpc.call_instance_shutdown(instance.primary_node, instance)
3011
    if result.failed or not result.data:
3012
      if self.op.ignore_failures:
3013
        feedback_fn("Warning: can't shutdown instance")
3014
      else:
3015
        raise errors.OpExecError("Could not shutdown instance %s on node %s" %
3016
                                 (instance.name, instance.primary_node))
3017

    
3018
    logging.info("Removing block devices for instance %s", instance.name)
3019

    
3020
    if not _RemoveDisks(self, instance):
3021
      if self.op.ignore_failures:
3022
        feedback_fn("Warning: can't remove instance's disks")
3023
      else:
3024
        raise errors.OpExecError("Can't remove instance's disks")
3025

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

    
3028
    self.cfg.RemoveInstance(instance.name)
3029
    self.remove_locks[locking.LEVEL_INSTANCE] = instance.name
3030

    
3031

    
3032
class LUQueryInstances(NoHooksLU):
3033
  """Logical unit for querying instances.
3034

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

    
3055

    
3056
  def ExpandNames(self):
3057
    _CheckOutputFields(static=self._FIELDS_STATIC,
3058
                       dynamic=self._FIELDS_DYNAMIC,
3059
                       selected=self.op.output_fields)
3060

    
3061
    self.needed_locks = {}
3062
    self.share_locks[locking.LEVEL_INSTANCE] = 1
3063
    self.share_locks[locking.LEVEL_NODE] = 1
3064

    
3065
    if self.op.names:
3066
      self.wanted = _GetWantedInstances(self, self.op.names)
3067
    else:
3068
      self.wanted = locking.ALL_SET
3069

    
3070
    self.do_locking = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
3071
    if self.do_locking:
3072
      self.needed_locks[locking.LEVEL_INSTANCE] = self.wanted
3073
      self.needed_locks[locking.LEVEL_NODE] = []
3074
      self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3075

    
3076
  def DeclareLocks(self, level):
3077
    if level == locking.LEVEL_NODE and self.do_locking:
3078
      self._LockInstancesNodes()
3079

    
3080
  def CheckPrereq(self):
3081
    """Check prerequisites.
3082

3083
    """
3084
    pass
3085

    
3086
  def Exec(self, feedback_fn):
3087
    """Computes the list of nodes and their attributes.
3088

3089
    """
3090
    all_info = self.cfg.GetAllInstancesInfo()
3091
    if self.do_locking:
3092
      instance_names = self.acquired_locks[locking.LEVEL_INSTANCE]
3093
    elif self.wanted != locking.ALL_SET:
3094
      instance_names = self.wanted
3095
      missing = set(instance_names).difference(all_info.keys())
3096
      if missing:
3097
        raise errors.OpExecError(
3098
          "Some instances were removed before retrieving their data: %s"
3099
          % missing)
3100
    else:
3101
      instance_names = all_info.keys()
3102

    
3103
    instance_names = utils.NiceSort(instance_names)
3104
    instance_list = [all_info[iname] for iname in instance_names]
3105

    
3106
    # begin data gathering
3107

    
3108
    nodes = frozenset([inst.primary_node for inst in instance_list])
3109
    hv_list = list(set([inst.hypervisor for inst in instance_list]))
3110

    
3111
    bad_nodes = []
3112
    off_nodes = []
3113
    if self.do_locking:
3114
      live_data = {}
3115
      node_data = self.rpc.call_all_instances_info(nodes, hv_list)
3116
      for name in nodes:
3117
        result = node_data[name]
3118
        if result.offline:
3119
          # offline nodes will be in both lists
3120
          off_nodes.append(name)
3121
        if result.failed:
3122
          bad_nodes.append(name)
3123
        else:
3124
          if result.data:
3125
            live_data.update(result.data)
3126
            # else no instance is alive
3127
    else:
3128
      live_data = dict([(name, {}) for name in instance_names])
3129

    
3130
    # end data gathering
3131

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

    
3257
    return output
3258

    
3259

    
3260
class LUFailoverInstance(LogicalUnit):
3261
  """Failover an instance.
3262

3263
  """
3264
  HPATH = "instance-failover"
3265
  HTYPE = constants.HTYPE_INSTANCE
3266
  _OP_REQP = ["instance_name", "ignore_consistency"]
3267
  REQ_BGL = False
3268

    
3269
  def ExpandNames(self):
3270
    self._ExpandAndLockInstance()
3271
    self.needed_locks[locking.LEVEL_NODE] = []
3272
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3273

    
3274
  def DeclareLocks(self, level):
3275
    if level == locking.LEVEL_NODE:
3276
      self._LockInstancesNodes()
3277

    
3278
  def BuildHooksEnv(self):
3279
    """Build hooks env.
3280

3281
    This runs on master, primary and secondary nodes of the instance.
3282

3283
    """
3284
    env = {
3285
      "IGNORE_CONSISTENCY": self.op.ignore_consistency,
3286
      }
3287
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3288
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3289
    return env, nl, nl
3290

    
3291
  def CheckPrereq(self):
3292
    """Check prerequisites.
3293

3294
    This checks that the instance is in the cluster.
3295

3296
    """
3297
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3298
    assert self.instance is not None, \
3299
      "Cannot retrieve locked instance %s" % self.op.instance_name
3300

    
3301
    bep = self.cfg.GetClusterInfo().FillBE(instance)
3302
    if instance.disk_template not in constants.DTS_NET_MIRROR:
3303
      raise errors.OpPrereqError("Instance's disk layout is not"
3304
                                 " network mirrored, cannot failover.")
3305

    
3306
    secondary_nodes = instance.secondary_nodes
3307
    if not secondary_nodes:
3308
      raise errors.ProgrammerError("no secondary node but using "
3309
                                   "a mirrored disk template")
3310

    
3311
    target_node = secondary_nodes[0]
3312
    _CheckNodeOnline(self, target_node)
3313
    # check memory requirements on the secondary node
3314
    _CheckNodeFreeMemory(self, target_node, "failing over instance %s" %
3315
                         instance.name, bep[constants.BE_MEMORY],
3316
                         instance.hypervisor)
3317

    
3318
    # check bridge existance
3319
    brlist = [nic.bridge for nic in instance.nics]
3320
    result = self.rpc.call_bridges_exist(target_node, brlist)
3321
    result.Raise()
3322
    if not result.data:
3323
      raise errors.OpPrereqError("One or more target bridges %s does not"
3324
                                 " exist on destination node '%s'" %
3325
                                 (brlist, target_node))
3326

    
3327
  def Exec(self, feedback_fn):
3328
    """Failover an instance.
3329

3330
    The failover is done by shutting it down on its present node and
3331
    starting it on the secondary.
3332

3333
    """
3334
    instance = self.instance
3335

    
3336
    source_node = instance.primary_node
3337
    target_node = instance.secondary_nodes[0]
3338

    
3339
    feedback_fn("* checking disk consistency between source and target")
3340
    for dev in instance.disks:
3341
      # for drbd, these are drbd over lvm
3342
      if not _CheckDiskConsistency(self, dev, target_node, False):
3343
        if instance.status == "up" and not self.op.ignore_consistency:
3344
          raise errors.OpExecError("Disk %s is degraded on target node,"
3345
                                   " aborting failover." % dev.iv_name)
3346

    
3347
    feedback_fn("* shutting down instance on source node")
3348
    logging.info("Shutting down instance %s on node %s",
3349
                 instance.name, source_node)
3350

    
3351
    result = self.rpc.call_instance_shutdown(source_node, instance)
3352
    if result.failed or not result.data:
3353
      if self.op.ignore_consistency:
3354
        self.proc.LogWarning("Could not shutdown instance %s on node %s."
3355
                             " Proceeding"
3356
                             " anyway. Please make sure node %s is down",
3357
                             instance.name, source_node, source_node)
3358
      else:
3359
        raise errors.OpExecError("Could not shutdown instance %s on node %s" %
3360
                                 (instance.name, source_node))
3361

    
3362
    feedback_fn("* deactivating the instance's disks on source node")
3363
    if not _ShutdownInstanceDisks(self, instance, ignore_primary=True):
3364
      raise errors.OpExecError("Can't shut down the instance's disks.")
3365

    
3366
    instance.primary_node = target_node
3367
    # distribute new instance config to the other nodes
3368
    self.cfg.Update(instance)
3369

    
3370
    # Only start the instance if it's marked as up
3371
    if instance.status == "up":
3372
      feedback_fn("* activating the instance's disks on target node")
3373
      logging.info("Starting instance %s on node %s",
3374
                   instance.name, target_node)
3375

    
3376
      disks_ok, dummy = _AssembleInstanceDisks(self, instance,
3377
                                               ignore_secondaries=True)
3378
      if not disks_ok:
3379
        _ShutdownInstanceDisks(self, instance)
3380
        raise errors.OpExecError("Can't activate the instance's disks")
3381

    
3382
      feedback_fn("* starting the instance on the target node")
3383
      result = self.rpc.call_instance_start(target_node, instance, None)
3384
      if result.failed or not result.data:
3385
        _ShutdownInstanceDisks(self, instance)
3386
        raise errors.OpExecError("Could not start instance %s on node %s." %
3387
                                 (instance.name, target_node))
3388

    
3389

    
3390
class LUMigrateInstance(LogicalUnit):
3391
  """Migrate an instance.
3392

3393
  This is migration without shutting down, compared to the failover,
3394
  which is done with shutdown.
3395

3396
  """
3397
  HPATH = "instance-migrate"
3398
  HTYPE = constants.HTYPE_INSTANCE
3399
  _OP_REQP = ["instance_name", "live", "cleanup"]
3400

    
3401
  REQ_BGL = False
3402

    
3403
  def ExpandNames(self):
3404
    self._ExpandAndLockInstance()
3405
    self.needed_locks[locking.LEVEL_NODE] = []
3406
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3407

    
3408
  def DeclareLocks(self, level):
3409
    if level == locking.LEVEL_NODE:
3410
      self._LockInstancesNodes()
3411

    
3412
  def BuildHooksEnv(self):
3413
    """Build hooks env.
3414

3415
    This runs on master, primary and secondary nodes of the instance.
3416

3417
    """
3418
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3419
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3420
    return env, nl, nl
3421

    
3422
  def CheckPrereq(self):
3423
    """Check prerequisites.
3424

3425
    This checks that the instance is in the cluster.
3426

3427
    """
3428
    instance = self.cfg.GetInstanceInfo(
3429
      self.cfg.ExpandInstanceName(self.op.instance_name))
3430
    if instance is None:
3431
      raise errors.OpPrereqError("Instance '%s' not known" %
3432
                                 self.op.instance_name)
3433

    
3434
    if instance.disk_template != constants.DT_DRBD8:
3435
      raise errors.OpPrereqError("Instance's disk layout is not"
3436
                                 " drbd8, cannot migrate.")
3437

    
3438
    secondary_nodes = instance.secondary_nodes
3439
    if not secondary_nodes:
3440
      raise errors.ProgrammerError("no secondary node but using "
3441
                                   "drbd8 disk template")
3442

    
3443
    i_be = self.cfg.GetClusterInfo().FillBE(instance)
3444

    
3445
    target_node = secondary_nodes[0]
3446
    # check memory requirements on the secondary node
3447
    _CheckNodeFreeMemory(self, target_node, "migrating instance %s" %
3448
                         instance.name, i_be[constants.BE_MEMORY],
3449
                         instance.hypervisor)
3450

    
3451
    # check bridge existance
3452
    brlist = [nic.bridge for nic in instance.nics]
3453
    result = self.rpc.call_bridges_exist(target_node, brlist)
3454
    if result.failed or not result.data:
3455
      raise errors.OpPrereqError("One or more target bridges %s does not"
3456
                                 " exist on destination node '%s'" %
3457
                                 (brlist, target_node))
3458

    
3459
    if not self.op.cleanup:
3460
      result = self.rpc.call_instance_migratable(instance.primary_node,
3461
                                                 instance)
3462
      msg = result.RemoteFailMsg()
3463
      if msg:
3464
        raise errors.OpPrereqError("Can't migrate: %s - please use failover" %
3465
                                   msg)
3466

    
3467
    self.instance = instance
3468

    
3469
  def _WaitUntilSync(self):
3470
    """Poll with custom rpc for disk sync.
3471

3472
    This uses our own step-based rpc call.
3473

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

    
3497
  def _EnsureSecondary(self, node):
3498
    """Demote a node to secondary.
3499

3500
    """
3501
    self.feedback_fn("* switching node %s to secondary mode" % node)
3502

    
3503
    for dev in self.instance.disks:
3504
      self.cfg.SetDiskID(dev, node)
3505

    
3506
    result = self.rpc.call_blockdev_close(node, self.instance.name,
3507
                                          self.instance.disks)
3508
    msg = result.RemoteFailMsg()
3509
    if msg:
3510
      raise errors.OpExecError("Cannot change disk to secondary on node %s,"
3511
                               " error %s" % (node, msg))
3512

    
3513
  def _GoStandalone(self):
3514
    """Disconnect from the network.
3515

3516
    """
3517
    self.feedback_fn("* changing into standalone mode")
3518
    result = self.rpc.call_drbd_disconnect_net(self.all_nodes, self.nodes_ip,
3519
                                               self.instance.disks)
3520
    for node, nres in result.items():
3521
      msg = nres.RemoteFailMsg()
3522
      if msg:
3523
        raise errors.OpExecError("Cannot disconnect disks node %s,"
3524
                                 " error %s" % (node, msg))
3525

    
3526
  def _GoReconnect(self, multimaster):
3527
    """Reconnect to the network.
3528

3529
    """
3530
    if multimaster:
3531
      msg = "dual-master"
3532
    else:
3533
      msg = "single-master"
3534
    self.feedback_fn("* changing disks into %s mode" % msg)
3535
    result = self.rpc.call_drbd_attach_net(self.all_nodes, self.nodes_ip,
3536
                                           self.instance.disks,
3537
                                           self.instance.name, multimaster)
3538
    for node, nres in result.items():
3539
      msg = nres.RemoteFailMsg()
3540
      if msg:
3541
        raise errors.OpExecError("Cannot change disks config on node %s,"
3542
                                 " error: %s" % (node, msg))
3543

    
3544
  def _ExecCleanup(self):
3545
    """Try to cleanup after a failed migration.
3546

3547
    The cleanup is done by:
3548
      - check that the instance is running only on one node
3549
        (and update the config if needed)
3550
      - change disks on its secondary node to secondary
3551
      - wait until disks are fully synchronized
3552
      - disconnect from the network
3553
      - change disks into single-master mode
3554
      - wait again until disks are fully synchronized
3555

3556
    """
3557
    instance = self.instance
3558
    target_node = self.target_node
3559
    source_node = self.source_node
3560

    
3561
    # check running on only one node
3562
    self.feedback_fn("* checking where the instance actually runs"
3563
                     " (if this hangs, the hypervisor might be in"
3564
                     " a bad state)")
3565
    ins_l = self.rpc.call_instance_list(self.all_nodes, [instance.hypervisor])
3566
    for node, result in ins_l.items():
3567
      result.Raise()
3568
      if not isinstance(result.data, list):
3569
        raise errors.OpExecError("Can't contact node '%s'" % node)
3570

    
3571
    runningon_source = instance.name in ins_l[source_node].data
3572
    runningon_target = instance.name in ins_l[target_node].data
3573

    
3574
    if runningon_source and runningon_target:
3575
      raise errors.OpExecError("Instance seems to be running on two nodes,"
3576
                               " or the hypervisor is confused. You will have"
3577
                               " to ensure manually that it runs only on one"
3578
                               " and restart this operation.")
3579

    
3580
    if not (runningon_source or runningon_target):
3581
      raise errors.OpExecError("Instance does not seem to be running at all."
3582
                               " In this case, it's safer to repair by"
3583
                               " running 'gnt-instance stop' to ensure disk"
3584
                               " shutdown, and then restarting it.")
3585

    
3586
    if runningon_target:
3587
      # the migration has actually succeeded, we need to update the config
3588
      self.feedback_fn("* instance running on secondary node (%s),"
3589
                       " updating config" % target_node)
3590
      instance.primary_node = target_node
3591
      self.cfg.Update(instance)
3592
      demoted_node = source_node
3593
    else:
3594
      self.feedback_fn("* instance confirmed to be running on its"
3595
                       " primary node (%s)" % source_node)
3596
      demoted_node = target_node
3597

    
3598
    self._EnsureSecondary(demoted_node)
3599
    try:
3600
      self._WaitUntilSync()
3601
    except errors.OpExecError:
3602
      # we ignore here errors, since if the device is standalone, it
3603
      # won't be able to sync
3604
      pass
3605
    self._GoStandalone()
3606
    self._GoReconnect(False)
3607
    self._WaitUntilSync()
3608

    
3609
    self.feedback_fn("* done")
3610

    
3611
  def _ExecMigration(self):
3612
    """Migrate an instance.
3613

3614
    The migrate is done by:
3615
      - change the disks into dual-master mode
3616
      - wait until disks are fully synchronized again
3617
      - migrate the instance
3618
      - change disks on the new secondary node (the old primary) to secondary
3619
      - wait until disks are fully synchronized
3620
      - change disks into single-master mode
3621

3622
    """
3623
    instance = self.instance
3624
    target_node = self.target_node
3625
    source_node = self.source_node
3626

    
3627
    self.feedback_fn("* checking disk consistency between source and target")
3628
    for dev in instance.disks:
3629
      if not _CheckDiskConsistency(self, dev, target_node, False):
3630
        raise errors.OpExecError("Disk %s is degraded or not fully"
3631
                                 " synchronized on target node,"
3632
                                 " aborting migrate." % dev.iv_name)
3633

    
3634
    self._EnsureSecondary(target_node)
3635
    self._GoStandalone()
3636
    self._GoReconnect(True)
3637
    self._WaitUntilSync()
3638

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

    
3659
      raise errors.OpExecError("Could not migrate instance %s: %s" %
3660
                               (instance.name, msg))
3661
    time.sleep(10)
3662

    
3663
    instance.primary_node = target_node
3664
    # distribute new instance config to the other nodes
3665
    self.cfg.Update(instance)
3666

    
3667
    self._EnsureSecondary(source_node)
3668
    self._WaitUntilSync()
3669
    self._GoStandalone()
3670
    self._GoReconnect(False)
3671
    self._WaitUntilSync()
3672

    
3673
    self.feedback_fn("* done")
3674

    
3675
  def Exec(self, feedback_fn):
3676
    """Perform the migration.
3677

3678
    """
3679
    self.feedback_fn = feedback_fn
3680

    
3681
    self.source_node = self.instance.primary_node
3682
    self.target_node = self.instance.secondary_nodes[0]
3683
    self.all_nodes = [self.source_node, self.target_node]
3684
    self.nodes_ip = {
3685
      self.source_node: self.cfg.GetNodeInfo(self.source_node).secondary_ip,
3686
      self.target_node: self.cfg.GetNodeInfo(self.target_node).secondary_ip,
3687
      }
3688
    if self.op.cleanup:
3689
      return self._ExecCleanup()
3690
    else:
3691
      return self._ExecMigration()
3692

    
3693

    
3694
def _CreateBlockDevOnPrimary(lu, node, instance, device, info):
3695
  """Create a tree of block devices on the primary node.
3696

3697
  This always creates all devices.
3698

3699
  """
3700
  if device.children:
3701
    for child in device.children:
3702
      if not _CreateBlockDevOnPrimary(lu, node, instance, child, info):
3703
        return False
3704

    
3705
  lu.cfg.SetDiskID(device, node)
3706
  new_id = lu.rpc.call_blockdev_create(node, device, device.size,
3707
                                       instance.name, True, info)
3708
  if new_id.failed or not new_id.data:
3709
    return False
3710
  if device.physical_id is None:
3711
    device.physical_id = new_id
3712
  return True
3713

    
3714

    
3715
def _CreateBlockDevOnSecondary(lu, node, instance, device, force, info):
3716
  """Create a tree of block devices on a secondary node.
3717

3718
  If this device type has to be created on secondaries, create it and
3719
  all its children.
3720

3721
  If not, just recurse to children keeping the same 'force' value.
3722

3723
  """
3724
  if device.CreateOnSecondary():
3725
    force = True
3726
  if device.children:
3727
    for child in device.children:
3728
      if not _CreateBlockDevOnSecondary(lu, node, instance,
3729
                                        child, force, info):
3730
        return False
3731

    
3732
  if not force:
3733
    return True
3734
  lu.cfg.SetDiskID(device, node)
3735
  new_id = lu.rpc.call_blockdev_create(node, device, device.size,
3736
                                       instance.name, False, info)
3737
  if new_id.failed or not new_id.data:
3738
    return False
3739
  if device.physical_id is None:
3740
    device.physical_id = new_id
3741
  return True
3742

    
3743

    
3744
def _GenerateUniqueNames(lu, exts):
3745
  """Generate a suitable LV name.
3746

3747
  This will generate a logical volume name for the given instance.
3748

3749
  """
3750
  results = []
3751
  for val in exts:
3752
    new_id = lu.cfg.GenerateUniqueID()
3753
    results.append("%s%s" % (new_id, val))
3754
  return results
3755

    
3756

    
3757
def _GenerateDRBD8Branch(lu, primary, secondary, size, names, iv_name,
3758
                         p_minor, s_minor):
3759
  """Generate a drbd8 device complete with its children.
3760

3761
  """
3762
  port = lu.cfg.AllocatePort()
3763
  vgname = lu.cfg.GetVGName()
3764
  shared_secret = lu.cfg.GenerateDRBDSecret()
3765
  dev_data = objects.Disk(dev_type=constants.LD_LV, size=size,
3766
                          logical_id=(vgname, names[0]))
3767
  dev_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
3768
                          logical_id=(vgname, names[1]))
3769
  drbd_dev = objects.Disk(dev_type=constants.LD_DRBD8, size=size,
3770
                          logical_id=(primary, secondary, port,
3771
                                      p_minor, s_minor,
3772
                                      shared_secret),
3773
                          children=[dev_data, dev_meta],
3774
                          iv_name=iv_name)
3775
  return drbd_dev
3776

    
3777

    
3778
def _GenerateDiskTemplate(lu, template_name,
3779
                          instance_name, primary_node,
3780
                          secondary_nodes, disk_info,
3781
                          file_storage_dir, file_driver,
3782
                          base_index):
3783
  """Generate the entire disk layout for a given template type.
3784

3785
  """
3786
  #TODO: compute space requirements
3787

    
3788
  vgname = lu.cfg.GetVGName()
3789
  disk_count = len(disk_info)
3790
  disks = []
3791
  if template_name == constants.DT_DISKLESS:
3792
    pass
3793
  elif template_name == constants.DT_PLAIN:
3794
    if len(secondary_nodes) != 0:
3795
      raise errors.ProgrammerError("Wrong template configuration")
3796

    
3797
    names = _GenerateUniqueNames(lu, [".disk%d" % i
3798
                                      for i in range(disk_count)])
3799
    for idx, disk in enumerate(disk_info):
3800
      disk_index = idx + base_index
3801
      disk_dev = objects.Disk(dev_type=constants.LD_LV, size=disk["size"],
3802
                              logical_id=(vgname, names[idx]),
3803
                              iv_name="disk/%d" % disk_index)
3804
      disks.append(disk_dev)
3805
  elif template_name == constants.DT_DRBD8:
3806
    if len(secondary_nodes) != 1:
3807
      raise errors.ProgrammerError("Wrong template configuration")
3808
    remote_node = secondary_nodes[0]
3809
    minors = lu.cfg.AllocateDRBDMinor(
3810
      [primary_node, remote_node] * len(disk_info), instance_name)
3811

    
3812
    names = _GenerateUniqueNames(lu,
3813
                                 [".disk%d_%s" % (i, s)
3814
                                  for i in range(disk_count)
3815
                                  for s in ("data", "meta")
3816
                                  ])
3817
    for idx, disk in enumerate(disk_info):
3818
      disk_index = idx + base_index
3819
      disk_dev = _GenerateDRBD8Branch(lu, primary_node, remote_node,
3820
                                      disk["size"], names[idx*2:idx*2+2],
3821
                                      "disk/%d" % disk_index,
3822
                                      minors[idx*2], minors[idx*2+1])
3823
      disks.append(disk_dev)
3824
  elif template_name == constants.DT_FILE:
3825
    if len(secondary_nodes) != 0:
3826
      raise errors.ProgrammerError("Wrong template configuration")
3827

    
3828
    for idx, disk in enumerate(disk_info):
3829
      disk_index = idx + base_index
3830
      disk_dev = objects.Disk(dev_type=constants.LD_FILE, size=disk["size"],
3831
                              iv_name="disk/%d" % disk_index,
3832
                              logical_id=(file_driver,
3833
                                          "%s/disk%d" % (file_storage_dir,
3834
                                                         idx)))
3835
      disks.append(disk_dev)
3836
  else:
3837
    raise errors.ProgrammerError("Invalid disk template '%s'" % template_name)
3838
  return disks
3839

    
3840

    
3841
def _GetInstanceInfoText(instance):
3842
  """Compute that text that should be added to the disk's metadata.
3843

3844
  """
3845
  return "originstname+%s" % instance.name
3846

    
3847

    
3848
def _CreateDisks(lu, instance):
3849
  """Create all disks for an instance.
3850

3851
  This abstracts away some work from AddInstance.
3852

3853
  @type lu: L{LogicalUnit}
3854
  @param lu: the logical unit on whose behalf we execute
3855
  @type instance: L{objects.Instance}
3856
  @param instance: the instance whose disks we should create
3857
  @rtype: boolean
3858
  @return: the success of the creation
3859

3860
  """
3861
  info = _GetInstanceInfoText(instance)
3862

    
3863
  if instance.disk_template == constants.DT_FILE:
3864
    file_storage_dir = os.path.dirname(instance.disks[0].logical_id[1])
3865
    result = lu.rpc.call_file_storage_dir_create(instance.primary_node,
3866
                                                 file_storage_dir)
3867

    
3868
    if result.failed or not result.data:
3869
      logging.error("Could not connect to node '%s'", instance.primary_node)
3870
      return False
3871

    
3872
    if not result.data[0]:
3873
      logging.error("Failed to create directory '%s'", file_storage_dir)
3874
      return False
3875

    
3876
  # Note: this needs to be kept in sync with adding of disks in
3877
  # LUSetInstanceParams
3878
  for device in instance.disks:
3879
    logging.info("Creating volume %s for instance %s",
3880
                 device.iv_name, instance.name)
3881
    #HARDCODE
3882
    for secondary_node in instance.secondary_nodes:
3883
      if not _CreateBlockDevOnSecondary(lu, secondary_node, instance,
3884
                                        device, False, info):
3885
        logging.error("Failed to create volume %s (%s) on secondary node %s!",
3886
                      device.iv_name, device, secondary_node)
3887
        return False
3888
    #HARDCODE
3889
    if not _CreateBlockDevOnPrimary(lu, instance.primary_node,
3890
                                    instance, device, info):
3891
      logging.error("Failed to create volume %s on primary!", device.iv_name)
3892
      return False
3893

    
3894
  return True
3895

    
3896

    
3897
def _RemoveDisks(lu, instance):
3898
  """Remove all disks for an instance.
3899

3900
  This abstracts away some work from `AddInstance()` and
3901
  `RemoveInstance()`. Note that in case some of the devices couldn't
3902
  be removed, the removal will continue with the other ones (compare
3903
  with `_CreateDisks()`).
3904

3905
  @type lu: L{LogicalUnit}
3906
  @param lu: the logical unit on whose behalf we execute
3907
  @type instance: L{objects.Instance}
3908
  @param instance: the instance whose disks we should remove
3909
  @rtype: boolean
3910
  @return: the success of the removal
3911

3912
  """
3913
  logging.info("Removing block devices for instance %s", instance.name)
3914

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

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

    
3933
  return result
3934

    
3935

    
3936
def _ComputeDiskSize(disk_template, disks):
3937
  """Compute disk size requirements in the volume group
3938

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

    
3949
  if disk_template not in req_size_dict:
3950
    raise errors.ProgrammerError("Disk template '%s' size requirement"
3951
                                 " is unknown" %  disk_template)
3952

    
3953
  return req_size_dict[disk_template]
3954

    
3955

    
3956
def _CheckHVParams(lu, nodenames, hvname, hvparams):
3957
  """Hypervisor parameter validation.
3958

3959
  This function abstract the hypervisor parameter validation to be
3960
  used in both instance create and instance modify.
3961

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

3972
  """
3973
  hvinfo = lu.rpc.call_hypervisor_validate_params(nodenames,
3974
                                                  hvname,
3975
                                                  hvparams)
3976
  for node in nodenames:
3977
    info = hvinfo[node]
3978
    info.Raise()
3979
    if not info.data or not isinstance(info.data, (tuple, list)):
3980
      raise errors.OpPrereqError("Cannot get current information"
3981
                                 " from node '%s' (%s)" % (node, info.data))
3982
    if not info.data[0]:
3983
      raise errors.OpPrereqError("Hypervisor parameter validation failed:"
3984
                                 " %s" % info.data[1])
3985

    
3986

    
3987
class LUCreateInstance(LogicalUnit):
3988
  """Create an instance.
3989