Statistics
| Branch: | Tag: | Revision:

root / lib / cmdlib.py @ 82e12743

History | View | Annotate | Download (236.4 kB)

1
#
2
#
3

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

    
21

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

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

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

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

    
48

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

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

61
  Note that all commands require root permissions.
62

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

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

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

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

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

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

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

    
109
  ssh = property(fget=__GetSSH)
110

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

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

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

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

126
    """
127
    pass
128

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

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

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

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

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

150
    Examples::
151

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

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

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

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

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

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

189
    """
190

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

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

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

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

205
    """
206
    raise NotImplementedError
207

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

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

215
    """
216
    raise NotImplementedError
217

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

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

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

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

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

237
    """
238
    raise NotImplementedError
239

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

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

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

258
    """
259
    return lu_result
260

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

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

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

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

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

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

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

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

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

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

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

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

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

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

    
326
    del self.recalculate_locks[locking.LEVEL_NODE]
327

    
328

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

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

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

    
339

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

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

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

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

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

    
366
  return utils.NiceSort(wanted)
367

    
368

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

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

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

    
385
  if instances:
386
    wanted = []
387

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

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

    
398

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

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

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

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

    
417

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

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

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

    
431

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

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

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

    
443

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

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

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

    
455

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

460
  This builds the hook environment from individual variables.
461

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

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

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

    
509
  env["INSTANCE_NIC_COUNT"] = nic_count
510

    
511
  return env
512

    
513

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

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

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

    
544

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

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

    
560

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

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

    
574

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

578
  """
579
  _OP_REQP = []
580

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

584
    This checks whether the cluster is empty.
585

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

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

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

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

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

    
614

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

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

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

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

636
    Test list:
637

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

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

654
    """
655
    node = nodeinfo.name
656

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

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

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

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

    
677
    bad = False
678

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

    
685
    # checks vg existence and size > 20G
686

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

    
699
    # checks config file checksum
700

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

    
728
    # checks ssh to any
729

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

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

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

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

    
770
    return bad
771

    
772
  def _VerifyInstance(self, instance, instanceconfig, node_vol_is,
773
                      node_instance, feedback_fn, n_offline):
774
    """Verify an instance.
775

776
    This function checks to see if the required block devices are
777
    available on the instance's node.
778

779
    """
780
    bad = False
781

    
782
    node_current = instanceconfig.primary_node
783

    
784
    node_vol_should = {}
785
    instanceconfig.MapLVsByNode(node_vol_should)
786

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

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

    
805
    for node in node_instance:
806
      if (not node == node_current):
807
        if instance in node_instance[node]:
808
          feedback_fn("  - ERROR: instance %s should not run on node %s" %
809
                          (instance, node))
810
          bad = True
811

    
812
    return bad
813

    
814
  def _VerifyOrphanVolumes(self, node_vol_should, node_vol_is, feedback_fn):
815
    """Verify if there are any unknown volumes in the cluster.
816

817
    The .os, .swap and backup volumes are ignored. All other volumes are
818
    reported as unknown.
819

820
    """
821
    bad = False
822

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

    
831
  def _VerifyOrphanInstances(self, instancelist, node_instance, feedback_fn):
832
    """Verify the list of running instances.
833

834
    This checks what instances are running but unknown to the cluster.
835

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

    
846
  def _VerifyNPlusOneMemory(self, node_info, instance_cfg, feedback_fn):
847
    """Verify N+1 Memory Resilience.
848

849
    Check that if one single node dies we can still start all the instances it
850
    was primary for.
851

852
    """
853
    bad = False
854

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

    
876
  def CheckPrereq(self):
877
    """Check prerequisites.
878

879
    Transform the list of checks we're going to skip into a set and check that
880
    all its members are valid.
881

882
    """
883
    self.skip_set = frozenset(self.op.skip_checks)
884
    if not constants.VERIFY_OPTIONAL_CHECKS.issuperset(self.skip_set):
885
      raise errors.OpPrereqError("Invalid checks to be skipped specified")
886

    
887
  def BuildHooksEnv(self):
888
    """Build hooks env.
889

890
    Cluster-Verify hooks just rone in the post phase and their failure makes
891
    the output be logged in the verify output and the verification to fail.
892

893
    """
894
    all_nodes = self.cfg.GetNodeList()
895
    # TODO: populate the environment with useful information for verify hooks
896
    env = {}
897
    return env, [], all_nodes
898

    
899
  def Exec(self, feedback_fn):
900
    """Verify integrity of cluster, performing various test on nodes.
901

902
    """
903
    bad = False
904
    feedback_fn("* Verifying global settings")
905
    for msg in self.cfg.VerifyConfig():
906
      feedback_fn("  - ERROR: %s" % msg)
907

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

    
924
    # FIXME: verify OS list
925
    # do local checksums
926
    master_files = [constants.CLUSTER_CONF_FILE]
927

    
928
    file_names = ssconf.SimpleStore().GetFileList()
929
    file_names.append(constants.SSL_CERT_FILE)
930
    file_names.append(constants.RAPI_CERT_FILE)
931
    file_names.extend(master_files)
932

    
933
    local_checksums = utils.FingerprintFiles(file_names)
934

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

    
954
    cluster = self.cfg.GetClusterInfo()
955
    master_node = self.cfg.GetMasterNode()
956
    all_drbd_map = self.cfg.ComputeDRBDMap()
957

    
958
    for node_i in nodeinfo:
959
      node = node_i.name
960
      nresult = all_nvinfo[node].data
961

    
962
      if node_i.offline:
963
        feedback_fn("* Skipping offline node %s" % (node,))
964
        n_offline.append(node)
965
        continue
966

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

    
978
      if all_nvinfo[node].failed or not isinstance(nresult, dict):
979
        feedback_fn("  - ERROR: connection to %s failed" % (node,))
980
        bad = True
981
        continue
982

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

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

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

    
1013
      node_instance[node] = idata
1014

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

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

    
1041
    node_vol_should = {}
1042

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

    
1051
      inst_config.MapLVsByNode(node_vol_should)
1052

    
1053
      instance_cfg[instance] = inst_config
1054

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

    
1063
      if pnode in n_offline:
1064
        inst_nodes_offline.append(pnode)
1065

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

    
1077
      if not cluster.FillBE(inst_config)[constants.BE_AUTO_BALANCE]:
1078
        i_non_a_balanced.append(instance)
1079

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

    
1093
      if inst_nodes_offline:
1094
        # warn that the instance lives on offline nodes, and set bad=True
1095
        feedback_fn("  - ERROR: instance lives on offline node(s) %s" %
1096
                    ", ".join(inst_nodes_offline))
1097
        bad = True
1098

    
1099
    feedback_fn("* Verifying orphan volumes")
1100
    result = self._VerifyOrphanVolumes(node_vol_should, node_volume,
1101
                                       feedback_fn)
1102
    bad = bad or result
1103

    
1104
    feedback_fn("* Verifying remaining instances")
1105
    result = self._VerifyOrphanInstances(instancelist, node_instance,
1106
                                         feedback_fn)
1107
    bad = bad or result
1108

    
1109
    if constants.VERIFY_NPLUSONE_MEM not in self.skip_set:
1110
      feedback_fn("* Verifying N+1 Memory redundancy")
1111
      result = self._VerifyNPlusOneMemory(node_info, instance_cfg, feedback_fn)
1112
      bad = bad or result
1113

    
1114
    feedback_fn("* Other Notes")
1115
    if i_non_redundant:
1116
      feedback_fn("  - NOTICE: %d non-redundant instance(s) found."
1117
                  % len(i_non_redundant))
1118

    
1119
    if i_non_a_balanced:
1120
      feedback_fn("  - NOTICE: %d non-auto-balanced instance(s) found."
1121
                  % len(i_non_a_balanced))
1122

    
1123
    if n_offline:
1124
      feedback_fn("  - NOTICE: %d offline node(s) found." % len(n_offline))
1125

    
1126
    if n_drained:
1127
      feedback_fn("  - NOTICE: %d drained node(s) found." % len(n_drained))
1128

    
1129
    return not bad
1130

    
1131
  def HooksCallBack(self, phase, hooks_results, feedback_fn, lu_result):
1132
    """Analize the post-hooks' result
1133

1134
    This method analyses the hook result, handles it, and sends some
1135
    nicely-formatted feedback back to the user.
1136

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

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

    
1178
      return lu_result
1179

    
1180

    
1181
class LUVerifyDisks(NoHooksLU):
1182
  """Verifies the cluster disks status.
1183

1184
  """
1185
  _OP_REQP = []
1186
  REQ_BGL = False
1187

    
1188
  def ExpandNames(self):
1189
    self.needed_locks = {
1190
      locking.LEVEL_NODE: locking.ALL_SET,
1191
      locking.LEVEL_INSTANCE: locking.ALL_SET,
1192
    }
1193
    self.share_locks = dict(((i, 1) for i in locking.LEVELS))
1194

    
1195
  def CheckPrereq(self):
1196
    """Check prerequisites.
1197

1198
    This has no prerequisites.
1199

1200
    """
1201
    pass
1202

    
1203
  def Exec(self, feedback_fn):
1204
    """Verify integrity of cluster disks.
1205

1206
    """
1207
    result = res_nodes, res_nlvm, res_instances, res_missing = [], {}, [], {}
1208

    
1209
    vg_name = self.cfg.GetVGName()
1210
    nodes = utils.NiceSort(self.cfg.GetNodeList())
1211
    instances = [self.cfg.GetInstanceInfo(name)
1212
                 for name in self.cfg.GetInstanceList()]
1213

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

    
1226
    if not nv_dict:
1227
      return result
1228

    
1229
    node_lvs = self.rpc.call_volume_list(nodes, vg_name)
1230

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

    
1250
      for lv_name, (_, lv_inactive, lv_online) in lvs.iteritems():
1251
        inst = nv_dict.pop((node, lv_name), None)
1252
        if (not lv_online and inst is not None
1253
            and inst.name not in res_instances):
1254
          res_instances.append(inst.name)
1255

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

    
1263
    return result
1264

    
1265

    
1266
class LURenameCluster(LogicalUnit):
1267
  """Rename the cluster.
1268

1269
  """
1270
  HPATH = "cluster-rename"
1271
  HTYPE = constants.HTYPE_CLUSTER
1272
  _OP_REQP = ["name"]
1273

    
1274
  def BuildHooksEnv(self):
1275
    """Build hooks env.
1276

1277
    """
1278
    env = {
1279
      "OP_TARGET": self.cfg.GetClusterName(),
1280
      "NEW_NAME": self.op.name,
1281
      }
1282
    mn = self.cfg.GetMasterNode()
1283
    return env, [mn], [mn]
1284

    
1285
  def CheckPrereq(self):
1286
    """Verify that the passed name is a valid one.
1287

1288
    """
1289
    hostname = utils.HostInfo(self.op.name)
1290

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

    
1304
    self.op.name = new_name
1305

    
1306
  def Exec(self, feedback_fn):
1307
    """Rename the cluster.
1308

1309
    """
1310
    clustername = self.op.name
1311
    ip = self.ip
1312

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

    
1319
    try:
1320
      cluster = self.cfg.GetClusterInfo()
1321
      cluster.cluster_name = clustername
1322
      cluster.master_ip = ip
1323
      self.cfg.Update(cluster)
1324

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

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

    
1345

    
1346
def _RecursiveCheckIfLVMBased(disk):
1347
  """Check if the given disk or its children are lvm-based.
1348

1349
  @type disk: L{objects.Disk}
1350
  @param disk: the disk to check
1351
  @rtype: booleean
1352
  @return: boolean indicating whether a LD_LV dev_type was found or not
1353

1354
  """
1355
  if disk.children:
1356
    for chdisk in disk.children:
1357
      if _RecursiveCheckIfLVMBased(chdisk):
1358
        return True
1359
  return disk.dev_type == constants.LD_LV
1360

    
1361

    
1362
class LUSetClusterParams(LogicalUnit):
1363
  """Change the parameters of the cluster.
1364

1365
  """
1366
  HPATH = "cluster-modify"
1367
  HTYPE = constants.HTYPE_CLUSTER
1368
  _OP_REQP = []
1369
  REQ_BGL = False
1370

    
1371
  def CheckParameters(self):
1372
    """Check parameters
1373

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

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

    
1394
  def BuildHooksEnv(self):
1395
    """Build hooks env.
1396

1397
    """
1398
    env = {
1399
      "OP_TARGET": self.cfg.GetClusterName(),
1400
      "NEW_VG_NAME": self.op.vg_name,
1401
      }
1402
    mn = self.cfg.GetMasterNode()
1403
    return env, [mn], [mn]
1404

    
1405
  def CheckPrereq(self):
1406
    """Check prerequisites.
1407

1408
    This checks whether the given params don't conflict and
1409
    if the given volume group is valid.
1410

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

    
1420
    node_list = self.acquired_locks[locking.LEVEL_NODE]
1421

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

    
1437
    self.cluster = cluster = self.cfg.GetClusterInfo()
1438
    # validate beparams changes
1439
    if self.op.beparams:
1440
      utils.ForceDictType(self.op.beparams, constants.BES_PARAMETER_TYPES)
1441
      self.new_beparams = cluster.FillDict(
1442
        cluster.beparams[constants.BEGR_DEFAULT], self.op.beparams)
1443

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

    
1455
    if self.op.enabled_hypervisors is not None:
1456
      self.hv_list = self.op.enabled_hypervisors
1457
    else:
1458
      self.hv_list = cluster.enabled_hypervisors
1459

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

    
1472
  def Exec(self, feedback_fn):
1473
    """Change the parameters of the cluster.
1474

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

    
1491
    self.cfg.Update(self.cluster)
1492

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

    
1498

    
1499
class LURedistributeConfig(NoHooksLU):
1500
  """Force the redistribution of cluster configuration.
1501

1502
  This is a very simple LU.
1503

1504
  """
1505
  _OP_REQP = []
1506
  REQ_BGL = False
1507

    
1508
  def ExpandNames(self):
1509
    self.needed_locks = {
1510
      locking.LEVEL_NODE: locking.ALL_SET,
1511
    }
1512
    self.share_locks[locking.LEVEL_NODE] = 1
1513

    
1514
  def CheckPrereq(self):
1515
    """Check prerequisites.
1516

1517
    """
1518

    
1519
  def Exec(self, feedback_fn):
1520
    """Redistribute the configuration.
1521

1522
    """
1523
    self.cfg.Update(self.cfg.GetClusterInfo())
1524

    
1525

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

1529
  """
1530
  if not instance.disks:
1531
    return True
1532

    
1533
  if not oneshot:
1534
    lu.proc.LogInfo("Waiting for instance %s to sync disks." % instance.name)
1535

    
1536
  node = instance.primary_node
1537

    
1538
  for dev in instance.disks:
1539
    lu.cfg.SetDiskID(dev, node)
1540

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

    
1577
    time.sleep(min(60, max_time))
1578

    
1579
  if done:
1580
    lu.proc.LogInfo("Instance %s's disks are in sync." % instance.name)
1581
  return not cumul_degraded
1582

    
1583

    
1584
def _CheckDiskConsistency(lu, dev, node, on_primary, ldisk=False):
1585
  """Check that mirrors are not degraded.
1586

1587
  The ldisk parameter, if True, will change the test from the
1588
  is_degraded attribute (which represents overall non-ok status for
1589
  the device(s)) to the ldisk (representing the local storage status).
1590

1591
  """
1592
  lu.cfg.SetDiskID(dev, node)
1593
  if ldisk:
1594
    idx = 6
1595
  else:
1596
    idx = 5
1597

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

    
1614
  return result
1615

    
1616

    
1617
class LUDiagnoseOS(NoHooksLU):
1618
  """Logical unit for OS diagnose/query.
1619

1620
  """
1621
  _OP_REQP = ["output_fields", "names"]
1622
  REQ_BGL = False
1623
  _FIELDS_STATIC = utils.FieldSet()
1624
  _FIELDS_DYNAMIC = utils.FieldSet("name", "valid", "node_status")
1625

    
1626
  def ExpandNames(self):
1627
    if self.op.names:
1628
      raise errors.OpPrereqError("Selective OS query not supported")
1629

    
1630
    _CheckOutputFields(static=self._FIELDS_STATIC,
1631
                       dynamic=self._FIELDS_DYNAMIC,
1632
                       selected=self.op.output_fields)
1633

    
1634
    # Lock all nodes, in shared mode
1635
    self.needed_locks = {}
1636
    self.share_locks[locking.LEVEL_NODE] = 1
1637
    self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
1638

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

1642
    """
1643

    
1644
  @staticmethod
1645
  def _DiagnoseByOS(node_list, rlist):
1646
    """Remaps a per-node return list into an a per-os per-node dictionary
1647

1648
    @param node_list: a list with the names of all nodes
1649
    @param rlist: a map with node names as keys and OS objects as values
1650

1651
    @rtype: dict
1652
    @returns: a dictionary with osnames as keys and as value another map, with
1653
        nodes as keys and list of OS objects as values, eg::
1654

1655
          {"debian-etch": {"node1": [<object>,...],
1656
                           "node2": [<object>,]}
1657
          }
1658

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

    
1674
  def Exec(self, feedback_fn):
1675
    """Compute the list of OSes.
1676

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

    
1702
    return output
1703

    
1704

    
1705
class LURemoveNode(LogicalUnit):
1706
  """Logical unit for removing a node.
1707

1708
  """
1709
  HPATH = "node-remove"
1710
  HTYPE = constants.HTYPE_NODE
1711
  _OP_REQP = ["node_name"]
1712

    
1713
  def BuildHooksEnv(self):
1714
    """Build hooks env.
1715

1716
    This doesn't run on the target node in the pre phase as a failed
1717
    node would then be impossible to remove.
1718

1719
    """
1720
    env = {
1721
      "OP_TARGET": self.op.node_name,
1722
      "NODE_NAME": self.op.node_name,
1723
      }
1724
    all_nodes = self.cfg.GetNodeList()
1725
    all_nodes.remove(self.op.node_name)
1726
    return env, all_nodes, all_nodes
1727

    
1728
  def CheckPrereq(self):
1729
    """Check prerequisites.
1730

1731
    This checks:
1732
     - the node exists in the configuration
1733
     - it does not have primary or secondary instances
1734
     - it's not the master
1735

1736
    Any errors are signalled by raising errors.OpPrereqError.
1737

1738
    """
1739
    node = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.node_name))
1740
    if node is None:
1741
      raise errors.OpPrereqError, ("Node '%s' is unknown." % self.op.node_name)
1742

    
1743
    instance_list = self.cfg.GetInstanceList()
1744

    
1745
    masternode = self.cfg.GetMasterNode()
1746
    if node.name == masternode:
1747
      raise errors.OpPrereqError("Node is the master node,"
1748
                                 " you need to failover first.")
1749

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

    
1758
  def Exec(self, feedback_fn):
1759
    """Removes the node from the cluster.
1760

1761
    """
1762
    node = self.node
1763
    logging.info("Stopping the node daemon and removing configs from node %s",
1764
                 node.name)
1765

    
1766
    self.context.RemoveNode(node.name)
1767

    
1768
    self.rpc.call_node_leave_cluster(node.name)
1769

    
1770
    # Promote nodes to master candidate as needed
1771
    _AdjustCandidatePool(self)
1772

    
1773

    
1774
class LUQueryNodes(NoHooksLU):
1775
  """Logical unit for querying nodes.
1776

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

    
1787
  _FIELDS_STATIC = utils.FieldSet(
1788
    "name", "pinst_cnt", "sinst_cnt",
1789
    "pinst_list", "sinst_list",
1790
    "pip", "sip", "tags",
1791
    "serial_no",
1792
    "master_candidate",
1793
    "master",
1794
    "offline",
1795
    "drained",
1796
    )
1797

    
1798
  def ExpandNames(self):
1799
    _CheckOutputFields(static=self._FIELDS_STATIC,
1800
                       dynamic=self._FIELDS_DYNAMIC,
1801
                       selected=self.op.output_fields)
1802

    
1803
    self.needed_locks = {}
1804
    self.share_locks[locking.LEVEL_NODE] = 1
1805

    
1806
    if self.op.names:
1807
      self.wanted = _GetWantedNodes(self, self.op.names)
1808
    else:
1809
      self.wanted = locking.ALL_SET
1810

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

    
1817

    
1818
  def CheckPrereq(self):
1819
    """Check prerequisites.
1820

1821
    """
1822
    # The validation of the node list is done in the _GetWantedNodes,
1823
    # if non empty, and if empty, there's no validation to do
1824
    pass
1825

    
1826
  def Exec(self, feedback_fn):
1827
    """Computes the list of nodes and their attributes.
1828

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

    
1842
    nodenames = utils.NiceSort(nodenames)
1843
    nodelist = [all_info[name] for name in nodenames]
1844

    
1845
    # begin data gathering
1846

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

    
1872
    node_to_primary = dict([(name, set()) for name in nodenames])
1873
    node_to_secondary = dict([(name, set()) for name in nodenames])
1874

    
1875
    inst_fields = frozenset(("pinst_cnt", "pinst_list",
1876
                             "sinst_cnt", "sinst_list"))
1877
    if inst_fields & frozenset(self.op.output_fields):
1878
      instancelist = self.cfg.GetInstanceList()
1879

    
1880
      for instance_name in instancelist:
1881
        inst = self.cfg.GetInstanceInfo(instance_name)
1882
        if inst.primary_node in node_to_primary:
1883
          node_to_primary[inst.primary_node].add(inst.name)
1884
        for secnode in inst.secondary_nodes:
1885
          if secnode in node_to_secondary:
1886
            node_to_secondary[secnode].add(inst.name)
1887

    
1888
    master_node = self.cfg.GetMasterNode()
1889

    
1890
    # end data gathering
1891

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

    
1929
    return output
1930

    
1931

    
1932
class LUQueryNodeVolumes(NoHooksLU):
1933
  """Logical unit for getting volumes on node(s).
1934

1935
  """
1936
  _OP_REQP = ["nodes", "output_fields"]
1937
  REQ_BGL = False
1938
  _FIELDS_DYNAMIC = utils.FieldSet("phys", "vg", "name", "size", "instance")
1939
  _FIELDS_STATIC = utils.FieldSet("node")
1940

    
1941
  def ExpandNames(self):
1942
    _CheckOutputFields(static=self._FIELDS_STATIC,
1943
                       dynamic=self._FIELDS_DYNAMIC,
1944
                       selected=self.op.output_fields)
1945

    
1946
    self.needed_locks = {}
1947
    self.share_locks[locking.LEVEL_NODE] = 1
1948
    if not self.op.nodes:
1949
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
1950
    else:
1951
      self.needed_locks[locking.LEVEL_NODE] = \
1952
        _GetWantedNodes(self, self.op.nodes)
1953

    
1954
  def CheckPrereq(self):
1955
    """Check prerequisites.
1956

1957
    This checks that the fields required are valid output fields.
1958

1959
    """
1960
    self.nodes = self.acquired_locks[locking.LEVEL_NODE]
1961

    
1962
  def Exec(self, feedback_fn):
1963
    """Computes the list of nodes and their attributes.
1964

1965
    """
1966
    nodenames = self.nodes
1967
    volumes = self.rpc.call_node_volumes(nodenames)
1968

    
1969
    ilist = [self.cfg.GetInstanceInfo(iname) for iname
1970
             in self.cfg.GetInstanceList()]
1971

    
1972
    lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
1973

    
1974
    output = []
1975
    for node in nodenames:
1976
      if node not in volumes or volumes[node].failed or not volumes[node].data:
1977
        continue
1978

    
1979
      node_vols = volumes[node].data[:]
1980
      node_vols.sort(key=lambda vol: vol['dev'])
1981

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

    
2008
        output.append(node_output)
2009

    
2010
    return output
2011

    
2012

    
2013
class LUAddNode(LogicalUnit):
2014
  """Logical unit for adding node to the cluster.
2015

2016
  """
2017
  HPATH = "node-add"
2018
  HTYPE = constants.HTYPE_NODE
2019
  _OP_REQP = ["node_name"]
2020

    
2021
  def BuildHooksEnv(self):
2022
    """Build hooks env.
2023

2024
    This will run on all nodes before, and on all nodes + the new node after.
2025

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

    
2037
  def CheckPrereq(self):
2038
    """Check prerequisites.
2039

2040
    This checks:
2041
     - the new node is not already in the config
2042
     - it is resolvable
2043
     - its parameters (single/dual homed) matches the cluster
2044

2045
    Any errors are signalled by raising errors.OpPrereqError.
2046

2047
    """
2048
    node_name = self.op.node_name
2049
    cfg = self.cfg
2050

    
2051
    dns_data = utils.HostInfo(node_name)
2052

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

    
2062
    node_list = cfg.GetNodeList()
2063
    if not self.op.readd and node in node_list:
2064
      raise errors.OpPrereqError("Node %s is already in the configuration" %
2065
                                 node)
2066
    elif self.op.readd and node not in node_list:
2067
      raise errors.OpPrereqError("Node %s is not in the configuration" % node)
2068

    
2069
    for existing_node_name in node_list:
2070
      existing_node = cfg.GetNodeInfo(existing_node_name)
2071

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

    
2079
      if (existing_node.primary_ip == primary_ip or
2080
          existing_node.secondary_ip == primary_ip or
2081
          existing_node.primary_ip == secondary_ip or
2082
          existing_node.secondary_ip == secondary_ip):
2083
        raise errors.OpPrereqError("New node ip address(es) conflict with"
2084
                                   " existing node %s" % existing_node.name)
2085

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

    
2099
    # checks reachablity
2100
    if not utils.TcpPing(primary_ip, constants.DEFAULT_NODED_PORT):
2101
      raise errors.OpPrereqError("Node not reachable by ping")
2102

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

    
2110
    cp_size = self.cfg.GetClusterInfo().candidate_pool_size
2111
    mc_now, _ = self.cfg.GetMasterCandidateStats()
2112
    master_candidate = mc_now < cp_size
2113

    
2114
    self.new_node = objects.Node(name=node,
2115
                                 primary_ip=primary_ip,
2116
                                 secondary_ip=secondary_ip,
2117
                                 master_candidate=master_candidate,
2118
                                 offline=False, drained=False)
2119

    
2120
  def Exec(self, feedback_fn):
2121
    """Adds the new node to the cluster.
2122

2123
    """
2124
    new_node = self.new_node
2125
    node = new_node.name
2126

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

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

    
2149
    for i in keyfiles:
2150
      f = open(i, 'r')
2151
      try:
2152
        keyarray.append(f.read())
2153
      finally:
2154
        f.close()
2155

    
2156
    result = self.rpc.call_node_add(node, keyarray[0], keyarray[1],
2157
                                    keyarray[2],
2158
                                    keyarray[3], keyarray[4], keyarray[5])
2159

    
2160
    msg = result.RemoteFailMsg()
2161
    if msg:
2162
      raise errors.OpExecError("Cannot transfer ssh keys to the"
2163
                               " new node: %s" % msg)
2164

    
2165
    # Add node to our /etc/hosts, and add key to known_hosts
2166
    utils.AddHostToEtcHosts(new_node.name)
2167

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

    
2176
    node_verify_list = [self.cfg.GetMasterNode()]
2177
    node_verify_param = {
2178
      'nodelist': [node],
2179
      # TODO: do a node-net-test as well?
2180
    }
2181

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

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

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

    
2210
    to_copy = []
2211
    enabled_hypervisors = self.cfg.GetClusterInfo().enabled_hypervisors
2212
    if constants.HTS_USE_VNC.intersection(enabled_hypervisors):
2213
      to_copy.append(constants.VNC_PASSWORD_FILE)
2214

    
2215
    for fname in to_copy:
2216
      result = self.rpc.call_upload_file([node], fname)
2217
      if result[node].failed or not result[node]:
2218
        logging.error("Could not copy file %s to node %s", fname, node)
2219

    
2220
    if self.op.readd:
2221
      self.context.ReaddNode(new_node)
2222
    else:
2223
      self.context.AddNode(new_node)
2224

    
2225

    
2226
class LUSetNodeParams(LogicalUnit):
2227
  """Modifies the parameters of a node.
2228

2229
  """
2230
  HPATH = "node-modify"
2231
  HTYPE = constants.HTYPE_NODE
2232
  _OP_REQP = ["node_name"]
2233
  REQ_BGL = False
2234

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

    
2250
  def ExpandNames(self):
2251
    self.needed_locks = {locking.LEVEL_NODE: self.op.node_name}
2252

    
2253
  def BuildHooksEnv(self):
2254
    """Build hooks env.
2255

2256
    This runs on the master node.
2257

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

    
2269
  def CheckPrereq(self):
2270
    """Check prerequisites.
2271

2272
    This only checks the instance list against the existing names.
2273

2274
    """
2275
    node = self.node = self.cfg.GetNodeInfo(self.op.node_name)
2276

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

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

    
2299
    return
2300

    
2301
  def Exec(self, feedback_fn):
2302
    """Modifies a node.
2303

2304
    """
2305
    node = self.node
2306

    
2307
    result = []
2308
    changed_mc = False
2309

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

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

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

    
2344
    # this will trigger configuration file update, if needed
2345
    self.cfg.Update(node)
2346
    # this will trigger job queue propagation or cleanup
2347
    if changed_mc:
2348
      self.context.ReaddNode(node)
2349

    
2350
    return result
2351

    
2352

    
2353
class LUQueryClusterInfo(NoHooksLU):
2354
  """Query cluster configuration.
2355

2356
  """
2357
  _OP_REQP = []
2358
  REQ_BGL = False
2359

    
2360
  def ExpandNames(self):
2361
    self.needed_locks = {}
2362

    
2363
  def CheckPrereq(self):
2364
    """No prerequsites needed for this LU.
2365

2366
    """
2367
    pass
2368

    
2369
  def Exec(self, feedback_fn):
2370
    """Return cluster config.
2371

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

    
2391
    return result
2392

    
2393

    
2394
class LUQueryConfigValues(NoHooksLU):
2395
  """Return configuration values.
2396

2397
  """
2398
  _OP_REQP = []
2399
  REQ_BGL = False
2400
  _FIELDS_DYNAMIC = utils.FieldSet()
2401
  _FIELDS_STATIC = utils.FieldSet("cluster_name", "master_node", "drain_flag")
2402

    
2403
  def ExpandNames(self):
2404
    self.needed_locks = {}
2405

    
2406
    _CheckOutputFields(static=self._FIELDS_STATIC,
2407
                       dynamic=self._FIELDS_DYNAMIC,
2408
                       selected=self.op.output_fields)
2409

    
2410
  def CheckPrereq(self):
2411
    """No prerequisites.
2412

2413
    """
2414
    pass
2415

    
2416
  def Exec(self, feedback_fn):
2417
    """Dump a representation of the cluster config to the standard output.
2418

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

    
2433

    
2434
class LUActivateInstanceDisks(NoHooksLU):
2435
  """Bring up an instance's disks.
2436

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

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

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

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

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

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

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

2464
    """
2465
    disks_ok, disks_info = _AssembleInstanceDisks(self, self.instance)
2466
    if not disks_ok:
2467
      raise errors.OpExecError("Cannot activate block devices")
2468

    
2469
    return disks_info
2470

    
2471

    
2472
def _AssembleInstanceDisks(lu, instance, ignore_secondaries=False):
2473
  """Prepare the block devices for an instance.
2474

2475
  This sets up the block devices on all nodes.
2476

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

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

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

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

    
2514
  # FIXME: race condition on drbd migration to primary
2515

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

    
2531
  # leave the disks configured for the primary node
2532
  # this is a workaround that would be fixed better by
2533
  # improving the logical/physical id handling
2534
  for disk in instance.disks:
2535
    lu.cfg.SetDiskID(disk, instance.primary_node)
2536

    
2537
  return disks_ok, device_info
2538

    
2539

    
2540
def _StartInstanceDisks(lu, instance, force):
2541
  """Start the disks of an instance.
2542

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

    
2554

    
2555
class LUDeactivateInstanceDisks(NoHooksLU):
2556
  """Shutdown an instance's disks.
2557

2558
  """
2559
  _OP_REQP = ["instance_name"]
2560
  REQ_BGL = False
2561

    
2562
  def ExpandNames(self):
2563
    self._ExpandAndLockInstance()
2564
    self.needed_locks[locking.LEVEL_NODE] = []
2565
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2566

    
2567
  def DeclareLocks(self, level):
2568
    if level == locking.LEVEL_NODE:
2569
      self._LockInstancesNodes()
2570

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

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

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

    
2581
  def Exec(self, feedback_fn):
2582
    """Deactivate the disks
2583

2584
    """
2585
    instance = self.instance
2586
    _SafeShutdownInstanceDisks(self, instance)
2587

    
2588

    
2589
def _SafeShutdownInstanceDisks(lu, instance):
2590
  """Shutdown block devices of an instance.
2591

2592
  This function checks if an instance is running, before calling
2593
  _ShutdownInstanceDisks.
2594

2595
  """
2596
  ins_l = lu.rpc.call_instance_list([instance.primary_node],
2597
                                      [instance.hypervisor])
2598
  ins_l = ins_l[instance.primary_node]
2599
  if ins_l.failed or not isinstance(ins_l.data, list):
2600
    raise errors.OpExecError("Can't contact node '%s'" %
2601
                             instance.primary_node)
2602

    
2603
  if instance.name in ins_l.data:
2604
    raise errors.OpExecError("Instance is running, can't shutdown"
2605
                             " block devices.")
2606

    
2607
  _ShutdownInstanceDisks(lu, instance)
2608

    
2609

    
2610
def _ShutdownInstanceDisks(lu, instance, ignore_primary=False):
2611
  """Shutdown block devices of an instance.
2612

2613
  This does the shutdown on all nodes of the instance.
2614

2615
  If the ignore_primary is false, errors on the primary node are
2616
  ignored.
2617

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

    
2632

    
2633
def _CheckNodeFreeMemory(lu, node, reason, requested, hypervisor_name):
2634
  """Checks if a node has enough free memory.
2635

2636
  This function check if a given node has the needed amount of free
2637
  memory. In case the node has less memory or we cannot get the
2638
  information from the node, this function raise an OpPrereqError
2639
  exception.
2640

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

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

    
2666

    
2667
class LUStartupInstance(LogicalUnit):
2668
  """Starts an instance.
2669

2670
  """
2671
  HPATH = "instance-start"
2672
  HTYPE = constants.HTYPE_INSTANCE
2673
  _OP_REQP = ["instance_name", "force"]
2674
  REQ_BGL = False
2675

    
2676
  def ExpandNames(self):
2677
    self._ExpandAndLockInstance()
2678

    
2679
  def BuildHooksEnv(self):
2680
    """Build hooks env.
2681

2682
    This runs on master, primary and secondary nodes of the instance.
2683

2684
    """
2685
    env = {
2686
      "FORCE": self.op.force,
2687
      }
2688
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
2689
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2690
    return env, nl, nl
2691

    
2692
  def CheckPrereq(self):
2693
    """Check prerequisites.
2694

2695
    This checks that the instance is in the cluster.
2696

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

    
2702
    _CheckNodeOnline(self, instance.primary_node)
2703

    
2704
    bep = self.cfg.GetClusterInfo().FillBE(instance)
2705
    # check bridges existance
2706
    _CheckInstanceBridgesExist(self, instance)
2707

    
2708
    _CheckNodeFreeMemory(self, instance.primary_node,
2709
                         "starting instance %s" % instance.name,
2710
                         bep[constants.BE_MEMORY], instance.hypervisor)
2711

    
2712
  def Exec(self, feedback_fn):
2713
    """Start the instance.
2714

2715
    """
2716
    instance = self.instance
2717
    force = self.op.force
2718
    extra_args = getattr(self.op, "extra_args", "")
2719

    
2720
    self.cfg.MarkInstanceUp(instance.name)
2721

    
2722
    node_current = instance.primary_node
2723

    
2724
    _StartInstanceDisks(self, instance, force)
2725

    
2726
    result = self.rpc.call_instance_start(node_current, instance, extra_args)
2727
    msg = result.RemoteFailMsg()
2728
    if msg:
2729
      _ShutdownInstanceDisks(self, instance)
2730
      raise errors.OpExecError("Could not start instance: %s" % msg)
2731

    
2732

    
2733
class LURebootInstance(LogicalUnit):
2734
  """Reboot an instance.
2735

2736
  """
2737
  HPATH = "instance-reboot"
2738
  HTYPE = constants.HTYPE_INSTANCE
2739
  _OP_REQP = ["instance_name", "ignore_secondaries", "reboot_type"]
2740
  REQ_BGL = False
2741

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

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

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

2757
    """
2758
    env = {
2759
      "IGNORE_SECONDARIES": self.op.ignore_secondaries,
2760
      }
2761
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
2762
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2763
    return env, nl, nl
2764

    
2765
  def CheckPrereq(self):
2766
    """Check prerequisites.
2767

2768
    This checks that the instance is in the cluster.
2769

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

    
2775
    _CheckNodeOnline(self, instance.primary_node)
2776

    
2777
    # check bridges existance
2778
    _CheckInstanceBridgesExist(self, instance)
2779

    
2780
  def Exec(self, feedback_fn):
2781
    """Reboot the instance.
2782

2783
    """
2784
    instance = self.instance
2785
    ignore_secondaries = self.op.ignore_secondaries
2786
    reboot_type = self.op.reboot_type
2787
    extra_args = getattr(self.op, "extra_args", "")
2788

    
2789
    node_current = instance.primary_node
2790

    
2791
    if reboot_type in [constants.INSTANCE_REBOOT_SOFT,
2792
                       constants.INSTANCE_REBOOT_HARD]:
2793
      result = self.rpc.call_instance_reboot(node_current, instance,
2794
                                             reboot_type, extra_args)
2795
      if result.failed or not result.data:
2796
        raise errors.OpExecError("Could not reboot instance")
2797
    else:
2798
      if not self.rpc.call_instance_shutdown(node_current, instance):
2799
        raise errors.OpExecError("could not shutdown instance for full reboot")
2800
      _ShutdownInstanceDisks(self, instance)
2801
      _StartInstanceDisks(self, instance, ignore_secondaries)
2802
      result = self.rpc.call_instance_start(node_current, instance, extra_args)
2803
      msg = result.RemoteFailMsg()
2804
      if msg:
2805
        _ShutdownInstanceDisks(self, instance)
2806
        raise errors.OpExecError("Could not start instance for"
2807
                                 " full reboot: %s" % msg)
2808

    
2809
    self.cfg.MarkInstanceUp(instance.name)
2810

    
2811

    
2812
class LUShutdownInstance(LogicalUnit):
2813
  """Shutdown an instance.
2814

2815
  """
2816
  HPATH = "instance-stop"
2817
  HTYPE = constants.HTYPE_INSTANCE
2818
  _OP_REQP = ["instance_name"]
2819
  REQ_BGL = False
2820

    
2821
  def ExpandNames(self):
2822
    self._ExpandAndLockInstance()
2823

    
2824
  def BuildHooksEnv(self):
2825
    """Build hooks env.
2826

2827
    This runs on master, primary and secondary nodes of the instance.
2828

2829
    """
2830
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2831
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2832
    return env, nl, nl
2833

    
2834
  def CheckPrereq(self):
2835
    """Check prerequisites.
2836

2837
    This checks that the instance is in the cluster.
2838

2839
    """
2840
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2841
    assert self.instance is not None, \
2842
      "Cannot retrieve locked instance %s" % self.op.instance_name
2843
    _CheckNodeOnline(self, self.instance.primary_node)
2844

    
2845
  def Exec(self, feedback_fn):
2846
    """Shutdown the instance.
2847

2848
    """
2849
    instance = self.instance
2850
    node_current = instance.primary_node
2851
    self.cfg.MarkInstanceDown(instance.name)
2852
    result = self.rpc.call_instance_shutdown(node_current, instance)
2853
    if result.failed or not result.data:
2854
      self.proc.LogWarning("Could not shutdown instance")
2855

    
2856
    _ShutdownInstanceDisks(self, instance)
2857

    
2858

    
2859
class LUReinstallInstance(LogicalUnit):
2860
  """Reinstall an instance.
2861

2862
  """
2863
  HPATH = "instance-reinstall"
2864
  HTYPE = constants.HTYPE_INSTANCE
2865
  _OP_REQP = ["instance_name"]
2866
  REQ_BGL = False
2867

    
2868
  def ExpandNames(self):
2869
    self._ExpandAndLockInstance()
2870

    
2871
  def BuildHooksEnv(self):
2872
    """Build hooks env.
2873

2874
    This runs on master, primary and secondary nodes of the instance.
2875

2876
    """
2877
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2878
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2879
    return env, nl, nl
2880

    
2881
  def CheckPrereq(self):
2882
    """Check prerequisites.
2883

2884
    This checks that the instance is in the cluster and is not running.
2885

2886
    """
2887
    instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2888
    assert instance is not None, \
2889
      "Cannot retrieve locked instance %s" % self.op.instance_name
2890
    _CheckNodeOnline(self, instance.primary_node)
2891

    
2892
    if instance.disk_template == constants.DT_DISKLESS:
2893
      raise errors.OpPrereqError("Instance '%s' has no disks" %
2894
                                 self.op.instance_name)
2895
    if instance.admin_up:
2896
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2897
                                 self.op.instance_name)
2898
    remote_info = self.rpc.call_instance_info(instance.primary_node,
2899
                                              instance.name,
2900
                                              instance.hypervisor)
2901
    if remote_info.failed or remote_info.data:
2902
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
2903
                                 (self.op.instance_name,
2904
                                  instance.primary_node))
2905

    
2906
    self.op.os_type = getattr(self.op, "os_type", None)
2907
    if self.op.os_type is not None:
2908
      # OS verification
2909
      pnode = self.cfg.GetNodeInfo(
2910
        self.cfg.ExpandNodeName(instance.primary_node))
2911
      if pnode is None:
2912
        raise errors.OpPrereqError("Primary node '%s' is unknown" %
2913
                                   self.op.pnode)
2914
      result = self.rpc.call_os_get(pnode.name, self.op.os_type)
2915
      result.Raise()
2916
      if not isinstance(result.data, objects.OS):
2917
        raise errors.OpPrereqError("OS '%s' not in supported OS list for"
2918
                                   " primary node"  % self.op.os_type)
2919

    
2920
    self.instance = instance
2921

    
2922
  def Exec(self, feedback_fn):
2923
    """Reinstall the instance.
2924

2925
    """
2926
    inst = self.instance
2927

    
2928
    if self.op.os_type is not None:
2929
      feedback_fn("Changing OS to '%s'..." % self.op.os_type)
2930
      inst.os = self.op.os_type
2931
      self.cfg.Update(inst)
2932

    
2933
    _StartInstanceDisks(self, inst, None)
2934
    try:
2935
      feedback_fn("Running the instance OS create scripts...")
2936
      result = self.rpc.call_instance_os_add(inst.primary_node, inst)
2937
      msg = result.RemoteFailMsg()
2938
      if msg:
2939
        raise errors.OpExecError("Could not install OS for instance %s"
2940
                                 " on node %s: %s" %
2941
                                 (inst.name, inst.primary_node, msg))
2942
    finally:
2943
      _ShutdownInstanceDisks(self, inst)
2944

    
2945

    
2946
class LURenameInstance(LogicalUnit):
2947
  """Rename an instance.
2948

2949
  """
2950
  HPATH = "instance-rename"
2951
  HTYPE = constants.HTYPE_INSTANCE
2952
  _OP_REQP = ["instance_name", "new_name"]
2953

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

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

2959
    """
2960
    env = _BuildInstanceHookEnvByObject(self, self.instance)
2961
    env["INSTANCE_NEW_NAME"] = self.op.new_name
2962
    nl = [self.cfg.GetMasterNode()] + list(self.instance.all_nodes)
2963
    return env, nl, nl
2964

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

2968
    This checks that the instance is in the cluster and is not running.
2969

2970
    """
2971
    instance = self.cfg.GetInstanceInfo(
2972
      self.cfg.ExpandInstanceName(self.op.instance_name))
2973
    if instance is None:
2974
      raise errors.OpPrereqError("Instance '%s' not known" %
2975
                                 self.op.instance_name)
2976
    _CheckNodeOnline(self, instance.primary_node)
2977

    
2978
    if instance.admin_up:
2979
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2980
                                 self.op.instance_name)
2981
    remote_info = self.rpc.call_instance_info(instance.primary_node,
2982
                                              instance.name,
2983
                                              instance.hypervisor)
2984
    remote_info.Raise()
2985
    if remote_info.data:
2986
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
2987
                                 (self.op.instance_name,
2988
                                  instance.primary_node))
2989
    self.instance = instance
2990

    
2991
    # new name verification
2992
    name_info = utils.HostInfo(self.op.new_name)
2993

    
2994
    self.op.new_name = new_name = name_info.name
2995
    instance_list = self.cfg.GetInstanceList()
2996
    if new_name in instance_list:
2997
      raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
2998
                                 new_name)
2999

    
3000
    if not getattr(self.op, "ignore_ip", False):
3001
      if utils.TcpPing(name_info.ip, constants.DEFAULT_NODED_PORT):
3002
        raise errors.OpPrereqError("IP %s of instance %s already in use" %
3003
                                   (name_info.ip, new_name))
3004

    
3005

    
3006
  def Exec(self, feedback_fn):
3007
    """Reinstall the instance.
3008

3009
    """
3010
    inst = self.instance
3011
    old_name = inst.name
3012

    
3013
    if inst.disk_template == constants.DT_FILE:
3014
      old_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
3015

    
3016
    self.cfg.RenameInstance(inst.name, self.op.new_name)
3017
    # Change the instance lock. This is definitely safe while we hold the BGL
3018
    self.context.glm.remove(locking.LEVEL_INSTANCE, old_name)
3019
    self.context.glm.add(locking.LEVEL_INSTANCE, self.op.new_name)
3020

    
3021
    # re-read the instance from the configuration after rename
3022
    inst = self.cfg.GetInstanceInfo(self.op.new_name)
3023

    
3024
    if inst.disk_template == constants.DT_FILE:
3025
      new_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
3026
      result = self.rpc.call_file_storage_dir_rename(inst.primary_node,
3027
                                                     old_file_storage_dir,
3028
                                                     new_file_storage_dir)
3029
      result.Raise()
3030
      if not result.data:
3031
        raise errors.OpExecError("Could not connect to node '%s' to rename"
3032
                                 " directory '%s' to '%s' (but the instance"
3033
                                 " has been renamed in Ganeti)" % (
3034
                                 inst.primary_node, old_file_storage_dir,
3035
                                 new_file_storage_dir))
3036

    
3037
      if not result.data[0]:
3038
        raise errors.OpExecError("Could not rename directory '%s' to '%s'"
3039
                                 " (but the instance has been renamed in"
3040
                                 " Ganeti)" % (old_file_storage_dir,
3041
                                               new_file_storage_dir))
3042

    
3043
    _StartInstanceDisks(self, inst, None)
3044
    try:
3045
      result = self.rpc.call_instance_run_rename(inst.primary_node, inst,
3046
                                                 old_name)
3047
      msg = result.RemoteFailMsg()
3048
      if msg:
3049
        msg = ("Could not run OS rename script for instance %s on node %s"
3050
               " (but the instance has been renamed in Ganeti): %s" %
3051
               (inst.name, inst.primary_node, msg))
3052
        self.proc.LogWarning(msg)
3053
    finally:
3054
      _ShutdownInstanceDisks(self, inst)
3055

    
3056

    
3057
class LURemoveInstance(LogicalUnit):
3058
  """Remove an instance.
3059

3060
  """
3061
  HPATH = "instance-remove"
3062
  HTYPE = constants.HTYPE_INSTANCE
3063
  _OP_REQP = ["instance_name", "ignore_failures"]
3064
  REQ_BGL = False
3065

    
3066
  def ExpandNames(self):
3067
    self._ExpandAndLockInstance()
3068
    self.needed_locks[locking.LEVEL_NODE] = []
3069
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3070

    
3071
  def DeclareLocks(self, level):
3072
    if level == locking.LEVEL_NODE:
3073
      self._LockInstancesNodes()
3074

    
3075
  def BuildHooksEnv(self):
3076
    """Build hooks env.
3077

3078
    This runs on master, primary and secondary nodes of the instance.
3079

3080
    """
3081
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3082
    nl = [self.cfg.GetMasterNode()]
3083
    return env, nl, nl
3084

    
3085
  def CheckPrereq(self):
3086
    """Check prerequisites.
3087

3088
    This checks that the instance is in the cluster.
3089

3090
    """
3091
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3092
    assert self.instance is not None, \
3093
      "Cannot retrieve locked instance %s" % self.op.instance_name
3094

    
3095
  def Exec(self, feedback_fn):
3096
    """Remove the instance.
3097

3098
    """
3099
    instance = self.instance
3100
    logging.info("Shutting down instance %s on node %s",
3101
                 instance.name, instance.primary_node)
3102

    
3103
    result = self.rpc.call_instance_shutdown(instance.primary_node, instance)
3104
    if result.failed or not result.data:
3105
      if self.op.ignore_failures:
3106
        feedback_fn("Warning: can't shutdown instance")
3107
      else:
3108
        raise errors.OpExecError("Could not shutdown instance %s on node %s" %
3109
                                 (instance.name, instance.primary_node))
3110

    
3111
    logging.info("Removing block devices for instance %s", instance.name)
3112

    
3113
    if not _RemoveDisks(self, instance):
3114
      if self.op.ignore_failures:
3115
        feedback_fn("Warning: can't remove instance's disks")
3116
      else:
3117
        raise errors.OpExecError("Can't remove instance's disks")
3118

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

    
3121
    self.cfg.RemoveInstance(instance.name)
3122
    self.remove_locks[locking.LEVEL_INSTANCE] = instance.name
3123

    
3124

    
3125
class LUQueryInstances(NoHooksLU):
3126
  """Logical unit for querying instances.
3127

3128
  """
3129
  _OP_REQP = ["output_fields", "names", "use_locking"]
3130
  REQ_BGL = False
3131
  _FIELDS_STATIC = utils.FieldSet(*["name", "os", "pnode", "snodes",
3132
                                    "admin_state", "admin_ram",
3133
                                    "disk_template", "ip", "mac", "bridge",
3134
                                    "sda_size", "sdb_size", "vcpus", "tags",
3135
                                    "network_port", "beparams",
3136
                                    "(disk).(size)/([0-9]+)",
3137
                                    "(disk).(sizes)", "disk_usage",
3138
                                    "(nic).(mac|ip|bridge)/([0-9]+)",
3139
                                    "(nic).(macs|ips|bridges)",
3140
                                    "(disk|nic).(count)",
3141
                                    "serial_no", "hypervisor", "hvparams",] +
3142
                                  ["hv/%s" % name
3143
                                   for name in constants.HVS_PARAMETERS] +
3144
                                  ["be/%s" % name
3145
                                   for name in constants.BES_PARAMETERS])
3146
  _FIELDS_DYNAMIC = utils.FieldSet("oper_state", "oper_ram", "status")
3147

    
3148

    
3149
  def ExpandNames(self):
3150
    _CheckOutputFields(static=self._FIELDS_STATIC,
3151
                       dynamic=self._FIELDS_DYNAMIC,
3152
                       selected=self.op.output_fields)
3153

    
3154
    self.needed_locks = {}
3155
    self.share_locks[locking.LEVEL_INSTANCE] = 1
3156
    self.share_locks[locking.LEVEL_NODE] = 1
3157

    
3158
    if self.op.names:
3159
      self.wanted = _GetWantedInstances(self, self.op.names)
3160
    else:
3161
      self.wanted = locking.ALL_SET
3162

    
3163
    self.do_node_query = self._FIELDS_STATIC.NonMatching(self.op.output_fields)
3164
    self.do_locking = self.do_node_query and self.op.use_locking
3165
    if self.do_locking:
3166
      self.needed_locks[locking.LEVEL_INSTANCE] = self.wanted
3167
      self.needed_locks[locking.LEVEL_NODE] = []
3168
      self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3169

    
3170
  def DeclareLocks(self, level):
3171
    if level == locking.LEVEL_NODE and self.do_locking:
3172
      self._LockInstancesNodes()
3173

    
3174
  def CheckPrereq(self):
3175
    """Check prerequisites.
3176

3177
    """
3178
    pass
3179

    
3180
  def Exec(self, feedback_fn):
3181
    """Computes the list of nodes and their attributes.
3182

3183
    """
3184
    all_info = self.cfg.GetAllInstancesInfo()
3185
    if self.wanted == locking.ALL_SET:
3186
      # caller didn't specify instance names, so ordering is not important
3187
      if self.do_locking:
3188
        instance_names = self.acquired_locks[locking.LEVEL_INSTANCE]
3189
      else:
3190
        instance_names = all_info.keys()
3191
      instance_names = utils.NiceSort(instance_names)
3192
    else:
3193
      # caller did specify names, so we must keep the ordering
3194
      if self.do_locking:
3195
        tgt_set = self.acquired_locks[locking.LEVEL_INSTANCE]
3196
      else:
3197
        tgt_set = all_info.keys()
3198
      missing = set(self.wanted).difference(tgt_set)
3199
      if missing:
3200
        raise errors.OpExecError("Some instances were removed before"
3201
                                 " retrieving their data: %s" % missing)
3202
      instance_names = self.wanted
3203

    
3204
    instance_list = [all_info[iname] for iname in instance_names]
3205

    
3206
    # begin data gathering
3207

    
3208
    nodes = frozenset([inst.primary_node for inst in instance_list])
3209
    hv_list = list(set([inst.hypervisor for inst in instance_list]))
3210

    
3211
    bad_nodes = []
3212
    off_nodes = []
3213
    if self.do_node_query:
3214
      live_data = {}
3215
      node_data = self.rpc.call_all_instances_info(nodes, hv_list)
3216
      for name in nodes:
3217
        result = node_data[name]
3218
        if result.offline:
3219
          # offline nodes will be in both lists
3220
          off_nodes.append(name)
3221
        if result.failed:
3222
          bad_nodes.append(name)
3223
        else:
3224
          if result.data:
3225
            live_data.update(result.data)
3226
            # else no instance is alive
3227
    else:
3228
      live_data = dict([(name, {}) for name in instance_names])
3229

    
3230
    # end data gathering
3231

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

    
3360
    return output
3361

    
3362

    
3363
class LUFailoverInstance(LogicalUnit):
3364
  """Failover an instance.
3365

3366
  """
3367
  HPATH = "instance-failover"
3368
  HTYPE = constants.HTYPE_INSTANCE
3369
  _OP_REQP = ["instance_name", "ignore_consistency"]
3370
  REQ_BGL = False
3371

    
3372
  def ExpandNames(self):
3373
    self._ExpandAndLockInstance()
3374
    self.needed_locks[locking.LEVEL_NODE] = []
3375
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3376

    
3377
  def DeclareLocks(self, level):
3378
    if level == locking.LEVEL_NODE:
3379
      self._LockInstancesNodes()
3380

    
3381
  def BuildHooksEnv(self):
3382
    """Build hooks env.
3383

3384
    This runs on master, primary and secondary nodes of the instance.
3385

3386
    """
3387
    env = {
3388
      "IGNORE_CONSISTENCY": self.op.ignore_consistency,
3389
      }
3390
    env.update(_BuildInstanceHookEnvByObject(self, self.instance))
3391
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3392
    return env, nl, nl
3393

    
3394
  def CheckPrereq(self):
3395
    """Check prerequisites.
3396

3397
    This checks that the instance is in the cluster.
3398

3399
    """
3400
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3401
    assert self.instance is not None, \
3402
      "Cannot retrieve locked instance %s" % self.op.instance_name
3403

    
3404
    bep = self.cfg.GetClusterInfo().FillBE(instance)
3405
    if instance.disk_template not in constants.DTS_NET_MIRROR:
3406
      raise errors.OpPrereqError("Instance's disk layout is not"
3407
                                 " network mirrored, cannot failover.")
3408

    
3409
    secondary_nodes = instance.secondary_nodes
3410
    if not secondary_nodes:
3411
      raise errors.ProgrammerError("no secondary node but using "
3412
                                   "a mirrored disk template")
3413

    
3414
    target_node = secondary_nodes[0]
3415
    _CheckNodeOnline(self, target_node)
3416
    _CheckNodeNotDrained(self, target_node)
3417
    # check memory requirements on the secondary node
3418
    _CheckNodeFreeMemory(self, target_node, "failing over instance %s" %
3419
                         instance.name, bep[constants.BE_MEMORY],
3420
                         instance.hypervisor)
3421

    
3422
    # check bridge existance
3423
    brlist = [nic.bridge for nic in instance.nics]
3424
    result = self.rpc.call_bridges_exist(target_node, brlist)
3425
    result.Raise()
3426
    if not result.data:
3427
      raise errors.OpPrereqError("One or more target bridges %s does not"
3428
                                 " exist on destination node '%s'" %
3429
                                 (brlist, target_node))
3430

    
3431
  def Exec(self, feedback_fn):
3432
    """Failover an instance.
3433

3434
    The failover is done by shutting it down on its present node and
3435
    starting it on the secondary.
3436

3437
    """
3438
    instance = self.instance
3439

    
3440
    source_node = instance.primary_node
3441
    target_node = instance.secondary_nodes[0]
3442

    
3443
    feedback_fn("* checking disk consistency between source and target")
3444
    for dev in instance.disks:
3445
      # for drbd, these are drbd over lvm
3446
      if not _CheckDiskConsistency(self, dev, target_node, False):
3447
        if instance.admin_up and not self.op.ignore_consistency:
3448
          raise errors.OpExecError("Disk %s is degraded on target node,"
3449
                                   " aborting failover." % dev.iv_name)
3450

    
3451
    feedback_fn("* shutting down instance on source node")
3452
    logging.info("Shutting down instance %s on node %s",
3453
                 instance.name, source_node)
3454

    
3455
    result = self.rpc.call_instance_shutdown(source_node, instance)
3456
    if result.failed or not result.data:
3457
      if self.op.ignore_consistency:
3458
        self.proc.LogWarning("Could not shutdown instance %s on node %s."
3459
                             " Proceeding"
3460
                             " anyway. Please make sure node %s is down",
3461
                             instance.name, source_node, source_node)
3462
      else:
3463
        raise errors.OpExecError("Could not shutdown instance %s on node %s" %
3464
                                 (instance.name, source_node))
3465

    
3466
    feedback_fn("* deactivating the instance's disks on source node")
3467
    if not _ShutdownInstanceDisks(self, instance, ignore_primary=True):
3468
      raise errors.OpExecError("Can't shut down the instance's disks.")
3469

    
3470
    instance.primary_node = target_node
3471
    # distribute new instance config to the other nodes
3472
    self.cfg.Update(instance)
3473

    
3474
    # Only start the instance if it's marked as up
3475
    if instance.admin_up:
3476
      feedback_fn("* activating the instance's disks on target node")
3477
      logging.info("Starting instance %s on node %s",
3478
                   instance.name, target_node)
3479

    
3480
      disks_ok, dummy = _AssembleInstanceDisks(self, instance,
3481
                                               ignore_secondaries=True)
3482
      if not disks_ok:
3483
        _ShutdownInstanceDisks(self, instance)
3484
        raise errors.OpExecError("Can't activate the instance's disks")
3485

    
3486
      feedback_fn("* starting the instance on the target node")
3487
      result = self.rpc.call_instance_start(target_node, instance, None)
3488
      msg = result.RemoteFailMsg()
3489
      if msg:
3490
        _ShutdownInstanceDisks(self, instance)
3491
        raise errors.OpExecError("Could not start instance %s on node %s: %s" %
3492
                                 (instance.name, target_node, msg))
3493

    
3494

    
3495
class LUMigrateInstance(LogicalUnit):
3496
  """Migrate an instance.
3497

3498
  This is migration without shutting down, compared to the failover,
3499
  which is done with shutdown.
3500

3501
  """
3502
  HPATH = "instance-migrate"
3503
  HTYPE = constants.HTYPE_INSTANCE
3504
  _OP_REQP = ["instance_name", "live", "cleanup"]
3505

    
3506
  REQ_BGL = False
3507

    
3508
  def ExpandNames(self):
3509
    self._ExpandAndLockInstance()
3510
    self.needed_locks[locking.LEVEL_NODE] = []
3511
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
3512

    
3513
  def DeclareLocks(self, level):
3514
    if level == locking.LEVEL_NODE:
3515
      self._LockInstancesNodes()
3516

    
3517
  def BuildHooksEnv(self):
3518
    """Build hooks env.
3519

3520
    This runs on master, primary and secondary nodes of the instance.
3521

3522
    """
3523
    env = _BuildInstanceHookEnvByObject(self, self.instance)
3524
    nl = [self.cfg.GetMasterNode()] + list(self.instance.secondary_nodes)
3525
    return env, nl, nl
3526

    
3527
  def CheckPrereq(self):
3528
    """Check prerequisites.
3529

3530
    This checks that the instance is in the cluster.
3531

3532
    """
3533
    instance = self.cfg.GetInstanceInfo(
3534
      self.cfg.ExpandInstanceName(self.op.instance_name))
3535
    if instance is None:
3536
      raise errors.OpPrereqError("Instance '%s' not known" %
3537
                                 self.op.instance_name)
3538

    
3539
    if instance.disk_template != constants.DT_DRBD8:
3540
      raise errors.OpPrereqError("Instance's disk layout is not"
3541
                                 " drbd8, cannot migrate.")
3542

    
3543
    secondary_nodes = instance.secondary_nodes
3544
    if not secondary_nodes:
3545
      raise errors.ConfigurationError("No secondary node but using"
3546
                                      " drbd8 disk template")
3547

    
3548
    i_be = self.cfg.GetClusterInfo().FillBE(instance)
3549

    
3550
    target_node = secondary_nodes[0]
3551
    # check memory requirements on the secondary node
3552
    _CheckNodeFreeMemory(self, target_node, "migrating instance %s" %
3553
                         instance.name, i_be[constants.BE_MEMORY],
3554
                         instance.hypervisor)
3555

    
3556
    # check bridge existance
3557
    brlist = [nic.bridge for nic in instance.nics]
3558
    result = self.rpc.call_bridges_exist(target_node, brlist)
3559
    if result.failed or not result.data:
3560
      raise errors.OpPrereqError("One or more target bridges %s does not"
3561
                                 " exist on destination node '%s'" %
3562
                                 (brlist, target_node))
3563

    
3564
    if not self.op.cleanup:
3565
      _CheckNodeNotDrained(self, target_node)
3566
      result = self.rpc.call_instance_migratable(instance.primary_node,
3567
                                                 instance)
3568
      msg = result.RemoteFailMsg()
3569
      if msg:
3570
        raise errors.OpPrereqError("Can't migrate: %s - please use failover" %
3571
                                   msg)
3572

    
3573
    self.instance = instance
3574

    
3575
  def _WaitUntilSync(self):
3576
    """Poll with custom rpc for disk sync.
3577

3578
    This uses our own step-based rpc call.
3579

3580
    """
3581
    self.feedback_fn("* wait until resync is done")
3582
    all_done = False
3583
    while not all_done:
3584
      all_done = True
3585
      result = self.rpc.call_drbd_wait_sync(self.all_nodes,
3586
                                            self.nodes_ip,
3587
                                            self.instance.disks)
3588
      min_percent = 100
3589
      for node, nres in result.items():
3590
        msg = nres.RemoteFailMsg()
3591
        if msg:
3592
          raise errors.OpExecError("Cannot resync disks on node %s: %s" %
3593
                                   (node, msg))
3594
        node_done, node_percent = nres.payload
3595
        all_done = all_done and node_done
3596
        if node_percent is not None:
3597
          min_percent = min(min_percent, node_percent)
3598
      if not all_done:
3599
        if min_percent < 100:
3600
          self.feedback_fn("   - progress: %.1f%%" % min_percent)
3601
        time.sleep(2)
3602

    
3603
  def _EnsureSecondary(self, node):
3604
    """Demote a node to secondary.
3605

3606
    """
3607
    self.feedback_fn("* switching node %s to secondary mode" % node)
3608

    
3609
    for dev in self.instance.disks:
3610
      self.cfg.SetDiskID(dev, node)
3611

    
3612
    result = self.rpc.call_blockdev_close(node, self.instance.name,
3613
                                          self.instance.disks)
3614
    msg = result.RemoteFailMsg()
3615
    if msg:
3616
      raise errors.OpExecError("Cannot change disk to secondary on node %s,"
3617
                               " error %s" % (node, msg))
3618

    
3619
  def _GoStandalone(self):
3620
    """Disconnect from the network.
3621

3622
    """
3623
    self.feedback_fn("* changing into standalone mode")
3624
    result = self.rpc.call_drbd_disconnect_net(self.all_nodes, self.nodes_ip,
3625
                                               self.instance.disks)
3626
    for node, nres in result.items():
3627
      msg = nres.RemoteFailMsg()
3628
      if msg:
3629
        raise errors.OpExecError("Cannot disconnect disks node %s,"
3630
                                 " error %s" % (node, msg))
3631

    
3632
  def _GoReconnect(self, multimaster):
3633
    """Reconnect to the network.
3634

3635
    """
3636
    if multimaster:
3637
      msg = "dual-master"
3638
    else:
3639
      msg = "single-master"
3640
    self.feedback_fn("* changing disks into %s mode" % msg)
3641
    result = self.rpc.call_drbd_attach_net(self.all_nodes, self.nodes_ip,
3642
                                           self.instance.disks,
3643
                                           self.instance.name, multimaster)
3644
    for node, nres in result.items():
3645
      msg = nres.RemoteFailMsg()
3646
      if msg:
3647
        raise errors.OpExecError("Cannot change disks config on node %s,"
3648
                                 " error: %s" % (node, msg))
3649

    
3650
  def _ExecCleanup(self):
3651
    """Try to cleanup after a failed migration.
3652

3653
    The cleanup is done by:
3654
      - check that the instance is running only on one node
3655
        (and update the config if needed)
3656
      - change disks on its secondary node to secondary
3657
      - wait until disks are fully synchronized
3658
      - disconnect from the network
3659
      - change disks into single-master mode
3660
      - wait again until disks are fully synchronized
3661

3662
    """
3663
    instance = self.instance
3664
    target_node = self.target_node
3665
    source_node = self.source_node
3666

    
3667
    # check running on only one node
3668
    self.feedback_fn("* checking where the instance actually runs"
3669
                     " (if this hangs, the hypervisor might be in"
3670
                     " a bad state)")
3671
    ins_l = self.rpc.call_instance_list(self.all_nodes, [instance.hypervisor])
3672
    for node, result in ins_l.items():
3673
      result.Raise()
3674
      if not isinstance(result.data, list):
3675
        raise errors.OpExecError("Can't contact node '%s'" % node)
3676

    
3677
    runningon_source = instance.name in ins_l[source_node].data
3678
    runningon_target = instance.name in ins_l[target_node].data
3679

    
3680
    if runningon_source and runningon_target:
3681
      raise errors.OpExecError("Instance seems to be running on two nodes,"
3682
                               " or the hypervisor is confused. You will have"
3683
                               " to ensure manually that it runs only on one"
3684
                               " and restart this operation.")
3685

    
3686
    if not (runningon_source or runningon_target):
3687
      raise errors.OpExecError("Instance does not seem to be running at all."
3688
                               " In this case, it's safer to repair by"
3689
                               " running 'gnt-instance stop' to ensure disk"
3690
                               " shutdown, and then restarting it.")
3691

    
3692
    if runningon_target:
3693
      # the migration has actually succeeded, we need to update the config
3694
      self.feedback_fn("* instance running on secondary node (%s),"
3695
                       " updating config" % target_node)
3696
      instance.primary_node = target_node
3697
      self.cfg.Update(instance)
3698
      demoted_node = source_node
3699
    else:
3700
      self.feedback_fn("* instance confirmed to be running on its"
3701
                       " primary node (%s)" % source_node)
3702
      demoted_node = target_node
3703

    
3704
    self._EnsureSecondary(demoted_node)
3705
    try:
3706
      self._WaitUntilSync()
3707
    except errors.OpExecError:
3708
      # we ignore here errors, since if the device is standalone, it
3709
      # won't be able to sync
3710
      pass
3711
    self._GoStandalone()
3712
    self._GoReconnect(False)
3713
    self._WaitUntilSync()
3714

    
3715
    self.feedback_fn("* done")
3716

    
3717
  def _RevertDiskStatus(self):
3718
    """Try to revert the disk status after a failed migration.
3719

3720
    """
3721
    target_node = self.target_node
3722
    try:
3723
      self._EnsureSecondary(target_node)
3724
      self._GoStandalone()
3725
      self._GoReconnect(False)
3726
      self._WaitUntilSync()
3727
    except errors.OpExecError, err:
3728
      self.LogWarning("Migration failed and I can't reconnect the"
3729
                      " drives: error '%s'\n"
3730
                      "Please look and recover the instance status" %
3731
                      str(err))
3732

    
3733
  def _AbortMigration(self):
3734
    """Call the hypervisor code to abort a started migration.
3735

3736
    """
3737
    instance = self.instance
3738
    target_node = self.target_node
3739
    migration_info = self.migration_info
3740

    
3741
    abort_result = self.rpc.call_finalize_migration(target_node,
3742
                                                    instance,
3743
                                                    migration_info,
3744
                                                    False)
3745
    abort_msg = abort_result.RemoteFailMsg()
3746
    if abort_msg:
3747
      logging.error("Aborting migration failed on target node %s: %s" %
3748
                    (target_node, abort_msg))
3749
      # Don't raise an exception here, as we stil have to try to revert the
3750
      # disk status, even if this step failed.
3751

    
3752
  def _ExecMigration(self):
3753
    """Migrate an instance.
3754

3755
    The migrate is done by:
3756
      - change the disks into dual-master mode
3757
      - wait until disks are fully synchronized again
3758
      - migrate the instance
3759
      - change disks on the new secondary node (the old primary) to secondary
3760
      - wait until disks are fully synchronized
3761
      - change disks into single-master mode
3762

3763
    """
3764
    instance = self.instance
3765
    target_node = self.target_node
3766
    source_node = self.source_node
3767

    
3768
    self.feedback_fn("* checking disk consistency between source and target")
3769
    for dev in instance.disks:
3770
      if not _CheckDiskConsistency(self, dev, target_node, False):
3771
        raise errors.OpExecError("Disk %s is degraded or not fully"
3772
                                 " synchronized on target node,"
3773
                                 " aborting migrate." % dev.iv_name)
3774

    
3775
    # First get the migration information from the remote node
3776
    result = self.rpc.call_migration_info(source_node, instance)
3777
    msg = result.RemoteFailMsg()
3778
    if msg:
3779
      log_err = ("Failed fetching source migration information from %s: %s" %
3780
                 (source_node, msg))
3781
      logging.error(log_err)
3782
      raise errors.OpExecError(log_err)
3783

    
3784
    self.migration_info = migration_info = result.payload
3785

    
3786
    # Then switch the disks to master/master mode
3787
    self._EnsureSecondary(target_node)
3788
    self._GoStandalone()
3789
    self._GoReconnect(True)
3790
    self._WaitUntilSync()
3791

    
3792
    self.feedback_fn("* preparing %s to accept the instance" % target_node)
3793
    result = self.rpc.call_accept_instance(target_node,
3794
                                           instance,
3795
                                           migration_info,
3796
                                           self.nodes_ip[target_node])
3797

    
3798
    msg = result.RemoteFailMsg()
3799
    if msg:
3800
      logging.error("Instance pre-migration failed, trying to revert"
3801
                    " disk status: %s", msg)
3802
      self._AbortMigration()
3803
      self._RevertDiskStatus()
3804
      raise errors.OpExecError("Could not pre-migrate instance %s: %s" %
3805
                               (instance.name, msg))
3806

    
3807
    self.feedback_fn("* migrating instance to %s" % target_node)
3808
    time.sleep(10)
3809
    result = self.rpc.call_instance_migrate(source_node, instance,
3810
                                            self.nodes_ip[target_node],
3811
                                            self.op.live)
3812
    msg = result.RemoteFailMsg()
3813
    if msg:
3814
      logging.error("Instance migration failed, trying to revert"
3815
                    " disk status: %s", msg)
3816
      self._AbortMigration()
3817
      self._RevertDiskStatus()
3818
      raise errors.OpExecError("Could not migrate instance %s: %s" %
3819
                               (instance.name, msg))
3820
    time.sleep(10)
3821

    
3822
    instance.primary_node = target_node
3823
    # distribute new instance config to the other nodes
3824
    self.cfg.Update(instance)
3825

    
3826
    result = self.rpc.call_finalize_migration(target_node,
3827
                                              instance,
3828
                                              migration_info,
3829
                                              True)
3830
    msg = result.RemoteFailMsg()
3831
    if msg:
3832
      logging.error("Instance migration succeeded, but finalization failed:"
3833
                    " %s" % msg)
3834
      raise errors.OpExecError("Could not finalize instance migration: %s" %
3835
                               msg)
3836

    
3837
    self._EnsureSecondary(source_node)
3838
    self._WaitUntilSync()
3839
    self._GoStandalone()
3840
    self._GoReconnect(False)
3841
    self._WaitUntilSync()
3842

    
3843
    self.feedback_fn("* done")
3844

    
3845
  def Exec(self, feedback_fn):
3846
    """Perform the migration.
3847

3848
    """
3849
    self.feedback_fn = feedback_fn
3850

    
3851
    self.source_node = self.instance.primary_node
3852
    self.target_node = self.instance.secondary_nodes[0]
3853
    self.all_nodes = [self.source_node, self.target_node]
3854
    self.nodes_ip = {
3855
      self.source_node: self.cfg.GetNodeInfo(self.source_node).secondary_ip,
3856
      self.target_node: self.cfg.GetNodeInfo(self.target_node).secondary_ip,
3857
      }
3858
    if self.op.cleanup:
3859
      return self._ExecCleanup()
3860
    else:
3861
      return self._ExecMigration()
3862

    
3863

    
3864
def _CreateBlockDev(lu, node, instance, device, force_create,
3865
                    info, force_open):
3866
  """Create a tree of block devices on a given node.
3867

3868
  If this device type has to be created on secondaries, create it and
3869
  all its children.
3870

3871
  If not, just recurse to children keeping the same 'force' value.
3872

3873
  @param lu: the lu on whose behalf we execute
3874
  @param node: the node on which to create the device
3875
  @type instance: L{objects.Instance}
3876
  @param instance: the instance which owns the device
3877
  @type device: L{objects.Disk}
3878
  @param device: the device to create
3879
  @type force_create: boolean
3880
  @param force_create: whether to force creation of this device; this
3881
      will be change to True whenever we find a device which has
3882
      CreateOnSecondary() attribute
3883
  @param info: the extra 'metadata' we should attach to the device
3884
      (this will be represented as a LVM tag)
3885
  @type force_open: boolean
3886
  @param force_open: this parameter will be passes to the
3887
      L{backend.BlockdevCreate} function where it specifies
3888
      whether we run on primary or not, and it affects both
3889
      the child assembly and the device own Open() execution
3890

3891
  """
3892
  if device.CreateOnSecondary():
3893
    force_create = True
3894

    
3895
  if device.children:
3896
    for child in device.children:
3897
      _CreateBlockDev(lu, node, instance, child, force_create,
3898
                      info, force_open)
3899

    
3900
  if not force_create:
3901
    return
3902

    
3903
  _CreateSingleBlockDev(lu, node, instance, device, info, force_open)
3904

    
3905

    
3906
def _CreateSingleBlockDev(lu, node, instance, device, info, force_open):
3907
  """Create a single block device on a given node.
3908

3909
  This will not recurse over children of the device, so they must be
3910
  created in advance.
3911

3912
  @param lu: the lu on whose behalf we execute
3913
  @param node: the node on which to create the device
3914
  @type instance: L{objects.Instance}
3915
  @param instance: the instance which owns the device
3916
  @type device: L{objects.Disk}
3917
  @param device: the device to create
3918
  @param info: the extra 'metadata' we should attach to the device
3919
      (this will be represented as a LVM tag)
3920
  @type force_open: boolean
3921
  @param force_open: this parameter will be passes to the
3922
      L{backend.BlockdevCreate} function where it specifies
3923
      whether we run on primary or not, and it affects both
3924
      the child assembly and the device own Open() execution
3925

3926
  """
3927
  lu.cfg.SetDiskID(device, node)
3928
  result = lu.rpc.call_blockdev_create(node, device, device.size,
3929
                                       instance.name, force_open, info)
3930
  msg = result.RemoteFailMsg()
3931
  if msg:
3932
    raise errors.OpExecError("Can't create block device %s on"
3933
                             " node %s for instance %s: %s" %
3934
                             (device, node, instance.name, msg))
3935
  if device.physical_id is None:
3936
    device.physical_id = result.payload
3937

    
3938

    
3939
def _GenerateUniqueNames(lu, exts):
3940
  """Generate a suitable LV name.
3941

3942
  This will generate a logical volume name for the given instance.
3943

3944
  """
3945
  results = []
3946
  for val in exts:
3947
    new_id = lu.cfg.GenerateUniqueID()
3948
    results.append("%s%s" % (new_id, val))
3949
  return results
3950

    
3951

    
3952
def _GenerateDRBD8Branch(lu, primary, secondary, size, names, iv_name,
3953
                         p_minor, s_minor):
3954
  """Generate a drbd8 device complete with its children.
3955

3956
  """
3957
  port = lu.cfg.AllocatePort()
3958
  vgname = lu.cfg.GetVGName()
3959
  shared_secret = lu.cfg.GenerateDRBDSecret()
3960
  dev_data = objects.Disk(dev_type=constants.LD_LV, size=size,
3961
                          logical_id=(vgname, names[0]))
3962
  dev_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
3963
                          logical_id=(vgname, names[1]))
3964
  drbd_dev = objects.Disk(dev_type=constants.LD_DRBD8, size=size,
3965
                          logical_id=(primary, secondary, port,
3966
                                      p_minor, s_minor,
3967
                                      shared_secret),
3968
                          children=[dev_data, dev_meta],
3969
                          iv_name=iv_name)
3970
  return drbd_dev
3971

    
3972

    
3973
def _GenerateDiskTemplate(lu, template_name,
3974
                          instance_name, primary_node,
3975
                          secondary_nodes, disk_info,
3976
                          file_storage_dir, file_driver,
3977
                          base_index):
3978
  """Generate the entire disk layout for a given template type.
3979

3980
  """
3981
  #TODO: compute space requirements
3982

    
3983
  vgname = lu.cfg.GetVGName()
3984
  disk_count = len(disk_info)
3985
  disks = []
3986
  if template_name == constants.DT_DISKLESS:
3987
    pass
3988
  elif template_name == constants.DT_PLAIN:
3989
    if len(secondary_nodes) != 0:
3990
      raise errors.ProgrammerError("Wrong template configuration")
3991

    
3992
    names = _GenerateUniqueNames(lu, [".disk%d" % i
3993
                                      for i in range(disk_count)])
3994
    for idx, disk in enumerate(disk_info):
3995
      disk_index = idx + base_index
3996
      disk_dev = objects.Disk(dev_type=constants.LD_LV, size=disk["size"],
3997
                              logical_id=(vgname, names[idx]),
3998
                              iv_name="disk/%d" % disk_index,
3999
                              mode=disk["mode"])
4000
      disks.append(disk_dev)
4001
  elif template_name == constants.DT_DRBD8:
4002
    if len(secondary_nodes) != 1:
4003
      raise errors.ProgrammerError("Wrong template configuration")
4004
    remote_node = secondary_nodes[0]
4005
    minors = lu.cfg.AllocateDRBDMinor(
4006
      [primary_node, remote_node] * len(disk_info), instance_name)
4007

    
4008
    names = []
4009
    for lv_prefix in _GenerateUniqueNames(lu, [".disk%d" % i
4010
                                               for i in range(disk_count)]):
4011
      names.append(lv_prefix + "_data")
4012
      names.append(lv_prefix + "_meta")
4013
    for idx, disk in enumerate(disk_info):
4014