Statistics
| Branch: | Tag: | Revision:

root / lib / cmdlib.py @ 155d6c75

History | View | Annotate | Download (181 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

    
34
from ganeti import rpc
35
from ganeti import ssh
36
from ganeti import logger
37
from ganeti import utils
38
from ganeti import errors
39
from ganeti import hypervisor
40
from ganeti import locking
41
from ganeti import constants
42
from ganeti import objects
43
from ganeti import opcodes
44
from ganeti import serializer
45

    
46

    
47
class LogicalUnit(object):
48
  """Logical Unit base class.
49

50
  Subclasses must follow these rules:
51
    - implement ExpandNames
52
    - implement CheckPrereq
53
    - implement Exec
54
    - implement BuildHooksEnv
55
    - redefine HPATH and HTYPE
56
    - optionally redefine their run requirements:
57
        REQ_MASTER: the LU needs to run on the master node
58
        REQ_WSSTORE: the LU needs a writable SimpleStore
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_MASTER = True
68
  REQ_WSSTORE = False
69
  REQ_BGL = True
70

    
71
  def __init__(self, processor, op, context, sstore):
72
    """Constructor for LogicalUnit.
73

74
    This needs to be overriden in derived classes in order to check op
75
    validity.
76

77
    """
78
    self.proc = processor
79
    self.op = op
80
    self.cfg = context.cfg
81
    self.sstore = sstore
82
    self.context = context
83
    self.needed_locks = None
84
    self.acquired_locks = {}
85
    self.share_locks = dict(((i, 0) for i in locking.LEVELS))
86
    # Used to force good behavior when calling helper functions
87
    self.recalculate_locks = {}
88
    self.__ssh = None
89

    
90
    for attr_name in self._OP_REQP:
91
      attr_val = getattr(op, attr_name, None)
92
      if attr_val is None:
93
        raise errors.OpPrereqError("Required parameter '%s' missing" %
94
                                   attr_name)
95

    
96
    if not self.cfg.IsCluster():
97
      raise errors.OpPrereqError("Cluster not initialized yet,"
98
                                 " use 'gnt-cluster init' first.")
99
    if self.REQ_MASTER:
100
      master = sstore.GetMasterNode()
101
      if master != utils.HostInfo().name:
102
        raise errors.OpPrereqError("Commands must be run on the master"
103
                                   " node %s" % master)
104

    
105
  def __GetSSH(self):
106
    """Returns the SshRunner object
107

108
    """
109
    if not self.__ssh:
110
      self.__ssh = ssh.SshRunner(self.sstore)
111
    return self.__ssh
112

    
113
  ssh = property(fget=__GetSSH)
114

    
115
  def ExpandNames(self):
116
    """Expand names for this LU.
117

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

123
    LUs which implement this method must also populate the self.needed_locks
124
    member, as a dict with lock levels as keys, and a list of needed lock names
125
    as values. Rules:
126
      - Use an empty dict if you don't need any lock
127
      - If you don't need any lock at a particular level omit that level
128
      - Don't put anything for the BGL level
129
      - If you want all locks at a level use locking.ALL_SET as a value
130

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

135
    Examples:
136
    # Acquire all nodes and one instance
137
    self.needed_locks = {
138
      locking.LEVEL_NODE: locking.ALL_SET,
139
      locking.LEVEL_INSTANCE: ['instance1.example.tld'],
140
    }
141
    # Acquire just two nodes
142
    self.needed_locks = {
143
      locking.LEVEL_NODE: ['node1.example.tld', 'node2.example.tld'],
144
    }
145
    # Acquire no locks
146
    self.needed_locks = {} # No, you can't leave it to the default value None
147

148
    """
149
    # The implementation of this method is mandatory only if the new LU is
150
    # concurrent, so that old LUs don't need to be changed all at the same
151
    # time.
152
    if self.REQ_BGL:
153
      self.needed_locks = {} # Exclusive LUs don't need locks.
154
    else:
155
      raise NotImplementedError
156

    
157
  def DeclareLocks(self, level):
158
    """Declare LU locking needs for a level
159

160
    While most LUs can just declare their locking needs at ExpandNames time,
161
    sometimes there's the need to calculate some locks after having acquired
162
    the ones before. This function is called just before acquiring locks at a
163
    particular level, but after acquiring the ones at lower levels, and permits
164
    such calculations. It can be used to modify self.needed_locks, and by
165
    default it does nothing.
166

167
    This function is only called if you have something already set in
168
    self.needed_locks for the level.
169

170
    @param level: Locking level which is going to be locked
171
    @type level: member of ganeti.locking.LEVELS
172

173
    """
174

    
175
  def CheckPrereq(self):
176
    """Check prerequisites for this LU.
177

178
    This method should check that the prerequisites for the execution
179
    of this LU are fulfilled. It can do internode communication, but
180
    it should be idempotent - no cluster or system changes are
181
    allowed.
182

183
    The method should raise errors.OpPrereqError in case something is
184
    not fulfilled. Its return value is ignored.
185

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

189
    """
190
    raise NotImplementedError
191

    
192
  def Exec(self, feedback_fn):
193
    """Execute the LU.
194

195
    This method should implement the actual work. It should raise
196
    errors.OpExecError for failures that are somewhat dealt with in
197
    code, or expected.
198

199
    """
200
    raise NotImplementedError
201

    
202
  def BuildHooksEnv(self):
203
    """Build hooks environment for this LU.
204

205
    This method should return a three-node tuple consisting of: a dict
206
    containing the environment that will be used for running the
207
    specific hook for this LU, a list of node names on which the hook
208
    should run before the execution, and a list of node names on which
209
    the hook should run after the execution.
210

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

216
    No nodes should be returned as an empty list (and not None).
217

218
    Note that if the HPATH for a LU class is None, this function will
219
    not be called.
220

221
    """
222
    raise NotImplementedError
223

    
224
  def HooksCallBack(self, phase, hook_results, feedback_fn, lu_result):
225
    """Notify the LU about the results of its hooks.
226

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

233
    Args:
234
      phase: the hooks phase that has just been run
235
      hooks_results: the results of the multi-node hooks rpc call
236
      feedback_fn: function to send feedback back to the caller
237
      lu_result: the previous result this LU had, or None in the PRE phase.
238

239
    """
240
    return lu_result
241

    
242
  def _ExpandAndLockInstance(self):
243
    """Helper function to expand and lock an instance.
244

245
    Many LUs that work on an instance take its name in self.op.instance_name
246
    and need to expand it and then declare the expanded name for locking. This
247
    function does it, and then updates self.op.instance_name to the expanded
248
    name. It also initializes needed_locks as a dict, if this hasn't been done
249
    before.
250

251
    """
252
    if self.needed_locks is None:
253
      self.needed_locks = {}
254
    else:
255
      assert locking.LEVEL_INSTANCE not in self.needed_locks, \
256
        "_ExpandAndLockInstance called with instance-level locks set"
257
    expanded_name = self.cfg.ExpandInstanceName(self.op.instance_name)
258
    if expanded_name is None:
259
      raise errors.OpPrereqError("Instance '%s' not known" %
260
                                  self.op.instance_name)
261
    self.needed_locks[locking.LEVEL_INSTANCE] = expanded_name
262
    self.op.instance_name = expanded_name
263

    
264
  def _LockInstancesNodes(self, primary_only=False):
265
    """Helper function to declare instances' nodes for locking.
266

267
    This function should be called after locking one or more instances to lock
268
    their nodes. Its effect is populating self.needed_locks[locking.LEVEL_NODE]
269
    with all primary or secondary nodes for instances already locked and
270
    present in self.needed_locks[locking.LEVEL_INSTANCE].
271

272
    It should be called from DeclareLocks, and for safety only works if
273
    self.recalculate_locks[locking.LEVEL_NODE] is set.
274

275
    In the future it may grow parameters to just lock some instance's nodes, or
276
    to just lock primaries or secondary nodes, if needed.
277

278
    If should be called in DeclareLocks in a way similar to:
279

280
    if level == locking.LEVEL_NODE:
281
      self._LockInstancesNodes()
282

283
    @type primary_only: boolean
284
    @param primary_only: only lock primary nodes of locked instances
285

286
    """
287
    assert locking.LEVEL_NODE in self.recalculate_locks, \
288
      "_LockInstancesNodes helper function called with no nodes to recalculate"
289

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

    
292
    # For now we'll replace self.needed_locks[locking.LEVEL_NODE], but in the
293
    # future we might want to have different behaviors depending on the value
294
    # of self.recalculate_locks[locking.LEVEL_NODE]
295
    wanted_nodes = []
296
    for instance_name in self.acquired_locks[locking.LEVEL_INSTANCE]:
297
      instance = self.context.cfg.GetInstanceInfo(instance_name)
298
      wanted_nodes.append(instance.primary_node)
299
      if not primary_only:
300
        wanted_nodes.extend(instance.secondary_nodes)
301
    self.needed_locks[locking.LEVEL_NODE] = wanted_nodes
302

    
303
    del self.recalculate_locks[locking.LEVEL_NODE]
304

    
305

    
306
class NoHooksLU(LogicalUnit):
307
  """Simple LU which runs no hooks.
308

309
  This LU is intended as a parent for other LogicalUnits which will
310
  run no hooks, in order to reduce duplicate code.
311

312
  """
313
  HPATH = None
314
  HTYPE = None
315

    
316

    
317
def _GetWantedNodes(lu, nodes):
318
  """Returns list of checked and expanded node names.
319

320
  Args:
321
    nodes: List of nodes (strings) or None for all
322

323
  """
324
  if not isinstance(nodes, list):
325
    raise errors.OpPrereqError("Invalid argument type 'nodes'")
326

    
327
  if not nodes:
328
    raise errors.ProgrammerError("_GetWantedNodes should only be called with a"
329
      " non-empty list of nodes whose name is to be expanded.")
330

    
331
  wanted = []
332
  for name in nodes:
333
    node = lu.cfg.ExpandNodeName(name)
334
    if node is None:
335
      raise errors.OpPrereqError("No such node name '%s'" % name)
336
    wanted.append(node)
337

    
338
  return utils.NiceSort(wanted)
339

    
340

    
341
def _GetWantedInstances(lu, instances):
342
  """Returns list of checked and expanded instance names.
343

344
  Args:
345
    instances: List of instances (strings) or None for all
346

347
  """
348
  if not isinstance(instances, list):
349
    raise errors.OpPrereqError("Invalid argument type 'instances'")
350

    
351
  if instances:
352
    wanted = []
353

    
354
    for name in instances:
355
      instance = lu.cfg.ExpandInstanceName(name)
356
      if instance is None:
357
        raise errors.OpPrereqError("No such instance name '%s'" % name)
358
      wanted.append(instance)
359

    
360
  else:
361
    wanted = lu.cfg.GetInstanceList()
362
  return utils.NiceSort(wanted)
363

    
364

    
365
def _CheckOutputFields(static, dynamic, selected):
366
  """Checks whether all selected fields are valid.
367

368
  Args:
369
    static: Static fields
370
    dynamic: Dynamic fields
371

372
  """
373
  static_fields = frozenset(static)
374
  dynamic_fields = frozenset(dynamic)
375

    
376
  all_fields = static_fields | dynamic_fields
377

    
378
  if not all_fields.issuperset(selected):
379
    raise errors.OpPrereqError("Unknown output fields selected: %s"
380
                               % ",".join(frozenset(selected).
381
                                          difference(all_fields)))
382

    
383

    
384
def _BuildInstanceHookEnv(name, primary_node, secondary_nodes, os_type, status,
385
                          memory, vcpus, nics):
386
  """Builds instance related env variables for hooks from single variables.
387

388
  Args:
389
    secondary_nodes: List of secondary nodes as strings
390
  """
391
  env = {
392
    "OP_TARGET": name,
393
    "INSTANCE_NAME": name,
394
    "INSTANCE_PRIMARY": primary_node,
395
    "INSTANCE_SECONDARIES": " ".join(secondary_nodes),
396
    "INSTANCE_OS_TYPE": os_type,
397
    "INSTANCE_STATUS": status,
398
    "INSTANCE_MEMORY": memory,
399
    "INSTANCE_VCPUS": vcpus,
400
  }
401

    
402
  if nics:
403
    nic_count = len(nics)
404
    for idx, (ip, bridge, mac) in enumerate(nics):
405
      if ip is None:
406
        ip = ""
407
      env["INSTANCE_NIC%d_IP" % idx] = ip
408
      env["INSTANCE_NIC%d_BRIDGE" % idx] = bridge
409
      env["INSTANCE_NIC%d_HWADDR" % idx] = mac
410
  else:
411
    nic_count = 0
412

    
413
  env["INSTANCE_NIC_COUNT"] = nic_count
414

    
415
  return env
416

    
417

    
418
def _BuildInstanceHookEnvByObject(instance, override=None):
419
  """Builds instance related env variables for hooks from an object.
420

421
  Args:
422
    instance: objects.Instance object of instance
423
    override: dict of values to override
424
  """
425
  args = {
426
    'name': instance.name,
427
    'primary_node': instance.primary_node,
428
    'secondary_nodes': instance.secondary_nodes,
429
    'os_type': instance.os,
430
    'status': instance.os,
431
    'memory': instance.memory,
432
    'vcpus': instance.vcpus,
433
    'nics': [(nic.ip, nic.bridge, nic.mac) for nic in instance.nics],
434
  }
435
  if override:
436
    args.update(override)
437
  return _BuildInstanceHookEnv(**args)
438

    
439

    
440
def _CheckInstanceBridgesExist(instance):
441
  """Check that the brigdes needed by an instance exist.
442

443
  """
444
  # check bridges existance
445
  brlist = [nic.bridge for nic in instance.nics]
446
  if not rpc.call_bridges_exist(instance.primary_node, brlist):
447
    raise errors.OpPrereqError("one or more target bridges %s does not"
448
                               " exist on destination node '%s'" %
449
                               (brlist, instance.primary_node))
450

    
451

    
452
class LUDestroyCluster(NoHooksLU):
453
  """Logical unit for destroying the cluster.
454

455
  """
456
  _OP_REQP = []
457

    
458
  def CheckPrereq(self):
459
    """Check prerequisites.
460

461
    This checks whether the cluster is empty.
462

463
    Any errors are signalled by raising errors.OpPrereqError.
464

465
    """
466
    master = self.sstore.GetMasterNode()
467

    
468
    nodelist = self.cfg.GetNodeList()
469
    if len(nodelist) != 1 or nodelist[0] != master:
470
      raise errors.OpPrereqError("There are still %d node(s) in"
471
                                 " this cluster." % (len(nodelist) - 1))
472
    instancelist = self.cfg.GetInstanceList()
473
    if instancelist:
474
      raise errors.OpPrereqError("There are still %d instance(s) in"
475
                                 " this cluster." % len(instancelist))
476

    
477
  def Exec(self, feedback_fn):
478
    """Destroys the cluster.
479

480
    """
481
    master = self.sstore.GetMasterNode()
482
    if not rpc.call_node_stop_master(master, False):
483
      raise errors.OpExecError("Could not disable the master role")
484
    priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
485
    utils.CreateBackup(priv_key)
486
    utils.CreateBackup(pub_key)
487
    return master
488

    
489

    
490
class LUVerifyCluster(LogicalUnit):
491
  """Verifies the cluster status.
492

493
  """
494
  HPATH = "cluster-verify"
495
  HTYPE = constants.HTYPE_CLUSTER
496
  _OP_REQP = ["skip_checks"]
497

    
498
  def _VerifyNode(self, node, file_list, local_cksum, vglist, node_result,
499
                  remote_version, feedback_fn):
500
    """Run multiple tests against a node.
501

502
    Test list:
503
      - compares ganeti version
504
      - checks vg existance and size > 20G
505
      - checks config file checksum
506
      - checks ssh to other nodes
507

508
    Args:
509
      node: name of the node to check
510
      file_list: required list of files
511
      local_cksum: dictionary of local files and their checksums
512

513
    """
514
    # compares ganeti version
515
    local_version = constants.PROTOCOL_VERSION
516
    if not remote_version:
517
      feedback_fn("  - ERROR: connection to %s failed" % (node))
518
      return True
519

    
520
    if local_version != remote_version:
521
      feedback_fn("  - ERROR: sw version mismatch: master %s, node(%s) %s" %
522
                      (local_version, node, remote_version))
523
      return True
524

    
525
    # checks vg existance and size > 20G
526

    
527
    bad = False
528
    if not vglist:
529
      feedback_fn("  - ERROR: unable to check volume groups on node %s." %
530
                      (node,))
531
      bad = True
532
    else:
533
      vgstatus = utils.CheckVolumeGroupSize(vglist, self.cfg.GetVGName(),
534
                                            constants.MIN_VG_SIZE)
535
      if vgstatus:
536
        feedback_fn("  - ERROR: %s on node %s" % (vgstatus, node))
537
        bad = True
538

    
539
    # checks config file checksum
540
    # checks ssh to any
541

    
542
    if 'filelist' not in node_result:
543
      bad = True
544
      feedback_fn("  - ERROR: node hasn't returned file checksum data")
545
    else:
546
      remote_cksum = node_result['filelist']
547
      for file_name in file_list:
548
        if file_name not in remote_cksum:
549
          bad = True
550
          feedback_fn("  - ERROR: file '%s' missing" % file_name)
551
        elif remote_cksum[file_name] != local_cksum[file_name]:
552
          bad = True
553
          feedback_fn("  - ERROR: file '%s' has wrong checksum" % file_name)
554

    
555
    if 'nodelist' not in node_result:
556
      bad = True
557
      feedback_fn("  - ERROR: node hasn't returned node ssh connectivity data")
558
    else:
559
      if node_result['nodelist']:
560
        bad = True
561
        for node in node_result['nodelist']:
562
          feedback_fn("  - ERROR: ssh communication with node '%s': %s" %
563
                          (node, node_result['nodelist'][node]))
564
    if 'node-net-test' not in node_result:
565
      bad = True
566
      feedback_fn("  - ERROR: node hasn't returned node tcp connectivity data")
567
    else:
568
      if node_result['node-net-test']:
569
        bad = True
570
        nlist = utils.NiceSort(node_result['node-net-test'].keys())
571
        for node in nlist:
572
          feedback_fn("  - ERROR: tcp communication with node '%s': %s" %
573
                          (node, node_result['node-net-test'][node]))
574

    
575
    hyp_result = node_result.get('hypervisor', None)
576
    if hyp_result is not None:
577
      feedback_fn("  - ERROR: hypervisor verify failure: '%s'" % hyp_result)
578
    return bad
579

    
580
  def _VerifyInstance(self, instance, instanceconfig, node_vol_is,
581
                      node_instance, feedback_fn):
582
    """Verify an instance.
583

584
    This function checks to see if the required block devices are
585
    available on the instance's node.
586

587
    """
588
    bad = False
589

    
590
    node_current = instanceconfig.primary_node
591

    
592
    node_vol_should = {}
593
    instanceconfig.MapLVsByNode(node_vol_should)
594

    
595
    for node in node_vol_should:
596
      for volume in node_vol_should[node]:
597
        if node not in node_vol_is or volume not in node_vol_is[node]:
598
          feedback_fn("  - ERROR: volume %s missing on node %s" %
599
                          (volume, node))
600
          bad = True
601

    
602
    if not instanceconfig.status == 'down':
603
      if (node_current not in node_instance or
604
          not instance in node_instance[node_current]):
605
        feedback_fn("  - ERROR: instance %s not running on node %s" %
606
                        (instance, node_current))
607
        bad = True
608

    
609
    for node in node_instance:
610
      if (not node == node_current):
611
        if instance in node_instance[node]:
612
          feedback_fn("  - ERROR: instance %s should not run on node %s" %
613
                          (instance, node))
614
          bad = True
615

    
616
    return bad
617

    
618
  def _VerifyOrphanVolumes(self, node_vol_should, node_vol_is, feedback_fn):
619
    """Verify if there are any unknown volumes in the cluster.
620

621
    The .os, .swap and backup volumes are ignored. All other volumes are
622
    reported as unknown.
623

624
    """
625
    bad = False
626

    
627
    for node in node_vol_is:
628
      for volume in node_vol_is[node]:
629
        if node not in node_vol_should or volume not in node_vol_should[node]:
630
          feedback_fn("  - ERROR: volume %s on node %s should not exist" %
631
                      (volume, node))
632
          bad = True
633
    return bad
634

    
635
  def _VerifyOrphanInstances(self, instancelist, node_instance, feedback_fn):
636
    """Verify the list of running instances.
637

638
    This checks what instances are running but unknown to the cluster.
639

640
    """
641
    bad = False
642
    for node in node_instance:
643
      for runninginstance in node_instance[node]:
644
        if runninginstance not in instancelist:
645
          feedback_fn("  - ERROR: instance %s on node %s should not exist" %
646
                          (runninginstance, node))
647
          bad = True
648
    return bad
649

    
650
  def _VerifyNPlusOneMemory(self, node_info, instance_cfg, feedback_fn):
651
    """Verify N+1 Memory Resilience.
652

653
    Check that if one single node dies we can still start all the instances it
654
    was primary for.
655

656
    """
657
    bad = False
658

    
659
    for node, nodeinfo in node_info.iteritems():
660
      # This code checks that every node which is now listed as secondary has
661
      # enough memory to host all instances it is supposed to should a single
662
      # other node in the cluster fail.
663
      # FIXME: not ready for failover to an arbitrary node
664
      # FIXME: does not support file-backed instances
665
      # WARNING: we currently take into account down instances as well as up
666
      # ones, considering that even if they're down someone might want to start
667
      # them even in the event of a node failure.
668
      for prinode, instances in nodeinfo['sinst-by-pnode'].iteritems():
669
        needed_mem = 0
670
        for instance in instances:
671
          needed_mem += instance_cfg[instance].memory
672
        if nodeinfo['mfree'] < needed_mem:
673
          feedback_fn("  - ERROR: not enough memory on node %s to accomodate"
674
                      " failovers should node %s fail" % (node, prinode))
675
          bad = True
676
    return bad
677

    
678
  def CheckPrereq(self):
679
    """Check prerequisites.
680

681
    Transform the list of checks we're going to skip into a set and check that
682
    all its members are valid.
683

684
    """
685
    self.skip_set = frozenset(self.op.skip_checks)
686
    if not constants.VERIFY_OPTIONAL_CHECKS.issuperset(self.skip_set):
687
      raise errors.OpPrereqError("Invalid checks to be skipped specified")
688

    
689
  def BuildHooksEnv(self):
690
    """Build hooks env.
691

692
    Cluster-Verify hooks just rone in the post phase and their failure makes
693
    the output be logged in the verify output and the verification to fail.
694

695
    """
696
    all_nodes = self.cfg.GetNodeList()
697
    # TODO: populate the environment with useful information for verify hooks
698
    env = {}
699
    return env, [], all_nodes
700

    
701
  def Exec(self, feedback_fn):
702
    """Verify integrity of cluster, performing various test on nodes.
703

704
    """
705
    bad = False
706
    feedback_fn("* Verifying global settings")
707
    for msg in self.cfg.VerifyConfig():
708
      feedback_fn("  - ERROR: %s" % msg)
709

    
710
    vg_name = self.cfg.GetVGName()
711
    nodelist = utils.NiceSort(self.cfg.GetNodeList())
712
    nodeinfo = [self.cfg.GetNodeInfo(nname) for nname in nodelist]
713
    instancelist = utils.NiceSort(self.cfg.GetInstanceList())
714
    i_non_redundant = [] # Non redundant instances
715
    node_volume = {}
716
    node_instance = {}
717
    node_info = {}
718
    instance_cfg = {}
719

    
720
    # FIXME: verify OS list
721
    # do local checksums
722
    file_names = list(self.sstore.GetFileList())
723
    file_names.append(constants.SSL_CERT_FILE)
724
    file_names.append(constants.CLUSTER_CONF_FILE)
725
    local_checksums = utils.FingerprintFiles(file_names)
726

    
727
    feedback_fn("* Gathering data (%d nodes)" % len(nodelist))
728
    all_volumeinfo = rpc.call_volume_list(nodelist, vg_name)
729
    all_instanceinfo = rpc.call_instance_list(nodelist)
730
    all_vglist = rpc.call_vg_list(nodelist)
731
    node_verify_param = {
732
      'filelist': file_names,
733
      'nodelist': nodelist,
734
      'hypervisor': None,
735
      'node-net-test': [(node.name, node.primary_ip, node.secondary_ip)
736
                        for node in nodeinfo]
737
      }
738
    all_nvinfo = rpc.call_node_verify(nodelist, node_verify_param)
739
    all_rversion = rpc.call_version(nodelist)
740
    all_ninfo = rpc.call_node_info(nodelist, self.cfg.GetVGName())
741

    
742
    for node in nodelist:
743
      feedback_fn("* Verifying node %s" % node)
744
      result = self._VerifyNode(node, file_names, local_checksums,
745
                                all_vglist[node], all_nvinfo[node],
746
                                all_rversion[node], feedback_fn)
747
      bad = bad or result
748

    
749
      # node_volume
750
      volumeinfo = all_volumeinfo[node]
751

    
752
      if isinstance(volumeinfo, basestring):
753
        feedback_fn("  - ERROR: LVM problem on node %s: %s" %
754
                    (node, volumeinfo[-400:].encode('string_escape')))
755
        bad = True
756
        node_volume[node] = {}
757
      elif not isinstance(volumeinfo, dict):
758
        feedback_fn("  - ERROR: connection to %s failed" % (node,))
759
        bad = True
760
        continue
761
      else:
762
        node_volume[node] = volumeinfo
763

    
764
      # node_instance
765
      nodeinstance = all_instanceinfo[node]
766
      if type(nodeinstance) != list:
767
        feedback_fn("  - ERROR: connection to %s failed" % (node,))
768
        bad = True
769
        continue
770

    
771
      node_instance[node] = nodeinstance
772

    
773
      # node_info
774
      nodeinfo = all_ninfo[node]
775
      if not isinstance(nodeinfo, dict):
776
        feedback_fn("  - ERROR: connection to %s failed" % (node,))
777
        bad = True
778
        continue
779

    
780
      try:
781
        node_info[node] = {
782
          "mfree": int(nodeinfo['memory_free']),
783
          "dfree": int(nodeinfo['vg_free']),
784
          "pinst": [],
785
          "sinst": [],
786
          # dictionary holding all instances this node is secondary for,
787
          # grouped by their primary node. Each key is a cluster node, and each
788
          # value is a list of instances which have the key as primary and the
789
          # current node as secondary.  this is handy to calculate N+1 memory
790
          # availability if you can only failover from a primary to its
791
          # secondary.
792
          "sinst-by-pnode": {},
793
        }
794
      except ValueError:
795
        feedback_fn("  - ERROR: invalid value returned from node %s" % (node,))
796
        bad = True
797
        continue
798

    
799
    node_vol_should = {}
800

    
801
    for instance in instancelist:
802
      feedback_fn("* Verifying instance %s" % instance)
803
      inst_config = self.cfg.GetInstanceInfo(instance)
804
      result =  self._VerifyInstance(instance, inst_config, node_volume,
805
                                     node_instance, feedback_fn)
806
      bad = bad or result
807

    
808
      inst_config.MapLVsByNode(node_vol_should)
809

    
810
      instance_cfg[instance] = inst_config
811

    
812
      pnode = inst_config.primary_node
813
      if pnode in node_info:
814
        node_info[pnode]['pinst'].append(instance)
815
      else:
816
        feedback_fn("  - ERROR: instance %s, connection to primary node"
817
                    " %s failed" % (instance, pnode))
818
        bad = True
819

    
820
      # If the instance is non-redundant we cannot survive losing its primary
821
      # node, so we are not N+1 compliant. On the other hand we have no disk
822
      # templates with more than one secondary so that situation is not well
823
      # supported either.
824
      # FIXME: does not support file-backed instances
825
      if len(inst_config.secondary_nodes) == 0:
826
        i_non_redundant.append(instance)
827
      elif len(inst_config.secondary_nodes) > 1:
828
        feedback_fn("  - WARNING: multiple secondaries for instance %s"
829
                    % instance)
830

    
831
      for snode in inst_config.secondary_nodes:
832
        if snode in node_info:
833
          node_info[snode]['sinst'].append(instance)
834
          if pnode not in node_info[snode]['sinst-by-pnode']:
835
            node_info[snode]['sinst-by-pnode'][pnode] = []
836
          node_info[snode]['sinst-by-pnode'][pnode].append(instance)
837
        else:
838
          feedback_fn("  - ERROR: instance %s, connection to secondary node"
839
                      " %s failed" % (instance, snode))
840

    
841
    feedback_fn("* Verifying orphan volumes")
842
    result = self._VerifyOrphanVolumes(node_vol_should, node_volume,
843
                                       feedback_fn)
844
    bad = bad or result
845

    
846
    feedback_fn("* Verifying remaining instances")
847
    result = self._VerifyOrphanInstances(instancelist, node_instance,
848
                                         feedback_fn)
849
    bad = bad or result
850

    
851
    if constants.VERIFY_NPLUSONE_MEM not in self.skip_set:
852
      feedback_fn("* Verifying N+1 Memory redundancy")
853
      result = self._VerifyNPlusOneMemory(node_info, instance_cfg, feedback_fn)
854
      bad = bad or result
855

    
856
    feedback_fn("* Other Notes")
857
    if i_non_redundant:
858
      feedback_fn("  - NOTICE: %d non-redundant instance(s) found."
859
                  % len(i_non_redundant))
860

    
861
    return not bad
862

    
863
  def HooksCallBack(self, phase, hooks_results, feedback_fn, lu_result):
864
    """Analize the post-hooks' result, handle it, and send some
865
    nicely-formatted feedback back to the user.
866

867
    Args:
868
      phase: the hooks phase that has just been run
869
      hooks_results: the results of the multi-node hooks rpc call
870
      feedback_fn: function to send feedback back to the caller
871
      lu_result: previous Exec result
872

873
    """
874
    # We only really run POST phase hooks, and are only interested in
875
    # their results
876
    if phase == constants.HOOKS_PHASE_POST:
877
      # Used to change hooks' output to proper indentation
878
      indent_re = re.compile('^', re.M)
879
      feedback_fn("* Hooks Results")
880
      if not hooks_results:
881
        feedback_fn("  - ERROR: general communication failure")
882
        lu_result = 1
883
      else:
884
        for node_name in hooks_results:
885
          show_node_header = True
886
          res = hooks_results[node_name]
887
          if res is False or not isinstance(res, list):
888
            feedback_fn("    Communication failure")
889
            lu_result = 1
890
            continue
891
          for script, hkr, output in res:
892
            if hkr == constants.HKR_FAIL:
893
              # The node header is only shown once, if there are
894
              # failing hooks on that node
895
              if show_node_header:
896
                feedback_fn("  Node %s:" % node_name)
897
                show_node_header = False
898
              feedback_fn("    ERROR: Script %s failed, output:" % script)
899
              output = indent_re.sub('      ', output)
900
              feedback_fn("%s" % output)
901
              lu_result = 1
902

    
903
      return lu_result
904

    
905

    
906
class LUVerifyDisks(NoHooksLU):
907
  """Verifies the cluster disks status.
908

909
  """
910
  _OP_REQP = []
911

    
912
  def CheckPrereq(self):
913
    """Check prerequisites.
914

915
    This has no prerequisites.
916

917
    """
918
    pass
919

    
920
  def Exec(self, feedback_fn):
921
    """Verify integrity of cluster disks.
922

923
    """
924
    result = res_nodes, res_nlvm, res_instances, res_missing = [], {}, [], {}
925

    
926
    vg_name = self.cfg.GetVGName()
927
    nodes = utils.NiceSort(self.cfg.GetNodeList())
928
    instances = [self.cfg.GetInstanceInfo(name)
929
                 for name in self.cfg.GetInstanceList()]
930

    
931
    nv_dict = {}
932
    for inst in instances:
933
      inst_lvs = {}
934
      if (inst.status != "up" or
935
          inst.disk_template not in constants.DTS_NET_MIRROR):
936
        continue
937
      inst.MapLVsByNode(inst_lvs)
938
      # transform { iname: {node: [vol,],},} to {(node, vol): iname}
939
      for node, vol_list in inst_lvs.iteritems():
940
        for vol in vol_list:
941
          nv_dict[(node, vol)] = inst
942

    
943
    if not nv_dict:
944
      return result
945

    
946
    node_lvs = rpc.call_volume_list(nodes, vg_name)
947

    
948
    to_act = set()
949
    for node in nodes:
950
      # node_volume
951
      lvs = node_lvs[node]
952

    
953
      if isinstance(lvs, basestring):
954
        logger.Info("error enumerating LVs on node %s: %s" % (node, lvs))
955
        res_nlvm[node] = lvs
956
      elif not isinstance(lvs, dict):
957
        logger.Info("connection to node %s failed or invalid data returned" %
958
                    (node,))
959
        res_nodes.append(node)
960
        continue
961

    
962
      for lv_name, (_, lv_inactive, lv_online) in lvs.iteritems():
963
        inst = nv_dict.pop((node, lv_name), None)
964
        if (not lv_online and inst is not None
965
            and inst.name not in res_instances):
966
          res_instances.append(inst.name)
967

    
968
    # any leftover items in nv_dict are missing LVs, let's arrange the
969
    # data better
970
    for key, inst in nv_dict.iteritems():
971
      if inst.name not in res_missing:
972
        res_missing[inst.name] = []
973
      res_missing[inst.name].append(key)
974

    
975
    return result
976

    
977

    
978
class LURenameCluster(LogicalUnit):
979
  """Rename the cluster.
980

981
  """
982
  HPATH = "cluster-rename"
983
  HTYPE = constants.HTYPE_CLUSTER
984
  _OP_REQP = ["name"]
985
  REQ_WSSTORE = True
986

    
987
  def BuildHooksEnv(self):
988
    """Build hooks env.
989

990
    """
991
    env = {
992
      "OP_TARGET": self.sstore.GetClusterName(),
993
      "NEW_NAME": self.op.name,
994
      }
995
    mn = self.sstore.GetMasterNode()
996
    return env, [mn], [mn]
997

    
998
  def CheckPrereq(self):
999
    """Verify that the passed name is a valid one.
1000

1001
    """
1002
    hostname = utils.HostInfo(self.op.name)
1003

    
1004
    new_name = hostname.name
1005
    self.ip = new_ip = hostname.ip
1006
    old_name = self.sstore.GetClusterName()
1007
    old_ip = self.sstore.GetMasterIP()
1008
    if new_name == old_name and new_ip == old_ip:
1009
      raise errors.OpPrereqError("Neither the name nor the IP address of the"
1010
                                 " cluster has changed")
1011
    if new_ip != old_ip:
1012
      if utils.TcpPing(new_ip, constants.DEFAULT_NODED_PORT):
1013
        raise errors.OpPrereqError("The given cluster IP address (%s) is"
1014
                                   " reachable on the network. Aborting." %
1015
                                   new_ip)
1016

    
1017
    self.op.name = new_name
1018

    
1019
  def Exec(self, feedback_fn):
1020
    """Rename the cluster.
1021

1022
    """
1023
    clustername = self.op.name
1024
    ip = self.ip
1025
    ss = self.sstore
1026

    
1027
    # shutdown the master IP
1028
    master = ss.GetMasterNode()
1029
    if not rpc.call_node_stop_master(master, False):
1030
      raise errors.OpExecError("Could not disable the master role")
1031

    
1032
    try:
1033
      # modify the sstore
1034
      ss.SetKey(ss.SS_MASTER_IP, ip)
1035
      ss.SetKey(ss.SS_CLUSTER_NAME, clustername)
1036

    
1037
      # Distribute updated ss config to all nodes
1038
      myself = self.cfg.GetNodeInfo(master)
1039
      dist_nodes = self.cfg.GetNodeList()
1040
      if myself.name in dist_nodes:
1041
        dist_nodes.remove(myself.name)
1042

    
1043
      logger.Debug("Copying updated ssconf data to all nodes")
1044
      for keyname in [ss.SS_CLUSTER_NAME, ss.SS_MASTER_IP]:
1045
        fname = ss.KeyToFilename(keyname)
1046
        result = rpc.call_upload_file(dist_nodes, fname)
1047
        for to_node in dist_nodes:
1048
          if not result[to_node]:
1049
            logger.Error("copy of file %s to node %s failed" %
1050
                         (fname, to_node))
1051
    finally:
1052
      if not rpc.call_node_start_master(master, False):
1053
        logger.Error("Could not re-enable the master role on the master,"
1054
                     " please restart manually.")
1055

    
1056

    
1057
def _RecursiveCheckIfLVMBased(disk):
1058
  """Check if the given disk or its children are lvm-based.
1059

1060
  Args:
1061
    disk: ganeti.objects.Disk object
1062

1063
  Returns:
1064
    boolean indicating whether a LD_LV dev_type was found or not
1065

1066
  """
1067
  if disk.children:
1068
    for chdisk in disk.children:
1069
      if _RecursiveCheckIfLVMBased(chdisk):
1070
        return True
1071
  return disk.dev_type == constants.LD_LV
1072

    
1073

    
1074
class LUSetClusterParams(LogicalUnit):
1075
  """Change the parameters of the cluster.
1076

1077
  """
1078
  HPATH = "cluster-modify"
1079
  HTYPE = constants.HTYPE_CLUSTER
1080
  _OP_REQP = []
1081

    
1082
  def BuildHooksEnv(self):
1083
    """Build hooks env.
1084

1085
    """
1086
    env = {
1087
      "OP_TARGET": self.sstore.GetClusterName(),
1088
      "NEW_VG_NAME": self.op.vg_name,
1089
      }
1090
    mn = self.sstore.GetMasterNode()
1091
    return env, [mn], [mn]
1092

    
1093
  def CheckPrereq(self):
1094
    """Check prerequisites.
1095

1096
    This checks whether the given params don't conflict and
1097
    if the given volume group is valid.
1098

1099
    """
1100
    if not self.op.vg_name:
1101
      instances = [self.cfg.GetInstanceInfo(name)
1102
                   for name in self.cfg.GetInstanceList()]
1103
      for inst in instances:
1104
        for disk in inst.disks:
1105
          if _RecursiveCheckIfLVMBased(disk):
1106
            raise errors.OpPrereqError("Cannot disable lvm storage while"
1107
                                       " lvm-based instances exist")
1108

    
1109
    # if vg_name not None, checks given volume group on all nodes
1110
    if self.op.vg_name:
1111
      node_list = self.cfg.GetNodeList()
1112
      vglist = rpc.call_vg_list(node_list)
1113
      for node in node_list:
1114
        vgstatus = utils.CheckVolumeGroupSize(vglist[node], self.op.vg_name,
1115
                                              constants.MIN_VG_SIZE)
1116
        if vgstatus:
1117
          raise errors.OpPrereqError("Error on node '%s': %s" %
1118
                                     (node, vgstatus))
1119

    
1120
  def Exec(self, feedback_fn):
1121
    """Change the parameters of the cluster.
1122

1123
    """
1124
    if self.op.vg_name != self.cfg.GetVGName():
1125
      self.cfg.SetVGName(self.op.vg_name)
1126
    else:
1127
      feedback_fn("Cluster LVM configuration already in desired"
1128
                  " state, not changing")
1129

    
1130

    
1131
def _WaitForSync(cfgw, instance, proc, oneshot=False, unlock=False):
1132
  """Sleep and poll for an instance's disk to sync.
1133

1134
  """
1135
  if not instance.disks:
1136
    return True
1137

    
1138
  if not oneshot:
1139
    proc.LogInfo("Waiting for instance %s to sync disks." % instance.name)
1140

    
1141
  node = instance.primary_node
1142

    
1143
  for dev in instance.disks:
1144
    cfgw.SetDiskID(dev, node)
1145

    
1146
  retries = 0
1147
  while True:
1148
    max_time = 0
1149
    done = True
1150
    cumul_degraded = False
1151
    rstats = rpc.call_blockdev_getmirrorstatus(node, instance.disks)
1152
    if not rstats:
1153
      proc.LogWarning("Can't get any data from node %s" % node)
1154
      retries += 1
1155
      if retries >= 10:
1156
        raise errors.RemoteError("Can't contact node %s for mirror data,"
1157
                                 " aborting." % node)
1158
      time.sleep(6)
1159
      continue
1160
    retries = 0
1161
    for i in range(len(rstats)):
1162
      mstat = rstats[i]
1163
      if mstat is None:
1164
        proc.LogWarning("Can't compute data for node %s/%s" %
1165
                        (node, instance.disks[i].iv_name))
1166
        continue
1167
      # we ignore the ldisk parameter
1168
      perc_done, est_time, is_degraded, _ = mstat
1169
      cumul_degraded = cumul_degraded or (is_degraded and perc_done is None)
1170
      if perc_done is not None:
1171
        done = False
1172
        if est_time is not None:
1173
          rem_time = "%d estimated seconds remaining" % est_time
1174
          max_time = est_time
1175
        else:
1176
          rem_time = "no time estimate"
1177
        proc.LogInfo("- device %s: %5.2f%% done, %s" %
1178
                     (instance.disks[i].iv_name, perc_done, rem_time))
1179
    if done or oneshot:
1180
      break
1181

    
1182
    time.sleep(min(60, max_time))
1183

    
1184
  if done:
1185
    proc.LogInfo("Instance %s's disks are in sync." % instance.name)
1186
  return not cumul_degraded
1187

    
1188

    
1189
def _CheckDiskConsistency(cfgw, dev, node, on_primary, ldisk=False):
1190
  """Check that mirrors are not degraded.
1191

1192
  The ldisk parameter, if True, will change the test from the
1193
  is_degraded attribute (which represents overall non-ok status for
1194
  the device(s)) to the ldisk (representing the local storage status).
1195

1196
  """
1197
  cfgw.SetDiskID(dev, node)
1198
  if ldisk:
1199
    idx = 6
1200
  else:
1201
    idx = 5
1202

    
1203
  result = True
1204
  if on_primary or dev.AssembleOnSecondary():
1205
    rstats = rpc.call_blockdev_find(node, dev)
1206
    if not rstats:
1207
      logger.ToStderr("Node %s: Disk degraded, not found or node down" % node)
1208
      result = False
1209
    else:
1210
      result = result and (not rstats[idx])
1211
  if dev.children:
1212
    for child in dev.children:
1213
      result = result and _CheckDiskConsistency(cfgw, child, node, on_primary)
1214

    
1215
  return result
1216

    
1217

    
1218
class LUDiagnoseOS(NoHooksLU):
1219
  """Logical unit for OS diagnose/query.
1220

1221
  """
1222
  _OP_REQP = ["output_fields", "names"]
1223
  REQ_BGL = False
1224

    
1225
  def ExpandNames(self):
1226
    if self.op.names:
1227
      raise errors.OpPrereqError("Selective OS query not supported")
1228

    
1229
    self.dynamic_fields = frozenset(["name", "valid", "node_status"])
1230
    _CheckOutputFields(static=[],
1231
                       dynamic=self.dynamic_fields,
1232
                       selected=self.op.output_fields)
1233

    
1234
    # Lock all nodes, in shared mode
1235
    self.needed_locks = {}
1236
    self.share_locks[locking.LEVEL_NODE] = 1
1237
    self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
1238

    
1239
  def CheckPrereq(self):
1240
    """Check prerequisites.
1241

1242
    """
1243

    
1244
  @staticmethod
1245
  def _DiagnoseByOS(node_list, rlist):
1246
    """Remaps a per-node return list into an a per-os per-node dictionary
1247

1248
      Args:
1249
        node_list: a list with the names of all nodes
1250
        rlist: a map with node names as keys and OS objects as values
1251

1252
      Returns:
1253
        map: a map with osnames as keys and as value another map, with
1254
             nodes as
1255
             keys and list of OS objects as values
1256
             e.g. {"debian-etch": {"node1": [<object>,...],
1257
                                   "node2": [<object>,]}
1258
                  }
1259

1260
    """
1261
    all_os = {}
1262
    for node_name, nr in rlist.iteritems():
1263
      if not nr:
1264
        continue
1265
      for os_obj in nr:
1266
        if os_obj.name not in all_os:
1267
          # build a list of nodes for this os containing empty lists
1268
          # for each node in node_list
1269
          all_os[os_obj.name] = {}
1270
          for nname in node_list:
1271
            all_os[os_obj.name][nname] = []
1272
        all_os[os_obj.name][node_name].append(os_obj)
1273
    return all_os
1274

    
1275
  def Exec(self, feedback_fn):
1276
    """Compute the list of OSes.
1277

1278
    """
1279
    node_list = self.acquired_locks[locking.LEVEL_NODE]
1280
    node_data = rpc.call_os_diagnose(node_list)
1281
    if node_data == False:
1282
      raise errors.OpExecError("Can't gather the list of OSes")
1283
    pol = self._DiagnoseByOS(node_list, node_data)
1284
    output = []
1285
    for os_name, os_data in pol.iteritems():
1286
      row = []
1287
      for field in self.op.output_fields:
1288
        if field == "name":
1289
          val = os_name
1290
        elif field == "valid":
1291
          val = utils.all([osl and osl[0] for osl in os_data.values()])
1292
        elif field == "node_status":
1293
          val = {}
1294
          for node_name, nos_list in os_data.iteritems():
1295
            val[node_name] = [(v.status, v.path) for v in nos_list]
1296
        else:
1297
          raise errors.ParameterError(field)
1298
        row.append(val)
1299
      output.append(row)
1300

    
1301
    return output
1302

    
1303

    
1304
class LURemoveNode(LogicalUnit):
1305
  """Logical unit for removing a node.
1306

1307
  """
1308
  HPATH = "node-remove"
1309
  HTYPE = constants.HTYPE_NODE
1310
  _OP_REQP = ["node_name"]
1311

    
1312
  def BuildHooksEnv(self):
1313
    """Build hooks env.
1314

1315
    This doesn't run on the target node in the pre phase as a failed
1316
    node would then be impossible to remove.
1317

1318
    """
1319
    env = {
1320
      "OP_TARGET": self.op.node_name,
1321
      "NODE_NAME": self.op.node_name,
1322
      }
1323
    all_nodes = self.cfg.GetNodeList()
1324
    all_nodes.remove(self.op.node_name)
1325
    return env, all_nodes, all_nodes
1326

    
1327
  def CheckPrereq(self):
1328
    """Check prerequisites.
1329

1330
    This checks:
1331
     - the node exists in the configuration
1332
     - it does not have primary or secondary instances
1333
     - it's not the master
1334

1335
    Any errors are signalled by raising errors.OpPrereqError.
1336

1337
    """
1338
    node = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.node_name))
1339
    if node is None:
1340
      raise errors.OpPrereqError, ("Node '%s' is unknown." % self.op.node_name)
1341

    
1342
    instance_list = self.cfg.GetInstanceList()
1343

    
1344
    masternode = self.sstore.GetMasterNode()
1345
    if node.name == masternode:
1346
      raise errors.OpPrereqError("Node is the master node,"
1347
                                 " you need to failover first.")
1348

    
1349
    for instance_name in instance_list:
1350
      instance = self.cfg.GetInstanceInfo(instance_name)
1351
      if node.name == instance.primary_node:
1352
        raise errors.OpPrereqError("Instance %s still running on the node,"
1353
                                   " please remove first." % instance_name)
1354
      if node.name in instance.secondary_nodes:
1355
        raise errors.OpPrereqError("Instance %s has node as a secondary,"
1356
                                   " please remove first." % instance_name)
1357
    self.op.node_name = node.name
1358
    self.node = node
1359

    
1360
  def Exec(self, feedback_fn):
1361
    """Removes the node from the cluster.
1362

1363
    """
1364
    node = self.node
1365
    logger.Info("stopping the node daemon and removing configs from node %s" %
1366
                node.name)
1367

    
1368
    self.context.RemoveNode(node.name)
1369

    
1370
    rpc.call_node_leave_cluster(node.name)
1371

    
1372

    
1373
class LUQueryNodes(NoHooksLU):
1374
  """Logical unit for querying nodes.
1375

1376
  """
1377
  _OP_REQP = ["output_fields", "names"]
1378
  REQ_BGL = False
1379

    
1380
  def ExpandNames(self):
1381
    self.dynamic_fields = frozenset([
1382
      "dtotal", "dfree",
1383
      "mtotal", "mnode", "mfree",
1384
      "bootid",
1385
      "ctotal",
1386
      ])
1387

    
1388
    _CheckOutputFields(static=["name", "pinst_cnt", "sinst_cnt",
1389
                               "pinst_list", "sinst_list",
1390
                               "pip", "sip", "tags"],
1391
                       dynamic=self.dynamic_fields,
1392
                       selected=self.op.output_fields)
1393

    
1394
    self.needed_locks = {}
1395
    self.share_locks[locking.LEVEL_NODE] = 1
1396
    # TODO: we could lock nodes only if the user asked for dynamic fields. For
1397
    # that we need atomic ways to get info for a group of nodes from the
1398
    # config, though.
1399
    if not self.op.names:
1400
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
1401
    else:
1402
      self.needed_locks[locking.LEVEL_NODE] = \
1403
        _GetWantedNodes(self, self.op.names)
1404

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

1408
    """
1409
    # This of course is valid only if we locked the nodes
1410
    self.wanted = self.acquired_locks[locking.LEVEL_NODE]
1411

    
1412
  def Exec(self, feedback_fn):
1413
    """Computes the list of nodes and their attributes.
1414

1415
    """
1416
    nodenames = self.wanted
1417
    nodelist = [self.cfg.GetNodeInfo(name) for name in nodenames]
1418

    
1419
    # begin data gathering
1420

    
1421
    if self.dynamic_fields.intersection(self.op.output_fields):
1422
      live_data = {}
1423
      node_data = rpc.call_node_info(nodenames, self.cfg.GetVGName())
1424
      for name in nodenames:
1425
        nodeinfo = node_data.get(name, None)
1426
        if nodeinfo:
1427
          live_data[name] = {
1428
            "mtotal": utils.TryConvert(int, nodeinfo['memory_total']),
1429
            "mnode": utils.TryConvert(int, nodeinfo['memory_dom0']),
1430
            "mfree": utils.TryConvert(int, nodeinfo['memory_free']),
1431
            "dtotal": utils.TryConvert(int, nodeinfo['vg_size']),
1432
            "dfree": utils.TryConvert(int, nodeinfo['vg_free']),
1433
            "ctotal": utils.TryConvert(int, nodeinfo['cpu_total']),
1434
            "bootid": nodeinfo['bootid'],
1435
            }
1436
        else:
1437
          live_data[name] = {}
1438
    else:
1439
      live_data = dict.fromkeys(nodenames, {})
1440

    
1441
    node_to_primary = dict([(name, set()) for name in nodenames])
1442
    node_to_secondary = dict([(name, set()) for name in nodenames])
1443

    
1444
    inst_fields = frozenset(("pinst_cnt", "pinst_list",
1445
                             "sinst_cnt", "sinst_list"))
1446
    if inst_fields & frozenset(self.op.output_fields):
1447
      instancelist = self.cfg.GetInstanceList()
1448

    
1449
      for instance_name in instancelist:
1450
        inst = self.cfg.GetInstanceInfo(instance_name)
1451
        if inst.primary_node in node_to_primary:
1452
          node_to_primary[inst.primary_node].add(inst.name)
1453
        for secnode in inst.secondary_nodes:
1454
          if secnode in node_to_secondary:
1455
            node_to_secondary[secnode].add(inst.name)
1456

    
1457
    # end data gathering
1458

    
1459
    output = []
1460
    for node in nodelist:
1461
      node_output = []
1462
      for field in self.op.output_fields:
1463
        if field == "name":
1464
          val = node.name
1465
        elif field == "pinst_list":
1466
          val = list(node_to_primary[node.name])
1467
        elif field == "sinst_list":
1468
          val = list(node_to_secondary[node.name])
1469
        elif field == "pinst_cnt":
1470
          val = len(node_to_primary[node.name])
1471
        elif field == "sinst_cnt":
1472
          val = len(node_to_secondary[node.name])
1473
        elif field == "pip":
1474
          val = node.primary_ip
1475
        elif field == "sip":
1476
          val = node.secondary_ip
1477
        elif field == "tags":
1478
          val = list(node.GetTags())
1479
        elif field in self.dynamic_fields:
1480
          val = live_data[node.name].get(field, None)
1481
        else:
1482
          raise errors.ParameterError(field)
1483
        node_output.append(val)
1484
      output.append(node_output)
1485

    
1486
    return output
1487

    
1488

    
1489
class LUQueryNodeVolumes(NoHooksLU):
1490
  """Logical unit for getting volumes on node(s).
1491

1492
  """
1493
  _OP_REQP = ["nodes", "output_fields"]
1494
  REQ_BGL = False
1495

    
1496
  def ExpandNames(self):
1497
    _CheckOutputFields(static=["node"],
1498
                       dynamic=["phys", "vg", "name", "size", "instance"],
1499
                       selected=self.op.output_fields)
1500

    
1501
    self.needed_locks = {}
1502
    self.share_locks[locking.LEVEL_NODE] = 1
1503
    if not self.op.nodes:
1504
      self.needed_locks[locking.LEVEL_NODE] = locking.ALL_SET
1505
    else:
1506
      self.needed_locks[locking.LEVEL_NODE] = \
1507
        _GetWantedNodes(self, self.op.nodes)
1508

    
1509
  def CheckPrereq(self):
1510
    """Check prerequisites.
1511

1512
    This checks that the fields required are valid output fields.
1513

1514
    """
1515
    self.nodes = self.acquired_locks[locking.LEVEL_NODE]
1516

    
1517
  def Exec(self, feedback_fn):
1518
    """Computes the list of nodes and their attributes.
1519

1520
    """
1521
    nodenames = self.nodes
1522
    volumes = rpc.call_node_volumes(nodenames)
1523

    
1524
    ilist = [self.cfg.GetInstanceInfo(iname) for iname
1525
             in self.cfg.GetInstanceList()]
1526

    
1527
    lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
1528

    
1529
    output = []
1530
    for node in nodenames:
1531
      if node not in volumes or not volumes[node]:
1532
        continue
1533

    
1534
      node_vols = volumes[node][:]
1535
      node_vols.sort(key=lambda vol: vol['dev'])
1536

    
1537
      for vol in node_vols:
1538
        node_output = []
1539
        for field in self.op.output_fields:
1540
          if field == "node":
1541
            val = node
1542
          elif field == "phys":
1543
            val = vol['dev']
1544
          elif field == "vg":
1545
            val = vol['vg']
1546
          elif field == "name":
1547
            val = vol['name']
1548
          elif field == "size":
1549
            val = int(float(vol['size']))
1550
          elif field == "instance":
1551
            for inst in ilist:
1552
              if node not in lv_by_node[inst]:
1553
                continue
1554
              if vol['name'] in lv_by_node[inst][node]:
1555
                val = inst.name
1556
                break
1557
            else:
1558
              val = '-'
1559
          else:
1560
            raise errors.ParameterError(field)
1561
          node_output.append(str(val))
1562

    
1563
        output.append(node_output)
1564

    
1565
    return output
1566

    
1567

    
1568
class LUAddNode(LogicalUnit):
1569
  """Logical unit for adding node to the cluster.
1570

1571
  """
1572
  HPATH = "node-add"
1573
  HTYPE = constants.HTYPE_NODE
1574
  _OP_REQP = ["node_name"]
1575

    
1576
  def BuildHooksEnv(self):
1577
    """Build hooks env.
1578

1579
    This will run on all nodes before, and on all nodes + the new node after.
1580

1581
    """
1582
    env = {
1583
      "OP_TARGET": self.op.node_name,
1584
      "NODE_NAME": self.op.node_name,
1585
      "NODE_PIP": self.op.primary_ip,
1586
      "NODE_SIP": self.op.secondary_ip,
1587
      }
1588
    nodes_0 = self.cfg.GetNodeList()
1589
    nodes_1 = nodes_0 + [self.op.node_name, ]
1590
    return env, nodes_0, nodes_1
1591

    
1592
  def CheckPrereq(self):
1593
    """Check prerequisites.
1594

1595
    This checks:
1596
     - the new node is not already in the config
1597
     - it is resolvable
1598
     - its parameters (single/dual homed) matches the cluster
1599

1600
    Any errors are signalled by raising errors.OpPrereqError.
1601

1602
    """
1603
    node_name = self.op.node_name
1604
    cfg = self.cfg
1605

    
1606
    dns_data = utils.HostInfo(node_name)
1607

    
1608
    node = dns_data.name
1609
    primary_ip = self.op.primary_ip = dns_data.ip
1610
    secondary_ip = getattr(self.op, "secondary_ip", None)
1611
    if secondary_ip is None:
1612
      secondary_ip = primary_ip
1613
    if not utils.IsValidIP(secondary_ip):
1614
      raise errors.OpPrereqError("Invalid secondary IP given")
1615
    self.op.secondary_ip = secondary_ip
1616

    
1617
    node_list = cfg.GetNodeList()
1618
    if not self.op.readd and node in node_list:
1619
      raise errors.OpPrereqError("Node %s is already in the configuration" %
1620
                                 node)
1621
    elif self.op.readd and node not in node_list:
1622
      raise errors.OpPrereqError("Node %s is not in the configuration" % node)
1623

    
1624
    for existing_node_name in node_list:
1625
      existing_node = cfg.GetNodeInfo(existing_node_name)
1626

    
1627
      if self.op.readd and node == existing_node_name:
1628
        if (existing_node.primary_ip != primary_ip or
1629
            existing_node.secondary_ip != secondary_ip):
1630
          raise errors.OpPrereqError("Readded node doesn't have the same IP"
1631
                                     " address configuration as before")
1632
        continue
1633

    
1634
      if (existing_node.primary_ip == primary_ip or
1635
          existing_node.secondary_ip == primary_ip or
1636
          existing_node.primary_ip == secondary_ip or
1637
          existing_node.secondary_ip == secondary_ip):
1638
        raise errors.OpPrereqError("New node ip address(es) conflict with"
1639
                                   " existing node %s" % existing_node.name)
1640

    
1641
    # check that the type of the node (single versus dual homed) is the
1642
    # same as for the master
1643
    myself = cfg.GetNodeInfo(self.sstore.GetMasterNode())
1644
    master_singlehomed = myself.secondary_ip == myself.primary_ip
1645
    newbie_singlehomed = secondary_ip == primary_ip
1646
    if master_singlehomed != newbie_singlehomed:
1647
      if master_singlehomed:
1648
        raise errors.OpPrereqError("The master has no private ip but the"
1649
                                   " new node has one")
1650
      else:
1651
        raise errors.OpPrereqError("The master has a private ip but the"
1652
                                   " new node doesn't have one")
1653

    
1654
    # checks reachablity
1655
    if not utils.TcpPing(primary_ip, constants.DEFAULT_NODED_PORT):
1656
      raise errors.OpPrereqError("Node not reachable by ping")
1657

    
1658
    if not newbie_singlehomed:
1659
      # check reachability from my secondary ip to newbie's secondary ip
1660
      if not utils.TcpPing(secondary_ip, constants.DEFAULT_NODED_PORT,
1661
                           source=myself.secondary_ip):
1662
        raise errors.OpPrereqError("Node secondary ip not reachable by TCP"
1663
                                   " based ping to noded port")
1664

    
1665
    self.new_node = objects.Node(name=node,
1666
                                 primary_ip=primary_ip,
1667
                                 secondary_ip=secondary_ip)
1668

    
1669
  def Exec(self, feedback_fn):
1670
    """Adds the new node to the cluster.
1671

1672
    """
1673
    new_node = self.new_node
1674
    node = new_node.name
1675

    
1676
    # check connectivity
1677
    result = rpc.call_version([node])[node]
1678
    if result:
1679
      if constants.PROTOCOL_VERSION == result:
1680
        logger.Info("communication to node %s fine, sw version %s match" %
1681
                    (node, result))
1682
      else:
1683
        raise errors.OpExecError("Version mismatch master version %s,"
1684
                                 " node version %s" %
1685
                                 (constants.PROTOCOL_VERSION, result))
1686
    else:
1687
      raise errors.OpExecError("Cannot get version from the new node")
1688

    
1689
    # setup ssh on node
1690
    logger.Info("copy ssh key to node %s" % node)
1691
    priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
1692
    keyarray = []
1693
    keyfiles = [constants.SSH_HOST_DSA_PRIV, constants.SSH_HOST_DSA_PUB,
1694
                constants.SSH_HOST_RSA_PRIV, constants.SSH_HOST_RSA_PUB,
1695
                priv_key, pub_key]
1696

    
1697
    for i in keyfiles:
1698
      f = open(i, 'r')
1699
      try:
1700
        keyarray.append(f.read())
1701
      finally:
1702
        f.close()
1703

    
1704
    result = rpc.call_node_add(node, keyarray[0], keyarray[1], keyarray[2],
1705
                               keyarray[3], keyarray[4], keyarray[5])
1706

    
1707
    if not result:
1708
      raise errors.OpExecError("Cannot transfer ssh keys to the new node")
1709

    
1710
    # Add node to our /etc/hosts, and add key to known_hosts
1711
    utils.AddHostToEtcHosts(new_node.name)
1712

    
1713
    if new_node.secondary_ip != new_node.primary_ip:
1714
      if not rpc.call_node_tcp_ping(new_node.name,
1715
                                    constants.LOCALHOST_IP_ADDRESS,
1716
                                    new_node.secondary_ip,
1717
                                    constants.DEFAULT_NODED_PORT,
1718
                                    10, False):
1719
        raise errors.OpExecError("Node claims it doesn't have the secondary ip"
1720
                                 " you gave (%s). Please fix and re-run this"
1721
                                 " command." % new_node.secondary_ip)
1722

    
1723
    node_verify_list = [self.sstore.GetMasterNode()]
1724
    node_verify_param = {
1725
      'nodelist': [node],
1726
      # TODO: do a node-net-test as well?
1727
    }
1728

    
1729
    result = rpc.call_node_verify(node_verify_list, node_verify_param)
1730
    for verifier in node_verify_list:
1731
      if not result[verifier]:
1732
        raise errors.OpExecError("Cannot communicate with %s's node daemon"
1733
                                 " for remote verification" % verifier)
1734
      if result[verifier]['nodelist']:
1735
        for failed in result[verifier]['nodelist']:
1736
          feedback_fn("ssh/hostname verification failed %s -> %s" %
1737
                      (verifier, result[verifier]['nodelist'][failed]))
1738
        raise errors.OpExecError("ssh/hostname verification failed.")
1739

    
1740
    # Distribute updated /etc/hosts and known_hosts to all nodes,
1741
    # including the node just added
1742
    myself = self.cfg.GetNodeInfo(self.sstore.GetMasterNode())
1743
    dist_nodes = self.cfg.GetNodeList()
1744
    if not self.op.readd:
1745
      dist_nodes.append(node)
1746
    if myself.name in dist_nodes:
1747
      dist_nodes.remove(myself.name)
1748

    
1749
    logger.Debug("Copying hosts and known_hosts to all nodes")
1750
    for fname in (constants.ETC_HOSTS, constants.SSH_KNOWN_HOSTS_FILE):
1751
      result = rpc.call_upload_file(dist_nodes, fname)
1752
      for to_node in dist_nodes:
1753
        if not result[to_node]:
1754
          logger.Error("copy of file %s to node %s failed" %
1755
                       (fname, to_node))
1756

    
1757
    to_copy = self.sstore.GetFileList()
1758
    if self.sstore.GetHypervisorType() == constants.HT_XEN_HVM31:
1759
      to_copy.append(constants.VNC_PASSWORD_FILE)
1760
    for fname in to_copy:
1761
      result = rpc.call_upload_file([node], fname)
1762
      if not result[node]:
1763
        logger.Error("could not copy file %s to node %s" % (fname, node))
1764

    
1765
    if self.op.readd:
1766
      self.context.ReaddNode(new_node)
1767
    else:
1768
      self.context.AddNode(new_node)
1769

    
1770

    
1771
class LUQueryClusterInfo(NoHooksLU):
1772
  """Query cluster configuration.
1773

1774
  """
1775
  _OP_REQP = []
1776
  REQ_MASTER = False
1777
  REQ_BGL = False
1778

    
1779
  def ExpandNames(self):
1780
    self.needed_locks = {}
1781

    
1782
  def CheckPrereq(self):
1783
    """No prerequsites needed for this LU.
1784

1785
    """
1786
    pass
1787

    
1788
  def Exec(self, feedback_fn):
1789
    """Return cluster config.
1790

1791
    """
1792
    result = {
1793
      "name": self.sstore.GetClusterName(),
1794
      "software_version": constants.RELEASE_VERSION,
1795
      "protocol_version": constants.PROTOCOL_VERSION,
1796
      "config_version": constants.CONFIG_VERSION,
1797
      "os_api_version": constants.OS_API_VERSION,
1798
      "export_version": constants.EXPORT_VERSION,
1799
      "master": self.sstore.GetMasterNode(),
1800
      "architecture": (platform.architecture()[0], platform.machine()),
1801
      "hypervisor_type": self.sstore.GetHypervisorType(),
1802
      }
1803

    
1804
    return result
1805

    
1806

    
1807
class LUDumpClusterConfig(NoHooksLU):
1808
  """Return a text-representation of the cluster-config.
1809

1810
  """
1811
  _OP_REQP = []
1812
  REQ_BGL = False
1813

    
1814
  def ExpandNames(self):
1815
    self.needed_locks = {}
1816

    
1817
  def CheckPrereq(self):
1818
    """No prerequisites.
1819

1820
    """
1821
    pass
1822

    
1823
  def Exec(self, feedback_fn):
1824
    """Dump a representation of the cluster config to the standard output.
1825

1826
    """
1827
    return self.cfg.DumpConfig()
1828

    
1829

    
1830
class LUActivateInstanceDisks(NoHooksLU):
1831
  """Bring up an instance's disks.
1832

1833
  """
1834
  _OP_REQP = ["instance_name"]
1835

    
1836
  def CheckPrereq(self):
1837
    """Check prerequisites.
1838

1839
    This checks that the instance is in the cluster.
1840

1841
    """
1842
    instance = self.cfg.GetInstanceInfo(
1843
      self.cfg.ExpandInstanceName(self.op.instance_name))
1844
    if instance is None:
1845
      raise errors.OpPrereqError("Instance '%s' not known" %
1846
                                 self.op.instance_name)
1847
    self.instance = instance
1848

    
1849

    
1850
  def Exec(self, feedback_fn):
1851
    """Activate the disks.
1852

1853
    """
1854
    disks_ok, disks_info = _AssembleInstanceDisks(self.instance, self.cfg)
1855
    if not disks_ok:
1856
      raise errors.OpExecError("Cannot activate block devices")
1857

    
1858
    return disks_info
1859

    
1860

    
1861
def _AssembleInstanceDisks(instance, cfg, ignore_secondaries=False):
1862
  """Prepare the block devices for an instance.
1863

1864
  This sets up the block devices on all nodes.
1865

1866
  Args:
1867
    instance: a ganeti.objects.Instance object
1868
    ignore_secondaries: if true, errors on secondary nodes won't result
1869
                        in an error return from the function
1870

1871
  Returns:
1872
    false if the operation failed
1873
    list of (host, instance_visible_name, node_visible_name) if the operation
1874
         suceeded with the mapping from node devices to instance devices
1875
  """
1876
  device_info = []
1877
  disks_ok = True
1878
  iname = instance.name
1879
  # With the two passes mechanism we try to reduce the window of
1880
  # opportunity for the race condition of switching DRBD to primary
1881
  # before handshaking occured, but we do not eliminate it
1882

    
1883
  # The proper fix would be to wait (with some limits) until the
1884
  # connection has been made and drbd transitions from WFConnection
1885
  # into any other network-connected state (Connected, SyncTarget,
1886
  # SyncSource, etc.)
1887

    
1888
  # 1st pass, assemble on all nodes in secondary mode
1889
  for inst_disk in instance.disks:
1890
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
1891
      cfg.SetDiskID(node_disk, node)
1892
      result = rpc.call_blockdev_assemble(node, node_disk, iname, False)
1893
      if not result:
1894
        logger.Error("could not prepare block device %s on node %s"
1895
                     " (is_primary=False, pass=1)" % (inst_disk.iv_name, node))
1896
        if not ignore_secondaries:
1897
          disks_ok = False
1898

    
1899
  # FIXME: race condition on drbd migration to primary
1900

    
1901
  # 2nd pass, do only the primary node
1902
  for inst_disk in instance.disks:
1903
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
1904
      if node != instance.primary_node:
1905
        continue
1906
      cfg.SetDiskID(node_disk, node)
1907
      result = rpc.call_blockdev_assemble(node, node_disk, iname, True)
1908
      if not result:
1909
        logger.Error("could not prepare block device %s on node %s"
1910
                     " (is_primary=True, pass=2)" % (inst_disk.iv_name, node))
1911
        disks_ok = False
1912
    device_info.append((instance.primary_node, inst_disk.iv_name, result))
1913

    
1914
  # leave the disks configured for the primary node
1915
  # this is a workaround that would be fixed better by
1916
  # improving the logical/physical id handling
1917
  for disk in instance.disks:
1918
    cfg.SetDiskID(disk, instance.primary_node)
1919

    
1920
  return disks_ok, device_info
1921

    
1922

    
1923
def _StartInstanceDisks(cfg, instance, force):
1924
  """Start the disks of an instance.
1925

1926
  """
1927
  disks_ok, dummy = _AssembleInstanceDisks(instance, cfg,
1928
                                           ignore_secondaries=force)
1929
  if not disks_ok:
1930
    _ShutdownInstanceDisks(instance, cfg)
1931
    if force is not None and not force:
1932
      logger.Error("If the message above refers to a secondary node,"
1933
                   " you can retry the operation using '--force'.")
1934
    raise errors.OpExecError("Disk consistency error")
1935

    
1936

    
1937
class LUDeactivateInstanceDisks(NoHooksLU):
1938
  """Shutdown an instance's disks.
1939

1940
  """
1941
  _OP_REQP = ["instance_name"]
1942

    
1943
  def CheckPrereq(self):
1944
    """Check prerequisites.
1945

1946
    This checks that the instance is in the cluster.
1947

1948
    """
1949
    instance = self.cfg.GetInstanceInfo(
1950
      self.cfg.ExpandInstanceName(self.op.instance_name))
1951
    if instance is None:
1952
      raise errors.OpPrereqError("Instance '%s' not known" %
1953
                                 self.op.instance_name)
1954
    self.instance = instance
1955

    
1956
  def Exec(self, feedback_fn):
1957
    """Deactivate the disks
1958

1959
    """
1960
    instance = self.instance
1961
    _SafeShutdownInstanceDisks(instance, self.cfg)
1962

    
1963

    
1964
def _SafeShutdownInstanceDisks(instance, cfg):
1965
  """Shutdown block devices of an instance.
1966

1967
  This function checks if an instance is running, before calling
1968
  _ShutdownInstanceDisks.
1969

1970
  """
1971
  ins_l = rpc.call_instance_list([instance.primary_node])
1972
  ins_l = ins_l[instance.primary_node]
1973
  if not type(ins_l) is list:
1974
    raise errors.OpExecError("Can't contact node '%s'" %
1975
                             instance.primary_node)
1976

    
1977
  if instance.name in ins_l:
1978
    raise errors.OpExecError("Instance is running, can't shutdown"
1979
                             " block devices.")
1980

    
1981
  _ShutdownInstanceDisks(instance, cfg)
1982

    
1983

    
1984
def _ShutdownInstanceDisks(instance, cfg, ignore_primary=False):
1985
  """Shutdown block devices of an instance.
1986

1987
  This does the shutdown on all nodes of the instance.
1988

1989
  If the ignore_primary is false, errors on the primary node are
1990
  ignored.
1991

1992
  """
1993
  result = True
1994
  for disk in instance.disks:
1995
    for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
1996
      cfg.SetDiskID(top_disk, node)
1997
      if not rpc.call_blockdev_shutdown(node, top_disk):
1998
        logger.Error("could not shutdown block device %s on node %s" %
1999
                     (disk.iv_name, node))
2000
        if not ignore_primary or node != instance.primary_node:
2001
          result = False
2002
  return result
2003

    
2004

    
2005
def _CheckNodeFreeMemory(cfg, node, reason, requested):
2006
  """Checks if a node has enough free memory.
2007

2008
  This function check if a given node has the needed amount of free
2009
  memory. In case the node has less memory or we cannot get the
2010
  information from the node, this function raise an OpPrereqError
2011
  exception.
2012

2013
  Args:
2014
    - cfg: a ConfigWriter instance
2015
    - node: the node name
2016
    - reason: string to use in the error message
2017
    - requested: the amount of memory in MiB
2018

2019
  """
2020
  nodeinfo = rpc.call_node_info([node], cfg.GetVGName())
2021
  if not nodeinfo or not isinstance(nodeinfo, dict):
2022
    raise errors.OpPrereqError("Could not contact node %s for resource"
2023
                             " information" % (node,))
2024

    
2025
  free_mem = nodeinfo[node].get('memory_free')
2026
  if not isinstance(free_mem, int):
2027
    raise errors.OpPrereqError("Can't compute free memory on node %s, result"
2028
                             " was '%s'" % (node, free_mem))
2029
  if requested > free_mem:
2030
    raise errors.OpPrereqError("Not enough memory on node %s for %s:"
2031
                             " needed %s MiB, available %s MiB" %
2032
                             (node, reason, requested, free_mem))
2033

    
2034

    
2035
class LUStartupInstance(LogicalUnit):
2036
  """Starts an instance.
2037

2038
  """
2039
  HPATH = "instance-start"
2040
  HTYPE = constants.HTYPE_INSTANCE
2041
  _OP_REQP = ["instance_name", "force"]
2042
  REQ_BGL = False
2043

    
2044
  def ExpandNames(self):
2045
    self._ExpandAndLockInstance()
2046
    self.needed_locks[locking.LEVEL_NODE] = []
2047
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2048

    
2049
  def DeclareLocks(self, level):
2050
    if level == locking.LEVEL_NODE:
2051
      self._LockInstancesNodes()
2052

    
2053
  def BuildHooksEnv(self):
2054
    """Build hooks env.
2055

2056
    This runs on master, primary and secondary nodes of the instance.
2057

2058
    """
2059
    env = {
2060
      "FORCE": self.op.force,
2061
      }
2062
    env.update(_BuildInstanceHookEnvByObject(self.instance))
2063
    nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
2064
          list(self.instance.secondary_nodes))
2065
    return env, nl, nl
2066

    
2067
  def CheckPrereq(self):
2068
    """Check prerequisites.
2069

2070
    This checks that the instance is in the cluster.
2071

2072
    """
2073
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2074
    assert self.instance is not None, \
2075
      "Cannot retrieve locked instance %s" % self.op.instance_name
2076

    
2077
    # check bridges existance
2078
    _CheckInstanceBridgesExist(instance)
2079

    
2080
    _CheckNodeFreeMemory(self.cfg, instance.primary_node,
2081
                         "starting instance %s" % instance.name,
2082
                         instance.memory)
2083

    
2084
  def Exec(self, feedback_fn):
2085
    """Start the instance.
2086

2087
    """
2088
    instance = self.instance
2089
    force = self.op.force
2090
    extra_args = getattr(self.op, "extra_args", "")
2091

    
2092
    self.cfg.MarkInstanceUp(instance.name)
2093

    
2094
    node_current = instance.primary_node
2095

    
2096
    _StartInstanceDisks(self.cfg, instance, force)
2097

    
2098
    if not rpc.call_instance_start(node_current, instance, extra_args):
2099
      _ShutdownInstanceDisks(instance, self.cfg)
2100
      raise errors.OpExecError("Could not start instance")
2101

    
2102

    
2103
class LURebootInstance(LogicalUnit):
2104
  """Reboot an instance.
2105

2106
  """
2107
  HPATH = "instance-reboot"
2108
  HTYPE = constants.HTYPE_INSTANCE
2109
  _OP_REQP = ["instance_name", "ignore_secondaries", "reboot_type"]
2110
  REQ_BGL = False
2111

    
2112
  def ExpandNames(self):
2113
    if self.op.reboot_type not in [constants.INSTANCE_REBOOT_SOFT,
2114
                                   constants.INSTANCE_REBOOT_HARD,
2115
                                   constants.INSTANCE_REBOOT_FULL]:
2116
      raise errors.ParameterError("reboot type not in [%s, %s, %s]" %
2117
                                  (constants.INSTANCE_REBOOT_SOFT,
2118
                                   constants.INSTANCE_REBOOT_HARD,
2119
                                   constants.INSTANCE_REBOOT_FULL))
2120
    self._ExpandAndLockInstance()
2121
    self.needed_locks[locking.LEVEL_NODE] = []
2122
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2123

    
2124
  def DeclareLocks(self, level):
2125
    if level == locking.LEVEL_NODE:
2126
      primary_only = not constants.INSTANCE_REBOOT_FULL
2127
      self._LockInstancesNodes(primary_only=primary_only)
2128

    
2129
  def BuildHooksEnv(self):
2130
    """Build hooks env.
2131

2132
    This runs on master, primary and secondary nodes of the instance.
2133

2134
    """
2135
    env = {
2136
      "IGNORE_SECONDARIES": self.op.ignore_secondaries,
2137
      }
2138
    env.update(_BuildInstanceHookEnvByObject(self.instance))
2139
    nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
2140
          list(self.instance.secondary_nodes))
2141
    return env, nl, nl
2142

    
2143
  def CheckPrereq(self):
2144
    """Check prerequisites.
2145

2146
    This checks that the instance is in the cluster.
2147

2148
    """
2149
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2150
    assert self.instance is not None, \
2151
      "Cannot retrieve locked instance %s" % self.op.instance_name
2152

    
2153
    # check bridges existance
2154
    _CheckInstanceBridgesExist(instance)
2155

    
2156
  def Exec(self, feedback_fn):
2157
    """Reboot the instance.
2158

2159
    """
2160
    instance = self.instance
2161
    ignore_secondaries = self.op.ignore_secondaries
2162
    reboot_type = self.op.reboot_type
2163
    extra_args = getattr(self.op, "extra_args", "")
2164

    
2165
    node_current = instance.primary_node
2166

    
2167
    if reboot_type in [constants.INSTANCE_REBOOT_SOFT,
2168
                       constants.INSTANCE_REBOOT_HARD]:
2169
      if not rpc.call_instance_reboot(node_current, instance,
2170
                                      reboot_type, extra_args):
2171
        raise errors.OpExecError("Could not reboot instance")
2172
    else:
2173
      if not rpc.call_instance_shutdown(node_current, instance):
2174
        raise errors.OpExecError("could not shutdown instance for full reboot")
2175
      _ShutdownInstanceDisks(instance, self.cfg)
2176
      _StartInstanceDisks(self.cfg, instance, ignore_secondaries)
2177
      if not rpc.call_instance_start(node_current, instance, extra_args):
2178
        _ShutdownInstanceDisks(instance, self.cfg)
2179
        raise errors.OpExecError("Could not start instance for full reboot")
2180

    
2181
    self.cfg.MarkInstanceUp(instance.name)
2182

    
2183

    
2184
class LUShutdownInstance(LogicalUnit):
2185
  """Shutdown an instance.
2186

2187
  """
2188
  HPATH = "instance-stop"
2189
  HTYPE = constants.HTYPE_INSTANCE
2190
  _OP_REQP = ["instance_name"]
2191
  REQ_BGL = False
2192

    
2193
  def ExpandNames(self):
2194
    self._ExpandAndLockInstance()
2195
    self.needed_locks[locking.LEVEL_NODE] = []
2196
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2197

    
2198
  def DeclareLocks(self, level):
2199
    if level == locking.LEVEL_NODE:
2200
      self._LockInstancesNodes()
2201

    
2202
  def BuildHooksEnv(self):
2203
    """Build hooks env.
2204

2205
    This runs on master, primary and secondary nodes of the instance.
2206

2207
    """
2208
    env = _BuildInstanceHookEnvByObject(self.instance)
2209
    nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
2210
          list(self.instance.secondary_nodes))
2211
    return env, nl, nl
2212

    
2213
  def CheckPrereq(self):
2214
    """Check prerequisites.
2215

2216
    This checks that the instance is in the cluster.
2217

2218
    """
2219
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2220
    assert self.instance is not None, \
2221
      "Cannot retrieve locked instance %s" % self.op.instance_name
2222

    
2223
  def Exec(self, feedback_fn):
2224
    """Shutdown the instance.
2225

2226
    """
2227
    instance = self.instance
2228
    node_current = instance.primary_node
2229
    self.cfg.MarkInstanceDown(instance.name)
2230
    if not rpc.call_instance_shutdown(node_current, instance):
2231
      logger.Error("could not shutdown instance")
2232

    
2233
    _ShutdownInstanceDisks(instance, self.cfg)
2234

    
2235

    
2236
class LUReinstallInstance(LogicalUnit):
2237
  """Reinstall an instance.
2238

2239
  """
2240
  HPATH = "instance-reinstall"
2241
  HTYPE = constants.HTYPE_INSTANCE
2242
  _OP_REQP = ["instance_name"]
2243
  REQ_BGL = False
2244

    
2245
  def ExpandNames(self):
2246
    self._ExpandAndLockInstance()
2247
    self.needed_locks[locking.LEVEL_NODE] = []
2248
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2249

    
2250
  def DeclareLocks(self, level):
2251
    if level == locking.LEVEL_NODE:
2252
      self._LockInstancesNodes()
2253

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

2257
    This runs on master, primary and secondary nodes of the instance.
2258

2259
    """
2260
    env = _BuildInstanceHookEnvByObject(self.instance)
2261
    nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
2262
          list(self.instance.secondary_nodes))
2263
    return env, nl, nl
2264

    
2265
  def CheckPrereq(self):
2266
    """Check prerequisites.
2267

2268
    This checks that the instance is in the cluster and is not running.
2269

2270
    """
2271
    instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2272
    assert instance is not None, \
2273
      "Cannot retrieve locked instance %s" % self.op.instance_name
2274

    
2275
    if instance.disk_template == constants.DT_DISKLESS:
2276
      raise errors.OpPrereqError("Instance '%s' has no disks" %
2277
                                 self.op.instance_name)
2278
    if instance.status != "down":
2279
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2280
                                 self.op.instance_name)
2281
    remote_info = rpc.call_instance_info(instance.primary_node, instance.name)
2282
    if remote_info:
2283
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
2284
                                 (self.op.instance_name,
2285
                                  instance.primary_node))
2286

    
2287
    self.op.os_type = getattr(self.op, "os_type", None)
2288
    if self.op.os_type is not None:
2289
      # OS verification
2290
      pnode = self.cfg.GetNodeInfo(
2291
        self.cfg.ExpandNodeName(instance.primary_node))
2292
      if pnode is None:
2293
        raise errors.OpPrereqError("Primary node '%s' is unknown" %
2294
                                   self.op.pnode)
2295
      os_obj = rpc.call_os_get(pnode.name, self.op.os_type)
2296
      if not os_obj:
2297
        raise errors.OpPrereqError("OS '%s' not in supported OS list for"
2298
                                   " primary node"  % self.op.os_type)
2299

    
2300
    self.instance = instance
2301

    
2302
  def Exec(self, feedback_fn):
2303
    """Reinstall the instance.
2304

2305
    """
2306
    inst = self.instance
2307

    
2308
    if self.op.os_type is not None:
2309
      feedback_fn("Changing OS to '%s'..." % self.op.os_type)
2310
      inst.os = self.op.os_type
2311
      self.cfg.AddInstance(inst)
2312

    
2313
    _StartInstanceDisks(self.cfg, inst, None)
2314
    try:
2315
      feedback_fn("Running the instance OS create scripts...")
2316
      if not rpc.call_instance_os_add(inst.primary_node, inst, "sda", "sdb"):
2317
        raise errors.OpExecError("Could not install OS for instance %s"
2318
                                 " on node %s" %
2319
                                 (inst.name, inst.primary_node))
2320
    finally:
2321
      _ShutdownInstanceDisks(inst, self.cfg)
2322

    
2323

    
2324
class LURenameInstance(LogicalUnit):
2325
  """Rename an instance.
2326

2327
  """
2328
  HPATH = "instance-rename"
2329
  HTYPE = constants.HTYPE_INSTANCE
2330
  _OP_REQP = ["instance_name", "new_name"]
2331

    
2332
  def BuildHooksEnv(self):
2333
    """Build hooks env.
2334

2335
    This runs on master, primary and secondary nodes of the instance.
2336

2337
    """
2338
    env = _BuildInstanceHookEnvByObject(self.instance)
2339
    env["INSTANCE_NEW_NAME"] = self.op.new_name
2340
    nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
2341
          list(self.instance.secondary_nodes))
2342
    return env, nl, nl
2343

    
2344
  def CheckPrereq(self):
2345
    """Check prerequisites.
2346

2347
    This checks that the instance is in the cluster and is not running.
2348

2349
    """
2350
    instance = self.cfg.GetInstanceInfo(
2351
      self.cfg.ExpandInstanceName(self.op.instance_name))
2352
    if instance is None:
2353
      raise errors.OpPrereqError("Instance '%s' not known" %
2354
                                 self.op.instance_name)
2355
    if instance.status != "down":
2356
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2357
                                 self.op.instance_name)
2358
    remote_info = rpc.call_instance_info(instance.primary_node, instance.name)
2359
    if remote_info:
2360
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
2361
                                 (self.op.instance_name,
2362
                                  instance.primary_node))
2363
    self.instance = instance
2364

    
2365
    # new name verification
2366
    name_info = utils.HostInfo(self.op.new_name)
2367

    
2368
    self.op.new_name = new_name = name_info.name
2369
    instance_list = self.cfg.GetInstanceList()
2370
    if new_name in instance_list:
2371
      raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
2372
                                 new_name)
2373

    
2374
    if not getattr(self.op, "ignore_ip", False):
2375
      if utils.TcpPing(name_info.ip, constants.DEFAULT_NODED_PORT):
2376
        raise errors.OpPrereqError("IP %s of instance %s already in use" %
2377
                                   (name_info.ip, new_name))
2378

    
2379

    
2380
  def Exec(self, feedback_fn):
2381
    """Reinstall the instance.
2382

2383
    """
2384
    inst = self.instance
2385
    old_name = inst.name
2386

    
2387
    if inst.disk_template == constants.DT_FILE:
2388
      old_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
2389

    
2390
    self.cfg.RenameInstance(inst.name, self.op.new_name)
2391
    # Change the instance lock. This is definitely safe while we hold the BGL
2392
    self.context.glm.remove(locking.LEVEL_INSTANCE, inst.name)
2393
    self.context.glm.add(locking.LEVEL_INSTANCE, self.op.new_name)
2394

    
2395
    # re-read the instance from the configuration after rename
2396
    inst = self.cfg.GetInstanceInfo(self.op.new_name)
2397

    
2398
    if inst.disk_template == constants.DT_FILE:
2399
      new_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
2400
      result = rpc.call_file_storage_dir_rename(inst.primary_node,
2401
                                                old_file_storage_dir,
2402
                                                new_file_storage_dir)
2403

    
2404
      if not result:
2405
        raise errors.OpExecError("Could not connect to node '%s' to rename"
2406
                                 " directory '%s' to '%s' (but the instance"
2407
                                 " has been renamed in Ganeti)" % (
2408
                                 inst.primary_node, old_file_storage_dir,
2409
                                 new_file_storage_dir))
2410

    
2411
      if not result[0]:
2412
        raise errors.OpExecError("Could not rename directory '%s' to '%s'"
2413
                                 " (but the instance has been renamed in"
2414
                                 " Ganeti)" % (old_file_storage_dir,
2415
                                               new_file_storage_dir))
2416

    
2417
    _StartInstanceDisks(self.cfg, inst, None)
2418
    try:
2419
      if not rpc.call_instance_run_rename(inst.primary_node, inst, old_name,
2420
                                          "sda", "sdb"):
2421
        msg = ("Could not run OS rename script for instance %s on node %s"
2422
               " (but the instance has been renamed in Ganeti)" %
2423
               (inst.name, inst.primary_node))
2424
        logger.Error(msg)
2425
    finally:
2426
      _ShutdownInstanceDisks(inst, self.cfg)
2427

    
2428

    
2429
class LURemoveInstance(LogicalUnit):
2430
  """Remove an instance.
2431

2432
  """
2433
  HPATH = "instance-remove"
2434
  HTYPE = constants.HTYPE_INSTANCE
2435
  _OP_REQP = ["instance_name", "ignore_failures"]
2436

    
2437
  def BuildHooksEnv(self):
2438
    """Build hooks env.
2439

2440
    This runs on master, primary and secondary nodes of the instance.
2441

2442
    """
2443
    env = _BuildInstanceHookEnvByObject(self.instance)
2444
    nl = [self.sstore.GetMasterNode()]
2445
    return env, nl, nl
2446

    
2447
  def CheckPrereq(self):
2448
    """Check prerequisites.
2449

2450
    This checks that the instance is in the cluster.
2451

2452
    """
2453
    instance = self.cfg.GetInstanceInfo(
2454
      self.cfg.ExpandInstanceName(self.op.instance_name))
2455
    if instance is None:
2456
      raise errors.OpPrereqError("Instance '%s' not known" %
2457
                                 self.op.instance_name)
2458
    self.instance = instance
2459

    
2460
  def Exec(self, feedback_fn):
2461
    """Remove the instance.
2462

2463
    """
2464
    instance = self.instance
2465
    logger.Info("shutting down instance %s on node %s" %
2466
                (instance.name, instance.primary_node))
2467

    
2468
    if not rpc.call_instance_shutdown(instance.primary_node, instance):
2469
      if self.op.ignore_failures:
2470
        feedback_fn("Warning: can't shutdown instance")
2471
      else:
2472
        raise errors.OpExecError("Could not shutdown instance %s on node %s" %
2473
                                 (instance.name, instance.primary_node))
2474

    
2475
    logger.Info("removing block devices for instance %s" % instance.name)
2476

    
2477
    if not _RemoveDisks(instance, self.cfg):
2478
      if self.op.ignore_failures:
2479
        feedback_fn("Warning: can't remove instance's disks")
2480
      else:
2481
        raise errors.OpExecError("Can't remove instance's disks")
2482

    
2483
    logger.Info("removing instance %s out of cluster config" % instance.name)
2484

    
2485
    self.cfg.RemoveInstance(instance.name)
2486
    # Remove the new instance from the Ganeti Lock Manager
2487
    self.context.glm.remove(locking.LEVEL_INSTANCE, instance.name)
2488

    
2489

    
2490
class LUQueryInstances(NoHooksLU):
2491
  """Logical unit for querying instances.
2492

2493
  """
2494
  _OP_REQP = ["output_fields", "names"]
2495
  REQ_BGL = False
2496

    
2497
  def ExpandNames(self):
2498
    self.dynamic_fields = frozenset(["oper_state", "oper_ram", "status"])
2499
    _CheckOutputFields(static=["name", "os", "pnode", "snodes",
2500
                               "admin_state", "admin_ram",
2501
                               "disk_template", "ip", "mac", "bridge",
2502
                               "sda_size", "sdb_size", "vcpus", "tags",
2503
                               "auto_balance",
2504
                               "network_port", "kernel_path", "initrd_path",
2505
                               "hvm_boot_order", "hvm_acpi", "hvm_pae",
2506
                               "hvm_cdrom_image_path", "hvm_nic_type",
2507
                               "hvm_disk_type", "vnc_bind_address"],
2508
                       dynamic=self.dynamic_fields,
2509
                       selected=self.op.output_fields)
2510

    
2511
    self.needed_locks = {}
2512
    self.share_locks[locking.LEVEL_INSTANCE] = 1
2513
    self.share_locks[locking.LEVEL_NODE] = 1
2514

    
2515
    # TODO: we could lock instances (and nodes) only if the user asked for
2516
    # dynamic fields. For that we need atomic ways to get info for a group of
2517
    # instances from the config, though.
2518
    if not self.op.names:
2519
      self.needed_locks[locking.LEVEL_INSTANCE] = locking.ALL_SET
2520
    else:
2521
      self.needed_locks[locking.LEVEL_INSTANCE] = \
2522
        _GetWantedInstances(self, self.op.names)
2523

    
2524
    self.needed_locks[locking.LEVEL_NODE] = []
2525
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2526

    
2527
  def DeclareLocks(self, level):
2528
    # TODO: locking of nodes could be avoided when not querying them
2529
    if level == locking.LEVEL_NODE:
2530
      self._LockInstancesNodes()
2531

    
2532
  def CheckPrereq(self):
2533
    """Check prerequisites.
2534

2535
    """
2536
    # This of course is valid only if we locked the instances
2537
    self.wanted = self.acquired_locks[locking.LEVEL_INSTANCE]
2538

    
2539
  def Exec(self, feedback_fn):
2540
    """Computes the list of nodes and their attributes.
2541

2542
    """
2543
    instance_names = self.wanted
2544
    instance_list = [self.cfg.GetInstanceInfo(iname) for iname
2545
                     in instance_names]
2546

    
2547
    # begin data gathering
2548

    
2549
    nodes = frozenset([inst.primary_node for inst in instance_list])
2550

    
2551
    bad_nodes = []
2552
    if self.dynamic_fields.intersection(self.op.output_fields):
2553
      live_data = {}
2554
      node_data = rpc.call_all_instances_info(nodes)
2555
      for name in nodes:
2556
        result = node_data[name]
2557
        if result:
2558
          live_data.update(result)
2559
        elif result == False:
2560
          bad_nodes.append(name)
2561
        # else no instance is alive
2562
    else:
2563
      live_data = dict([(name, {}) for name in instance_names])
2564

    
2565
    # end data gathering
2566

    
2567
    output = []
2568
    for instance in instance_list:
2569
      iout = []
2570
      for field in self.op.output_fields:
2571
        if field == "name":
2572
          val = instance.name
2573
        elif field == "os":
2574
          val = instance.os
2575
        elif field == "pnode":
2576
          val = instance.primary_node
2577
        elif field == "snodes":
2578
          val = list(instance.secondary_nodes)
2579
        elif field == "admin_state":
2580
          val = (instance.status != "down")
2581
        elif field == "oper_state":
2582
          if instance.primary_node in bad_nodes:
2583
            val = None
2584
          else:
2585
            val = bool(live_data.get(instance.name))
2586
        elif field == "status":
2587
          if instance.primary_node in bad_nodes:
2588
            val = "ERROR_nodedown"
2589
          else:
2590
            running = bool(live_data.get(instance.name))
2591
            if running:
2592
              if instance.status != "down":
2593
                val = "running"
2594
              else:
2595
                val = "ERROR_up"
2596
            else:
2597
              if instance.status != "down":
2598
                val = "ERROR_down"
2599
              else:
2600
                val = "ADMIN_down"
2601
        elif field == "admin_ram":
2602
          val = instance.memory
2603
        elif field == "oper_ram":
2604
          if instance.primary_node in bad_nodes:
2605
            val = None
2606
          elif instance.name in live_data:
2607
            val = live_data[instance.name].get("memory", "?")
2608
          else:
2609
            val = "-"
2610
        elif field == "disk_template":
2611
          val = instance.disk_template
2612
        elif field == "ip":
2613
          val = instance.nics[0].ip
2614
        elif field == "bridge":
2615
          val = instance.nics[0].bridge
2616
        elif field == "mac":
2617
          val = instance.nics[0].mac
2618
        elif field == "sda_size" or field == "sdb_size":
2619
          disk = instance.FindDisk(field[:3])
2620
          if disk is None:
2621
            val = None
2622
          else:
2623
            val = disk.size
2624
        elif field == "vcpus":
2625
          val = instance.vcpus
2626
        elif field == "tags":
2627
          val = list(instance.GetTags())
2628
        elif field in ("network_port", "kernel_path", "initrd_path",
2629
                       "hvm_boot_order", "hvm_acpi", "hvm_pae",
2630
                       "hvm_cdrom_image_path", "hvm_nic_type",
2631
                       "hvm_disk_type", "vnc_bind_address"):
2632
          val = getattr(instance, field, None)
2633
          if val is not None:
2634
            pass
2635
          elif field in ("hvm_nic_type", "hvm_disk_type",
2636
                         "kernel_path", "initrd_path"):
2637
            val = "default"
2638
          else:
2639
            val = "-"
2640
        else:
2641
          raise errors.ParameterError(field)
2642
        iout.append(val)
2643
      output.append(iout)
2644

    
2645
    return output
2646

    
2647

    
2648
class LUFailoverInstance(LogicalUnit):
2649
  """Failover an instance.
2650

2651
  """
2652
  HPATH = "instance-failover"
2653
  HTYPE = constants.HTYPE_INSTANCE
2654
  _OP_REQP = ["instance_name", "ignore_consistency"]
2655
  REQ_BGL = False
2656

    
2657
  def ExpandNames(self):
2658
    self._ExpandAndLockInstance()
2659
    self.needed_locks[locking.LEVEL_NODE] = []
2660
    self.recalculate_locks[locking.LEVEL_NODE] = constants.LOCKS_REPLACE
2661

    
2662
  def DeclareLocks(self, level):
2663
    if level == locking.LEVEL_NODE:
2664
      self._LockInstancesNodes()
2665

    
2666
  def BuildHooksEnv(self):
2667
    """Build hooks env.
2668

2669
    This runs on master, primary and secondary nodes of the instance.
2670

2671
    """
2672
    env = {
2673
      "IGNORE_CONSISTENCY": self.op.ignore_consistency,
2674
      }
2675
    env.update(_BuildInstanceHookEnvByObject(self.instance))
2676
    nl = [self.sstore.GetMasterNode()] + list(self.instance.secondary_nodes)
2677
    return env, nl, nl
2678

    
2679
  def CheckPrereq(self):
2680
    """Check prerequisites.
2681

2682
    This checks that the instance is in the cluster.
2683

2684
    """
2685
    self.instance = instance = self.cfg.GetInstanceInfo(self.op.instance_name)
2686
    assert self.instance is not None, \
2687
      "Cannot retrieve locked instance %s" % self.op.instance_name
2688

    
2689
    if instance.disk_template not in constants.DTS_NET_MIRROR:
2690
      raise errors.OpPrereqError("Instance's disk layout is not"
2691
                                 " network mirrored, cannot failover.")
2692

    
2693
    secondary_nodes = instance.secondary_nodes
2694
    if not secondary_nodes:
2695
      raise errors.ProgrammerError("no secondary node but using "
2696
                                   "a mirrored disk template")
2697

    
2698
    target_node = secondary_nodes[0]
2699
    # check memory requirements on the secondary node
2700
    _CheckNodeFreeMemory(self.cfg, target_node, "failing over instance %s" %
2701
                         instance.name, instance.memory)
2702

    
2703
    # check bridge existance
2704
    brlist = [nic.bridge for nic in instance.nics]
2705
    if not rpc.call_bridges_exist(target_node, brlist):
2706
      raise errors.OpPrereqError("One or more target bridges %s does not"
2707
                                 " exist on destination node '%s'" %
2708
                                 (brlist, target_node))
2709

    
2710
  def Exec(self, feedback_fn):
2711
    """Failover an instance.
2712

2713
    The failover is done by shutting it down on its present node and
2714
    starting it on the secondary.
2715

2716
    """
2717
    instance = self.instance
2718

    
2719
    source_node = instance.primary_node
2720
    target_node = instance.secondary_nodes[0]
2721

    
2722
    feedback_fn("* checking disk consistency between source and target")
2723
    for dev in instance.disks:
2724
      # for drbd, these are drbd over lvm
2725
      if not _CheckDiskConsistency(self.cfg, dev, target_node, False):
2726
        if instance.status == "up" and not self.op.ignore_consistency:
2727
          raise errors.OpExecError("Disk %s is degraded on target node,"
2728
                                   " aborting failover." % dev.iv_name)
2729

    
2730
    feedback_fn("* shutting down instance on source node")
2731
    logger.Info("Shutting down instance %s on node %s" %
2732
                (instance.name, source_node))
2733

    
2734
    if not rpc.call_instance_shutdown(source_node, instance):
2735
      if self.op.ignore_consistency:
2736
        logger.Error("Could not shutdown instance %s on node %s. Proceeding"
2737
                     " anyway. Please make sure node %s is down"  %
2738
                     (instance.name, source_node, source_node))
2739
      else:
2740
        raise errors.OpExecError("Could not shutdown instance %s on node %s" %
2741
                                 (instance.name, source_node))
2742

    
2743
    feedback_fn("* deactivating the instance's disks on source node")
2744
    if not _ShutdownInstanceDisks(instance, self.cfg, ignore_primary=True):
2745
      raise errors.OpExecError("Can't shut down the instance's disks.")
2746

    
2747
    instance.primary_node = target_node
2748
    # distribute new instance config to the other nodes
2749
    self.cfg.Update(instance)
2750

    
2751
    # Only start the instance if it's marked as up
2752
    if instance.status == "up":
2753
      feedback_fn("* activating the instance's disks on target node")
2754
      logger.Info("Starting instance %s on node %s" %
2755
                  (instance.name, target_node))
2756

    
2757
      disks_ok, dummy = _AssembleInstanceDisks(instance, self.cfg,
2758
                                               ignore_secondaries=True)
2759
      if not disks_ok:
2760
        _ShutdownInstanceDisks(instance, self.cfg)
2761
        raise errors.OpExecError("Can't activate the instance's disks")
2762

    
2763
      feedback_fn("* starting the instance on the target node")
2764
      if not rpc.call_instance_start(target_node, instance, None):
2765
        _ShutdownInstanceDisks(instance, self.cfg)
2766
        raise errors.OpExecError("Could not start instance %s on node %s." %
2767
                                 (instance.name, target_node))
2768

    
2769

    
2770
def _CreateBlockDevOnPrimary(cfg, node, instance, device, info):
2771
  """Create a tree of block devices on the primary node.
2772

2773
  This always creates all devices.
2774

2775
  """
2776
  if device.children:
2777
    for child in device.children:
2778
      if not _CreateBlockDevOnPrimary(cfg, node, instance, child, info):
2779
        return False
2780

    
2781
  cfg.SetDiskID(device, node)
2782
  new_id = rpc.call_blockdev_create(node, device, device.size,
2783
                                    instance.name, True, info)
2784
  if not new_id:
2785
    return False
2786
  if device.physical_id is None:
2787
    device.physical_id = new_id
2788
  return True
2789

    
2790

    
2791
def _CreateBlockDevOnSecondary(cfg, node, instance, device, force, info):
2792
  """Create a tree of block devices on a secondary node.
2793

2794
  If this device type has to be created on secondaries, create it and
2795
  all its children.
2796

2797
  If not, just recurse to children keeping the same 'force' value.
2798

2799
  """
2800
  if device.CreateOnSecondary():
2801
    force = True
2802
  if device.children:
2803
    for child in device.children:
2804
      if not _CreateBlockDevOnSecondary(cfg, node, instance,
2805
                                        child, force, info):
2806
        return False
2807

    
2808
  if not force:
2809
    return True
2810
  cfg.SetDiskID(device, node)
2811
  new_id = rpc.call_blockdev_create(node, device, device.size,
2812
                                    instance.name, False, info)
2813
  if not new_id:
2814
    return False
2815
  if device.physical_id is None:
2816
    device.physical_id = new_id
2817
  return True
2818

    
2819

    
2820
def _GenerateUniqueNames(cfg, exts):
2821
  """Generate a suitable LV name.
2822

2823
  This will generate a logical volume name for the given instance.
2824

2825
  """
2826
  results = []
2827
  for val in exts:
2828
    new_id = cfg.GenerateUniqueID()
2829
    results.append("%s%s" % (new_id, val))
2830
  return results
2831

    
2832

    
2833
def _GenerateDRBD8Branch(cfg, primary, secondary, size, names, iv_name):
2834
  """Generate a drbd8 device complete with its children.
2835

2836
  """
2837
  port = cfg.AllocatePort()
2838
  vgname = cfg.GetVGName()
2839
  dev_data = objects.Disk(dev_type=constants.LD_LV, size=size,
2840
                          logical_id=(vgname, names[0]))
2841
  dev_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
2842
                          logical_id=(vgname, names[1]))
2843
  drbd_dev = objects.Disk(dev_type=constants.LD_DRBD8, size=size,
2844
                          logical_id = (primary, secondary, port),
2845
                          children = [dev_data, dev_meta],
2846
                          iv_name=iv_name)
2847
  return drbd_dev
2848

    
2849

    
2850
def _GenerateDiskTemplate(cfg, template_name,
2851
                          instance_name, primary_node,
2852
                          secondary_nodes, disk_sz, swap_sz,
2853
                          file_storage_dir, file_driver):
2854
  """Generate the entire disk layout for a given template type.
2855

2856
  """
2857
  #TODO: compute space requirements
2858

    
2859
  vgname = cfg.GetVGName()
2860
  if template_name == constants.DT_DISKLESS:
2861
    disks = []
2862
  elif template_name == constants.DT_PLAIN:
2863
    if len(secondary_nodes) != 0:
2864
      raise errors.ProgrammerError("Wrong template configuration")
2865

    
2866
    names = _GenerateUniqueNames(cfg, [".sda", ".sdb"])
2867
    sda_dev = objects.Disk(dev_type=constants.LD_LV, size=disk_sz,
2868
                           logical_id=(vgname, names[0]),
2869
                           iv_name = "sda")
2870
    sdb_dev = objects.Disk(dev_type=constants.LD_LV, size=swap_sz,
2871
                           logical_id=(vgname, names[1]),
2872
                           iv_name = "sdb")
2873
    disks = [sda_dev, sdb_dev]
2874
  elif template_name == constants.DT_DRBD8:
2875
    if len(secondary_nodes) != 1:
2876
      raise errors.ProgrammerError("Wrong template configuration")
2877
    remote_node = secondary_nodes[0]
2878
    names = _GenerateUniqueNames(cfg, [".sda_data", ".sda_meta",
2879
                                       ".sdb_data", ".sdb_meta"])
2880
    drbd_sda_dev = _GenerateDRBD8Branch(cfg, primary_node, remote_node,
2881
                                         disk_sz, names[0:2], "sda")
2882
    drbd_sdb_dev = _GenerateDRBD8Branch(cfg, primary_node, remote_node,
2883
                                         swap_sz, names[2:4], "sdb")
2884
    disks = [drbd_sda_dev, drbd_sdb_dev]
2885
  elif template_name == constants.DT_FILE:
2886
    if len(secondary_nodes) != 0:
2887
      raise errors.ProgrammerError("Wrong template configuration")
2888

    
2889
    file_sda_dev = objects.Disk(dev_type=constants.LD_FILE, size=disk_sz,
2890
                                iv_name="sda", logical_id=(file_driver,
2891
                                "%s/sda" % file_storage_dir))
2892
    file_sdb_dev = objects.Disk(dev_type=constants.LD_FILE, size=swap_sz,
2893
                                iv_name="sdb", logical_id=(file_driver,
2894
                                "%s/sdb" % file_storage_dir))
2895
    disks = [file_sda_dev, file_sdb_dev]
2896
  else:
2897
    raise errors.ProgrammerError("Invalid disk template '%s'" % template_name)
2898
  return disks
2899

    
2900

    
2901
def _GetInstanceInfoText(instance):
2902
  """Compute that text that should be added to the disk's metadata.
2903

2904
  """
2905
  return "originstname+%s" % instance.name
2906

    
2907

    
2908
def _CreateDisks(cfg, instance):
2909
  """Create all disks for an instance.
2910

2911
  This abstracts away some work from AddInstance.
2912

2913
  Args:
2914
    instance: the instance object
2915

2916
  Returns:
2917
    True or False showing the success of the creation process
2918

2919
  """
2920
  info = _GetInstanceInfoText(instance)
2921

    
2922
  if instance.disk_template == constants.DT_FILE:
2923
    file_storage_dir = os.path.dirname(instance.disks[0].logical_id[1])
2924
    result = rpc.call_file_storage_dir_create(instance.primary_node,
2925
                                              file_storage_dir)
2926

    
2927
    if not result:
2928
      logger.Error("Could not connect to node '%s'" % instance.primary_node)
2929
      return False
2930

    
2931
    if not result[0]:
2932
      logger.Error("failed to create directory '%s'" % file_storage_dir)
2933
      return False
2934

    
2935
  for device in instance.disks:
2936
    logger.Info("creating volume %s for instance %s" %
2937
                (device.iv_name, instance.name))
2938
    #HARDCODE
2939
    for secondary_node in instance.secondary_nodes:
2940
      if not _CreateBlockDevOnSecondary(cfg, secondary_node, instance,
2941
                                        device, False, info):
2942
        logger.Error("failed to create volume %s (%s) on secondary node %s!" %
2943
                     (device.iv_name, device, secondary_node))
2944
        return False
2945
    #HARDCODE
2946
    if not _CreateBlockDevOnPrimary(cfg, instance.primary_node,
2947
                                    instance, device, info):
2948
      logger.Error("failed to create volume %s on primary!" %
2949
                   device.iv_name)
2950
      return False
2951

    
2952
  return True
2953

    
2954

    
2955
def _RemoveDisks(instance, cfg):
2956
  """Remove all disks for an instance.
2957

2958
  This abstracts away some work from `AddInstance()` and
2959
  `RemoveInstance()`. Note that in case some of the devices couldn't
2960
  be removed, the removal will continue with the other ones (compare
2961
  with `_CreateDisks()`).
2962

2963
  Args:
2964
    instance: the instance object
2965

2966
  Returns:
2967
    True or False showing the success of the removal proces
2968

2969
  """
2970
  logger.Info("removing block devices for instance %s" % instance.name)
2971

    
2972
  result = True
2973
  for device in instance.disks:
2974
    for node, disk in device.ComputeNodeTree(instance.primary_node):
2975
      cfg.SetDiskID(disk, node)
2976
      if not rpc.call_blockdev_remove(node, disk):
2977
        logger.Error("could not remove block device %s on node %s,"
2978
                     " continuing anyway" %
2979
                     (device.iv_name, node))
2980
        result = False
2981

    
2982
  if instance.disk_template == constants.DT_FILE:
2983
    file_storage_dir = os.path.dirname(instance.disks[0].logical_id[1])
2984
    if not rpc.call_file_storage_dir_remove(instance.primary_node,
2985
                                            file_storage_dir):
2986
      logger.Error("could not remove directory '%s'" % file_storage_dir)
2987
      result = False
2988

    
2989
  return result
2990

    
2991

    
2992
def _ComputeDiskSize(disk_template, disk_size, swap_size):
2993
  """Compute disk size requirements in the volume group
2994

2995
  This is currently hard-coded for the two-drive layout.
2996

2997
  """
2998
  # Required free disk space as a function of disk and swap space
2999
  req_size_dict = {
3000
    constants.DT_DISKLESS: None,
3001
    constants.DT_PLAIN: disk_size + swap_size,
3002
    # 256 MB are added for drbd metadata, 128MB for each drbd device
3003
    constants.DT_DRBD8: disk_size + swap_size + 256,
3004
    constants.DT_FILE: None,
3005
  }
3006

    
3007
  if disk_template not in req_size_dict:
3008
    raise errors.ProgrammerError("Disk template '%s' size requirement"
3009
                                 " is unknown" %  disk_template)
3010

    
3011
  return req_size_dict[disk_template]
3012

    
3013

    
3014
class LUCreateInstance(LogicalUnit):
3015
  """Create an instance.
3016

3017
  """
3018
  HPATH = "instance-add"
3019
  HTYPE = constants.HTYPE_INSTANCE
3020
  _OP_REQP = ["instance_name", "mem_size", "disk_size",
3021
              "disk_template", "swap_size", "mode", "start", "vcpus",
3022
              "wait_for_sync", "ip_check", "mac"]
3023

    
3024
  def _RunAllocator(self):
3025
    """Run the allocator based on input opcode.
3026

3027
    """
3028
    disks = [{"size": self.op.disk_size, "mode": "w"},
3029
             {"size": self.op.swap_size, "mode": "w"}]
3030
    nics = [{"mac": self.op.mac, "ip": getattr(self.op, "ip", None),
3031
             "bridge": self.op.bridge}]
3032
    ial = IAllocator(self.cfg, self.sstore,
3033
                     mode=constants.IALLOCATOR_MODE_ALLOC,
3034
                     name=self.op.instance_name,
3035
                     disk_template=self.op.disk_template,
3036
                     tags=[],
3037
                     os=self.op.os_type,
3038
                     vcpus=self.op.vcpus,
3039
                     mem_size=self.op.mem_size,
3040
                     disks=disks,
3041
                     nics=nics,
3042
                     )
3043

    
3044
    ial.Run(self.op.iallocator)
3045

    
3046
    if not ial.success:
3047
      raise errors.OpPrereqError("Can't compute nodes using"
3048
                                 " iallocator '%s': %s" % (self.op.iallocator,
3049
                                                           ial.info))
3050
    if len(ial.nodes) != ial.required_nodes:
3051
      raise errors.OpPrereqError("iallocator '%s' returned invalid number"
3052
                                 " of nodes (%s), required %s" %
3053
                                 (len(ial.nodes), ial.required_nodes))
3054
    self.op.pnode = ial.nodes[0]
3055
    logger.ToStdout("Selected nodes for the instance: %s" %
3056
                    (", ".join(ial.nodes),))
3057
    logger.Info("Selected nodes for instance %s via iallocator %s: %s" %
3058
                (self.op.instance_name, self.op.iallocator, ial.nodes))
3059
    if ial.required_nodes == 2:
3060
      self.op.snode = ial.nodes[1]
3061

    
3062
  def BuildHooksEnv(self):
3063
    """Build hooks env.
3064

3065
    This runs on master, primary and secondary nodes of the instance.
3066

3067
    """
3068
    env = {
3069
      "INSTANCE_DISK_TEMPLATE": self.op.disk_template,
3070
      "INSTANCE_DISK_SIZE": self.op.disk_size,
3071
      "INSTANCE_SWAP_SIZE": self.op.swap_size,
3072
      "INSTANCE_ADD_MODE": self.op.mode,
3073
      }
3074
    if self.op.mode == constants.INSTANCE_IMPORT:
3075
      env["INSTANCE_SRC_NODE"] = self.op.src_node
3076
      env["INSTANCE_SRC_PATH"] = self.op.src_path
3077
      env["INSTANCE_SRC_IMAGE"] = self.src_image
3078

    
3079
    env.update(_BuildInstanceHookEnv(name=self.op.instance_name,
3080
      primary_node=self.op.pnode,
3081
      secondary_nodes=self.secondaries,
3082
      status=self.instance_status,
3083
      os_type=self.op.os_type,
3084
      memory=self.op.mem_size,
3085
      vcpus=self.op.vcpus,
3086
      nics=[(self.inst_ip, self.op.bridge, self.op.mac)],
3087
    ))
3088

    
3089
    nl = ([self.sstore.GetMasterNode(), self.op.pnode] +
3090
          self.secondaries)
3091
    return env, nl, nl
3092

    
3093

    
3094
  def CheckPrereq(self):
3095
    """Check prerequisites.
3096

3097
    """
3098
    # set optional parameters to none if they don't exist
3099
    for attr in ["kernel_path", "initrd_path", "hvm_boot_order", "pnode",
3100
                 "iallocator", "hvm_acpi", "hvm_pae", "hvm_cdrom_image_path",
3101
                 "hvm_nic_type", "hvm_disk_type", "vnc_bind_address"]:
3102
      if not hasattr(self.op, attr):
3103
        setattr(self.op, attr, None)
3104

    
3105
    if self.op.mode not in (constants.INSTANCE_CREATE,
3106
                            constants.INSTANCE_IMPORT):
3107
      raise errors.OpPrereqError("Invalid instance creation mode '%s'" %
3108
                                 self.op.mode)
3109

    
3110
    if (not self.cfg.GetVGName() and
3111
        self.op.disk_template not in constants.DTS_NOT_LVM):
3112
      raise errors.OpPrereqError("Cluster does not support lvm-based"
3113
                                 " instances")
3114

    
3115
    if self.op.mode == constants.INSTANCE_IMPORT:
3116
      src_node = getattr(self.op, "src_node", None)
3117
      src_path = getattr(self.op, "src_path", None)
3118
      if src_node is None or src_path is None:
3119
        raise errors.OpPrereqError("Importing an instance requires source"
3120
                                   " node and path options")
3121
      src_node_full = self.cfg.ExpandNodeName(src_node)
3122
      if src_node_full is None:
3123
        raise errors.OpPrereqError("Unknown source node '%s'" % src_node)
3124
      self.op.src_node = src_node = src_node_full
3125

    
3126
      if not os.path.isabs(src_path):
3127
        raise errors.OpPrereqError("The source path must be absolute")
3128

    
3129
      export_info = rpc.call_export_info(src_node, src_path)
3130

    
3131
      if not export_info:
3132
        raise errors.OpPrereqError("No export found in dir %s" % src_path)
3133

    
3134
      if not export_info.has_section(constants.INISECT_EXP):
3135
        raise errors.ProgrammerError("Corrupted export config")
3136

    
3137
      ei_version = export_info.get(constants.INISECT_EXP, 'version')
3138
      if (int(ei_version) != constants.EXPORT_VERSION):
3139
        raise errors.OpPrereqError("Wrong export version %s (wanted %d)" %
3140
                                   (ei_version, constants.EXPORT_VERSION))
3141

    
3142
      if int(export_info.get(constants.INISECT_INS, 'disk_count')) > 1:
3143
        raise errors.OpPrereqError("Can't import instance with more than"
3144
                                   " one data disk")
3145

    
3146
      # FIXME: are the old os-es, disk sizes, etc. useful?
3147
      self.op.os_type = export_info.get(constants.INISECT_EXP, 'os')
3148
      diskimage = os.path.join(src_path, export_info.get(constants.INISECT_INS,
3149
                                                         'disk0_dump'))
3150
      self.src_image = diskimage
3151
    else: # INSTANCE_CREATE
3152
      if getattr(self.op, "os_type", None) is None:
3153
        raise errors.OpPrereqError("No guest OS specified")
3154

    
3155
    #### instance parameters check
3156

    
3157
    # disk template and mirror node verification
3158
    if self.op.disk_template not in constants.DISK_TEMPLATES:
3159
      raise errors.OpPrereqError("Invalid disk template name")
3160

    
3161
    # instance name verification
3162
    hostname1 = utils.HostInfo(self.op.instance_name)
3163

    
3164
    self.op.instance_name = instance_name = hostname1.name
3165
    instance_list = self.cfg.GetInstanceList()
3166
    if instance_name in instance_list:
3167
      raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
3168
                                 instance_name)
3169

    
3170
    # ip validity checks
3171
    ip = getattr(self.op, "ip", None)
3172
    if ip is None or ip.lower() == "none":
3173
      inst_ip = None
3174
    elif ip.lower() == "auto":
3175
      inst_ip = hostname1.ip
3176
    else:
3177
      if not utils.IsValidIP(ip):
3178
        raise errors.OpPrereqError("given IP address '%s' doesn't look"
3179
                                   " like a valid IP" % ip)
3180
      inst_ip = ip
3181
    self.inst_ip = self.op.ip = inst_ip
3182

    
3183
    if self.op.start and not self.op.ip_check:
3184
      raise errors.OpPrereqError("Cannot ignore IP address conflicts when"
3185
                                 " adding an instance in start mode")
3186

    
3187
    if self.op.ip_check:
3188
      if utils.TcpPing(hostname1.ip, constants.DEFAULT_NODED_PORT):
3189
        raise errors.OpPrereqError("IP %s of instance %s already in use" %
3190
                                   (hostname1.ip, instance_name))
3191

    
3192
    # MAC address verification
3193
    if self.op.mac != "auto":
3194
      if not utils.IsValidMac(self.op.mac.lower()):
3195
        raise errors.OpPrereqError("invalid MAC address specified: %s" %
3196
                                   self.op.mac)
3197

    
3198
    # bridge verification
3199
    bridge = getattr(self.op, "bridge", None)
3200
    if bridge is None:
3201
      self.op.bridge = self.cfg.GetDefBridge()
3202
    else:
3203
      self.op.bridge = bridge
3204

    
3205
    # boot order verification
3206
    if self.op.hvm_boot_order is not None:
3207
      if len(self.op.hvm_boot_order.strip("acdn")) != 0:
3208
        raise errors.OpPrereqError("invalid boot order specified,"
3209
                                   " must be one or more of [acdn]")
3210
    # file storage checks
3211
    if (self.op.file_driver and
3212
        not self.op.file_driver in constants.FILE_DRIVER):
3213
      raise errors.OpPrereqError("Invalid file driver name '%s'" %
3214
                                 self.op.file_driver)
3215

    
3216
    if self.op.file_storage_dir and os.path.isabs(self.op.file_storage_dir):
3217
      raise errors.OpPrereqError("File storage directory not a relative"
3218
                                 " path")
3219
    #### allocator run
3220

    
3221
    if [self.op.iallocator, self.op.pnode].count(None) != 1:
3222
      raise errors.OpPrereqError("One and only one of iallocator and primary"
3223
                                 " node must be given")
3224

    
3225
    if self.op.iallocator is not None:
3226
      self._RunAllocator()
3227

    
3228
    #### node related checks
3229

    
3230
    # check primary node
3231
    pnode = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.pnode))
3232
    if pnode is None:
3233
      raise errors.OpPrereqError("Primary node '%s' is unknown" %
3234
                                 self.op.pnode)
3235
    self.op.pnode = pnode.name
3236
    self.pnode = pnode
3237
    self.secondaries = []
3238

    
3239
    # mirror node verification
3240
    if self.op.disk_template in constants.DTS_NET_MIRROR:
3241
      if getattr(self.op, "snode", None) is None:
3242
        raise errors.OpPrereqError("The networked disk templates need"
3243
                                   " a mirror node")
3244

    
3245
      snode_name = self.cfg.ExpandNodeName(self.op.snode)
3246
      if snode_name is None:
3247
        raise errors.OpPrereqError("Unknown secondary node '%s'" %
3248
                                   self.op.snode)
3249
      elif snode_name == pnode.name:
3250
        raise errors.OpPrereqError("The secondary node cannot be"
3251
                                   " the primary node.")
3252
      self.secondaries.append(snode_name)
3253

    
3254
    req_size = _ComputeDiskSize(self.op.disk_template,
3255
                                self.op.disk_size, self.op.swap_size)
3256

    
3257
    # Check lv size requirements
3258
    if req_size is not None:
3259
      nodenames = [pnode.name] + self.secondaries
3260
      nodeinfo = rpc.call_node_info(nodenames, self.cfg.GetVGName())
3261
      for node in nodenames:
3262
        info = nodeinfo.get(node, None)
3263
        if not info:
3264
          raise errors.OpPrereqError("Cannot get current information"
3265
                                     " from node '%s'" % node)
3266
        vg_free = info.get('vg_free', None)
3267
        if not isinstance(vg_free, int):
3268
          raise errors.OpPrereqError("Can't compute free disk space on"
3269
                                     " node %s" % node)
3270
        if req_size > info['vg_free']:
3271
          raise errors.OpPrereqError("Not enough disk space on target node %s."
3272
                                     " %d MB available, %d MB required" %
3273
                                     (node, info['vg_free'], req_size))
3274

    
3275
    # os verification
3276
    os_obj = rpc.call_os_get(pnode.name, self.op.os_type)
3277
    if not os_obj:
3278
      raise errors.OpPrereqError("OS '%s' not in supported os list for"
3279
                                 " primary node"  % self.op.os_type)
3280

    
3281
    if self.op.kernel_path == constants.VALUE_NONE:
3282
      raise errors.OpPrereqError("Can't set instance kernel to none")
3283

    
3284

    
3285
    # bridge check on primary node
3286
    if not rpc.call_bridges_exist(self.pnode.name, [self.op.bridge]):
3287
      raise errors.OpPrereqError("target bridge '%s' does not exist on"
3288
                                 " destination node '%s'" %
3289
                                 (self.op.bridge, pnode.name))
3290

    
3291
    # memory check on primary node
3292
    if self.op.start:
3293
      _CheckNodeFreeMemory(self.cfg, self.pnode.name,
3294
                           "creating instance %s" % self.op.instance_name,
3295
                           self.op.mem_size)
3296

    
3297
    # hvm_cdrom_image_path verification
3298
    if self.op.hvm_cdrom_image_path is not None:
3299
      if not os.path.isabs(self.op.hvm_cdrom_image_path):
3300
        raise errors.OpPrereqError("The path to the HVM CDROM image must"
3301
                                   " be an absolute path or None, not %s" %
3302
                                   self.op.hvm_cdrom_image_path)
3303
      if not os.path.isfile(self.op.hvm_cdrom_image_path):
3304
        raise errors.OpPrereqError("The HVM CDROM image must either be a"
3305
                                   " regular file or a symlink pointing to"
3306
                                   " an existing regular file, not %s" %
3307
                                   self.op.hvm_cdrom_image_path)
3308

    
3309
    # vnc_bind_address verification
3310
    if self.op.vnc_bind_address is not None:
3311
      if not utils.IsValidIP(self.op.vnc_bind_address):
3312
        raise errors.OpPrereqError("given VNC bind address '%s' doesn't look"
3313
                                   " like a valid IP address" %
3314
                                   self.op.vnc_bind_address)
3315

    
3316
    # Xen HVM device type checks
3317
    if self.sstore.GetHypervisorType() == constants.HT_XEN_HVM31:
3318
      if self.op.hvm_nic_type not in constants.HT_HVM_VALID_NIC_TYPES:
3319
        raise errors.OpPrereqError("Invalid NIC type %s specified for Xen HVM"
3320
                                   " hypervisor" % self.op.hvm_nic_type)
3321
      if self.op.hvm_disk_type not in constants.HT_HVM_VALID_DISK_TYPES:
3322
        raise errors.OpPrereqError("Invalid disk type %s specified for Xen HVM"
3323
                                   " hypervisor" % self.op.hvm_disk_type)
3324

    
3325
    if self.op.start:
3326
      self.instance_status = 'up'
3327
    else:
3328
      self.instance_status = 'down'
3329

    
3330
  def Exec(self, feedback_fn):
3331
    """Create and add the instance to the cluster.
3332

3333
    """
3334
    instance = self.op.instance_name
3335
    pnode_name = self.pnode.name
3336

    
3337
    if self.op.mac == "auto":
3338
      mac_address = self.cfg.GenerateMAC()
3339
    else:
3340
      mac_address = self.op.mac
3341

    
3342
    nic = objects.NIC(bridge=self.op.bridge, mac=mac_address)
3343
    if self.inst_ip is not None:
3344
      nic.ip = self.inst_ip
3345

    
3346
    ht_kind = self.sstore.GetHypervisorType()
3347
    if ht_kind in constants.HTS_REQ_PORT:
3348
      network_port = self.cfg.AllocatePort()
3349
    else:
3350
      network_port = None
3351

    
3352
    if self.op.vnc_bind_address is None:
3353
      self.op.vnc_bind_address = constants.VNC_DEFAULT_BIND_ADDRESS
3354

    
3355
    # this is needed because os.path.join does not accept None arguments
3356
    if self.op.file_storage_dir is None:
3357
      string_file_storage_dir = ""
3358
    else:
3359
      string_file_storage_dir = self.op.file_storage_dir
3360

    
3361
    # build the full file storage dir path
3362
    file_storage_dir = os.path.normpath(os.path.join(
3363
                                        self.sstore.GetFileStorageDir(),
3364
                                        string_file_storage_dir, instance))
3365

    
3366

    
3367
    disks = _GenerateDiskTemplate(self.cfg,
3368
                                  self.op.disk_template,
3369
                                  instance, pnode_name,
3370
                                  self.secondaries, self.op.disk_size,
3371
                                  self.op.swap_size,
3372
                                  file_storage_dir,
3373
                                  self.op.file_driver)
3374

    
3375
    iobj = objects.Instance(name=instance, os=self.op.os_type,
3376
                            primary_node=pnode_name,
3377
                            memory=self.op.mem_size,
3378
                            vcpus=self.op.vcpus,
3379
                            nics=[nic], disks=disks,
3380
                            disk_template=self.op.disk_template,
3381
                            status=self.instance_status,
3382
                            network_port=network_port,
3383
                            kernel_path=self.op.kernel_path,
3384
                            initrd_path=self.op.initrd_path,
3385
                            hvm_boot_order=self.op.hvm_boot_order,
3386
                            hvm_acpi=self.op.hvm_acpi,
3387
                            hvm_pae=self.op.hvm_pae,
3388
                            hvm_cdrom_image_path=self.op.hvm_cdrom_image_path,
3389
                            vnc_bind_address=self.op.vnc_bind_address,
3390
                            hvm_nic_type=self.op.hvm_nic_type,
3391
                            hvm_disk_type=self.op.hvm_disk_type,
3392
                            )
3393

    
3394
    feedback_fn("* creating instance disks...")
3395
    if not _CreateDisks(self.cfg, iobj):
3396
      _RemoveDisks(iobj, self.cfg)
3397
      raise errors.OpExecError("Device creation failed, reverting...")
3398

    
3399
    feedback_fn("adding instance %s to cluster config" % instance)
3400

    
3401
    self.cfg.AddInstance(iobj)
3402
    # Add the new instance to the Ganeti Lock Manager
3403
    self.context.glm.add(locking.LEVEL_INSTANCE, instance)
3404

    
3405
    if self.op.wait_for_sync:
3406
      disk_abort = not _WaitForSync(self.cfg, iobj, self.proc)
3407
    elif iobj.disk_template in constants.DTS_NET_MIRROR:
3408
      # make sure the disks are not degraded (still sync-ing is ok)
3409
      time.sleep(15)
3410
      feedback_fn("* checking mirrors status")
3411
      disk_abort = not _WaitForSync(self.cfg, iobj, self.proc, oneshot=True)
3412
    else:
3413
      disk_abort = False
3414

    
3415
    if disk_abort:
3416
      _RemoveDisks(iobj, self.cfg)
3417
      self.cfg.RemoveInstance(iobj.name)
3418
      # Remove the new instance from the Ganeti Lock Manager
3419
      self.context.glm.remove(locking.LEVEL_INSTANCE, iobj.name)
3420
      raise errors.OpExecError("There are some degraded disks for"
3421
                               " this instance")
3422

    
3423
    feedback_fn("creating os for instance %s on node %s" %
3424
                (instance, pnode_name))
3425

    
3426
    if iobj.disk_template != constants.DT_DISKLESS:
3427
      if self.op.mode == constants.INSTANCE_CREATE:
3428
        feedback_fn("* running the instance OS create scripts...")
3429
        if not rpc.call_instance_os_add(pnode_name, iobj, "sda", "sdb"):
3430
          raise errors.OpExecError("could not add os for instance %s"
3431
                                   " on node %s" %
3432
                                   (instance, pnode_name))
3433

    
3434
      elif self.op.mode == constants.INSTANCE_IMPORT:
3435
        feedback_fn("* running the instance OS import scripts...")
3436
        src_node = self.op.src_node
3437
        src_image = self.src_image
3438
        if not rpc.call_instance_os_import(pnode_name, iobj, "sda", "sdb",
3439
                                                src_node, src_image):
3440
          raise errors.OpExecError("Could not import os for instance"
3441
                                   " %s on node %s" %
3442
                                   (instance, pnode_name))
3443
      else:
3444
        # also checked in the prereq part
3445
        raise errors.ProgrammerError("Unknown OS initialization mode '%s'"
3446
                                     % self.op.mode)
3447

    
3448
    if self.op.start:
3449
      logger.Info("starting instance %s on node %s" % (instance, pnode_name))
3450
      feedback_fn("* starting instance...")
3451
      if not rpc.call_instance_start(pnode_name, iobj, None):
3452
        raise errors.OpExecError("Could not start instance")
3453

    
3454

    
3455
class LUConnectConsole(NoHooksLU):
3456
  """Connect to an instance's console.
3457

3458
  This is somewhat special in that it returns the command line that
3459
  you need to run on the master node in order to connect to the
3460
  console.
3461

3462
  """
3463
  _OP_REQP = ["instance_name"]
3464
  REQ_BGL = False
3465

    
3466
  def ExpandNames(self):
3467
    self._ExpandAndLockInstance()
3468

    
3469
  def CheckPrereq(self):
3470
    """Check prerequisites.
3471

3472
    This checks that the instance is in the cluster.
3473

3474
    """
3475
    self.instance = self.cfg.GetInstanceInfo(self.op.instance_name)
3476
    assert self.instance is not None, \
3477
      "Cannot retrieve locked instance %s" % self.op.instance_name
3478

    
3479
  def Exec(self, feedback_fn):
3480
    """Connect to the console of an instance
3481

3482
    """
3483
    instance = self.instance
3484
    node = instance.primary_node
3485

    
3486
    node_insts = rpc.call_instance_list([node])[node]
3487
    if node_insts is False:
3488
      raise errors.OpExecError("Can't connect to node %s." % node)
3489

    
3490
    if instance.name not in node_insts:
3491
      raise errors.OpExecError("Instance %s is not running." % instance.name)
3492

    
3493
    logger.Debug("connecting to console of %s on %s" % (instance.name, node))
3494

    
3495
    hyper = hypervisor.GetHypervisor()
3496
    console_cmd = hyper.GetShellCommandForConsole(instance)
3497

    
3498
    # build ssh cmdline
3499
    return self.ssh.BuildCmd(node, "root", console_cmd, batch=True, tty=True)
3500

    
3501

    
3502
class LUReplaceDisks(LogicalUnit):
3503
  """Replace the disks of an instance.
3504

3505
  """
3506
  HPATH = "mirrors-replace"
3507
  HTYPE = constants.HTYPE_INSTANCE
3508
  _OP_REQP = ["instance_name", "mode", "disks"]
3509

    
3510
  def _RunAllocator(self):
3511
    """Compute a new secondary node using an IAllocator.
3512

3513
    """
3514
    ial = IAllocator(self.cfg, self.sstore,
3515
                     mode=constants.IALLOCATOR_MODE_RELOC,
3516
                     name=self.op.instance_name,
3517
                     relocate_from=[self.sec_node])
3518

    
3519
    ial.Run(self.op.iallocator)
3520

    
3521
    if not ial.success:
3522
      raise errors.OpPrereqError("Can't compute nodes using"
3523
                                 " iallocator '%s': %s" % (self.op.iallocator,
3524
                                                           ial.info))
3525
    if len(ial.nodes) != ial.required_nodes:
3526
      raise errors.OpPrereqError("iallocator '%s' returned invalid number"
3527
                                 " of nodes (%s), required %s" %
3528
                                 (len(ial.nodes), ial.required_nodes))
3529
    self.op.remote_node = ial.nodes[0]
3530
    logger.ToStdout("Selected new secondary for the instance: %s" %
3531
                    self.op.remote_node)
3532

    
3533
  def BuildHooksEnv(self):
3534
    """Build hooks env.
3535

3536
    This runs on the master, the primary and all the secondaries.
3537

3538
    """
3539
    env = {
3540
      "MODE": self.op.mode,
3541
      "NEW_SECONDARY": self.op.remote_node,
3542
      "OLD_SECONDARY": self.instance.secondary_nodes[0],
3543
      }
3544
    env.update(_BuildInstanceHookEnvByObject(self.instance))
3545
    nl = [
3546
      self.sstore.GetMasterNode(),
3547
      self.instance.primary_node,
3548
      ]
3549
    if self.op.remote_node is not None:
3550
      nl.append(self.op.remote_node)
3551
    return env, nl, nl
3552

    
3553
  def CheckPrereq(self):
3554
    """Check prerequisites.
3555

3556
    This checks that the instance is in the cluster.
3557

3558
    """
3559
    if not hasattr(self.op, "remote_node"):
3560
      self.op.remote_node = None
3561

    
3562
    instance = self.cfg.GetInstanceInfo(
3563
      self.cfg.ExpandInstanceName(self.op.instance_name))
3564
    if instance is None:
3565
      raise errors.OpPrereqError("Instance '%s' not known" %
3566
                                 self.op.instance_name)
3567
    self.instance = instance
3568
    self.op.instance_name = instance.name
3569

    
3570
    if instance.disk_template not in constants.DTS_NET_MIRROR:
3571
      raise errors.OpPrereqError("Instance's disk layout is not"
3572
                                 " network mirrored.")
3573

    
3574
    if len(instance.secondary_nodes) != 1:
3575
      raise errors.OpPrereqError("The instance has a strange layout,"
3576
                                 " expected one secondary but found %d" %
3577
                                 len(instance.secondary_nodes))
3578

    
3579
    self.sec_node = instance.secondary_nodes[0]
3580

    
3581
    ia_name = getattr(self.op, "iallocator", None)
3582
    if ia_name is not None:
3583
      if self.op.remote_node is not None:
3584
        raise errors.OpPrereqError("Give either the iallocator or the new"
3585
                                   " secondary, not both")
3586
      self._RunAllocator()
3587

    
3588
    remote_node = self.op.remote_node
3589
    if remote_node is not None:
3590
      remote_node = self.cfg.ExpandNodeName(remote_node)
3591
      if remote_node is None:
3592
        raise errors.OpPrereqError("Node '%s' not known" %
3593
                                   self.op.remote_node)
3594
      self.remote_node_info = self.cfg.GetNodeInfo(remote_node)
3595
    else:
3596
      self.remote_node_info = None
3597
    if remote_node == instance.primary_node:
3598
      raise errors.OpPrereqError("The specified node is the primary node of"
3599
                                 " the instance.")
3600
    elif remote_node == self.sec_node:
3601
      if self.op.mode == constants.REPLACE_DISK_SEC:
3602
        # this is for DRBD8, where we can't execute the same mode of
3603
        # replacement as for drbd7 (no different port allocated)
3604
        raise errors.OpPrereqError("Same secondary given, cannot execute"
3605
                                   " replacement")
3606
    if instance.disk_template == constants.DT_DRBD8:
3607
      if (self.op.mode == constants.REPLACE_DISK_ALL and
3608
          remote_node is not None):
3609
        # switch to replace secondary mode
3610
        self.op.mode = constants.REPLACE_DISK_SEC
3611

    
3612
      if self.op.mode == constants.REPLACE_DISK_ALL:
3613
        raise errors.OpPrereqError("Template 'drbd' only allows primary or"
3614
                                   " secondary disk replacement, not"
3615
                                   " both at once")
3616
      elif self.op.mode == constants.REPLACE_DISK_PRI:
3617
        if remote_node is not None:
3618
          raise errors.OpPrereqError("Template 'drbd' does not allow changing"
3619
                                     " the secondary while doing a primary"
3620
                                     " node disk replacement")
3621
        self.tgt_node = instance.primary_node
3622
        self.oth_node = instance.secondary_nodes[0]
3623
      elif self.op.mode == constants.REPLACE_DISK_SEC:
3624
        self.new_node = remote_node # this can be None, in which case
3625
                                    # we don't change the secondary
3626
        self.tgt_node = instance.secondary_nodes[0]
3627
        self.oth_node = instance.primary_node
3628
      else:
3629
        raise errors.ProgrammerError("Unhandled disk replace mode")
3630

    
3631
    for name in self.op.disks:
3632
      if instance.FindDisk(name) is None:
3633
        raise errors.OpPrereqError("Disk '%s' not found for instance '%s'" %
3634
                                   (name, instance.name))
3635
    self.op.remote_node = remote_node
3636

    
3637
  def _ExecD8DiskOnly(self, feedback_fn):
3638
    """Replace a disk on the primary or secondary for dbrd8.
3639

3640
    The algorithm for replace is quite complicated:
3641
      - for each disk to be replaced:
3642
        - create new LVs on the target node with unique names
3643
        - detach old LVs from the drbd device
3644
        - rename old LVs to name_replaced.<time_t>
3645
        - rename new LVs to old LVs
3646
        - attach the new LVs (with the old names now) to the drbd device
3647
      - wait for sync across all devices
3648
      - for each modified disk:
3649
        - remove old LVs (which have the name name_replaces.<time_t>)
3650

3651
    Failures are not very well handled.
3652

3653
    """
3654
    steps_total = 6
3655
    warning, info = (self.proc.LogWarning, self.proc.LogInfo)
3656
    instance = self.instance
3657
    iv_names = {}
3658
    vgname = self.cfg.GetVGName()
3659
    # start of work
3660
    cfg = self.cfg
3661
    tgt_node = self.tgt_node
3662
    oth_node = self.oth_node
3663

    
3664
    # Step: check device activation
3665
    self.proc.LogStep(1, steps_total, "check device existence")
3666
    info("checking volume groups")
3667
    my_vg = cfg.GetVGName()
3668
    results = rpc.call_vg_list([oth_node, tgt_node])
3669
    if not results:
3670
      raise errors.OpExecError("Can't list volume groups on the nodes")
3671
    for node in oth_node, tgt_node:
3672
      res = results.get(node, False)
3673
      if not res or my_vg not in res:
3674
        raise errors.OpExecError("Volume group '%s' not found on %s" %
3675
                                 (my_vg, node))
3676
    for dev in instance.disks:
3677
      if not dev.iv_name in self.op.disks:
3678
        continue
3679
      for node in tgt_node, oth_node:
3680
        info("checking %s on %s" % (dev.iv_name, node))
3681
        cfg.SetDiskID(dev, node)
3682
        if not rpc.call_blockdev_find(node, dev):
3683
          raise errors.OpExecError("Can't find device %s on node %s" %
3684
                                   (dev.iv_name, node))
3685

    
3686
    # Step: check other node consistency
3687
    self.proc.LogStep(2, steps_total, "check peer consistency")
3688
    for dev in instance.disks:
3689
      if not dev.iv_name in self.op.disks:
3690
        continue
3691
      info("checking %s consistency on %s" % (dev.iv_name, oth_node))
3692
      if not _CheckDiskConsistency(self.cfg, dev, oth_node,
3693
                                   oth_node==instance.primary_node):
3694
        raise errors.OpExecError("Peer node (%s) has degraded storage, unsafe"
3695
                                 " to replace disks on this node (%s)" %
3696
                                 (oth_node, tgt_node))
3697

    
3698
    # Step: create new storage
3699
    self.proc.LogStep(3, steps_total, "allocate new storage")
3700
    for dev in instance.disks:
3701
      if not dev.iv_name in self.op.disks:
3702
        continue
3703
      size = dev.size
3704
      cfg.SetDiskID(dev, tgt_node)
3705
      lv_names = [".%s_%s" % (dev.iv_name, suf) for suf in ["data", "meta"]]
3706
      names = _GenerateUniqueNames(cfg, lv_names)
3707
      lv_data = objects.Disk(dev_type=constants.LD_LV, size=size,
3708
                             logical_id=(vgname, names[0]))
3709
      lv_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
3710
                             logical_id=(vgname, names[1]))
3711
      new_lvs = [lv_data, lv_meta]
3712
      old_lvs = dev.children
3713
      iv_names[dev.iv_name] = (dev, old_lvs, new_lvs)
3714
      info("creating new local storage on %s for %s" %
3715
           (tgt_node, dev.iv_name))
3716
      # since we *always* want to create this LV, we use the
3717
      # _Create...OnPrimary (which forces the creation), even if we
3718
      # are talking about the secondary node
3719
      for new_lv in new_lvs:
3720
        if not _CreateBlockDevOnPrimary(cfg, tgt_node, instance, new_lv,
3721
                                        _GetInstanceInfoText(instance)):
3722
          raise errors.OpExecError("Failed to create new LV named '%s' on"
3723
                                   " node '%s'" %
3724
                                   (new_lv.logical_id[1], tgt_node))
3725

    
3726
    # Step: for each lv, detach+rename*2+attach
3727
    self.proc.LogStep(4, steps_total, "change drbd configuration")
3728
    for dev, old_lvs, new_lvs in iv_names.itervalues():
3729
      info("detaching %s drbd from local storage" % dev.iv_name)
3730
      if not rpc.call_blockdev_removechildren(tgt_node, dev, old_lvs):
3731
        raise errors.OpExecError("Can't detach drbd from local storage on node"
3732
                                 " %s for device %s" % (tgt_node, dev.iv_name))
3733
      #dev.children = []
3734
      #cfg.Update(instance)
3735

    
3736
      # ok, we created the new LVs, so now we know we have the needed
3737
      # storage; as such, we proceed on the target node to rename
3738
      # old_lv to _old, and new_lv to old_lv; note that we rename LVs
3739
      # using the assumption that logical_id == physical_id (which in
3740
      # turn is the unique_id on that node)
3741

    
3742
      # FIXME(iustin): use a better name for the replaced LVs
3743
      temp_suffix = int(time.time())
3744
      ren_fn = lambda d, suff: (d.physical_id[0],
3745
                                d.physical_id[1] + "_replaced-%s" % suff)
3746
      # build the rename list based on what LVs exist on the node
3747
      rlist = []
3748
      for to_ren in old_lvs:
3749
        find_res = rpc.call_blockdev_find(tgt_node, to_ren)
3750
        if find_res is not None: # device exists
3751
          rlist.append((to_ren, ren_fn(to_ren, temp_suffix)))
3752

    
3753
      info("renaming the old LVs on the target node")
3754
      if not rpc.call_blockdev_rename(tgt_node, rlist):
3755
        raise errors.OpExecError("Can't rename old LVs on node %s" % tgt_node)
3756
      # now we rename the new LVs to the old LVs
3757
      info("renaming the new LVs on the target node")
3758
      rlist = [(new, old.physical_id) for old, new in zip(old_lvs, new_lvs)]
3759
      if not rpc.call_blockdev_rename(tgt_node, rlist):
3760
        raise errors.OpExecError("Can't rename new LVs on node %s" % tgt_node)
3761

    
3762
      for old, new in zip(old_lvs, new_lvs):
3763
        new.logical_id = old.logical_id
3764
        cfg.SetDiskID(new, tgt_node)
3765

    
3766
      for disk in old_lvs:
3767
        disk.logical_id = ren_fn(disk, temp_suffix)
3768
        cfg.SetDiskID(disk, tgt_node)
3769

    
3770
      # now that the new lvs have the old name, we can add them to the device
3771
      info("adding new mirror component on %s" % tgt_node)
3772
      if not rpc.call_blockdev_addchildren(tgt_node, dev, new_lvs):
3773
        for new_lv in new_lvs:
3774
          if not rpc.call_blockdev_remove(tgt_node, new_lv):
3775
            warning("Can't rollback device %s", hint="manually cleanup unused"
3776
                    " logical volumes")
3777
        raise errors.OpExecError("Can't add local storage to drbd")
3778

    
3779
      dev.children = new_lvs
3780
      cfg.Update(instance)
3781

    
3782
    # Step: wait for sync
3783

    
3784
    # this can fail as the old devices are degraded and _WaitForSync
3785
    # does a combined result over all disks, so we don't check its
3786
    # return value
3787
    self.proc.LogStep(5, steps_total, "sync devices")
3788
    _WaitForSync(cfg, instance, self.proc, unlock=True)
3789

    
3790
    # so check manually all the devices
3791
    for name, (dev, old_lvs, new_lvs) in iv_names.iteritems():
3792
      cfg.SetDiskID(dev, instance.primary_node)
3793
      is_degr = rpc.call_blockdev_find(instance.primary_node, dev)[5]
3794
      if is_degr:
3795
        raise errors.OpExecError("DRBD device %s is degraded!" % name)
3796

    
3797
    # Step: remove old storage
3798
    self.proc.LogStep(6, steps_total, "removing old storage")
3799
    for name, (dev, old_lvs, new_lvs) in iv_names.iteritems():
3800
      info("remove logical volumes for %s" % name)
3801
      for lv in old_lvs:
3802
        cfg.SetDiskID(lv, tgt_node)
3803
        if not rpc.call_blockdev_remove(tgt_node, lv):
3804
          warning("Can't remove old LV", hint="manually remove unused LVs")
3805
          continue
3806

    
3807
  def _ExecD8Secondary(self, feedback_fn):
3808
    """Replace the secondary node for drbd8.
3809

3810
    The algorithm for replace is quite complicated:
3811
      - for all disks of the instance:
3812
        - create new LVs on the new node with same names
3813
        - shutdown the drbd device on the old secondary
3814
        - disconnect the drbd network on the primary
3815
        - create the drbd device on the new secondary
3816
        - network attach the drbd on the primary, using an artifice:
3817
          the drbd code for Attach() will connect to the network if it
3818
          finds a device which is connected to the good local disks but
3819
          not network enabled
3820
      - wait for sync across all devices
3821
      - remove all disks from the old secondary
3822

3823
    Failures are not very well handled.
3824

3825
    """
3826
    steps_total = 6
3827
    warning, info = (self.proc.LogWarning, self.proc.LogInfo)
3828
    instance = self.instance
3829
    iv_names = {}
3830
    vgname = self.cfg.GetVGName()
3831
    # start of work
3832
    cfg = self.cfg
3833
    old_node = self.tgt_node
3834
    new_node = self.new_node
3835
    pri_node = instance.primary_node
3836

    
3837
    # Step: check device activation
3838
    self.proc.LogStep(1, steps_total, "check device existence")
3839
    info("checking volume groups")
3840
    my_vg = cfg.GetVGName()
3841
    results = rpc.call_vg_list([pri_node, new_node])
3842
    if not results:
3843
      raise errors.OpExecError("Can't list volume groups on the nodes")
3844
    for node in pri_node, new_node:
3845
      res = results.get(node, False)
3846
      if not res or my_vg not in res:
3847
        raise errors.OpExecError("Volume group '%s' not found on %s" %
3848
                                 (my_vg, node))
3849
    for dev in instance.disks:
3850
      if not dev.iv_name in self.op.disks:
3851
        continue
3852
      info("checking %s on %s" % (dev.iv_name, pri_node))
3853
      cfg.SetDiskID(dev, pri_node)
3854
      if not rpc.call_blockdev_find(pri_node, dev):
3855
        raise errors.OpExecError("Can't find device %s on node %s" %
3856
                                 (dev.iv_name, pri_node))
3857

    
3858
    # Step: check other node consistency
3859
    self.proc.LogStep(2, steps_total, "check peer consistency")
3860
    for dev in instance.disks:
3861
      if not dev.iv_name in self.op.disks:
3862
        continue
3863
      info("checking %s consistency on %s" % (dev.iv_name, pri_node))
3864
      if not _CheckDiskConsistency(self.cfg, dev, pri_node, True, ldisk=True):
3865
        raise errors.OpExecError("Primary node (%s) has degraded storage,"
3866
                                 " unsafe to replace the secondary" %
3867
                                 pri_node)
3868

    
3869
    # Step: create new storage
3870
    self.proc.LogStep(3, steps_total, "allocate new storage")
3871
    for dev in instance.disks:
3872
      size = dev.size
3873
      info("adding new local storage on %s for %s" % (new_node, dev.iv_name))
3874
      # since we *always* want to create this LV, we use the
3875
      # _Create...OnPrimary (which forces the creation), even if we
3876
      # are talking about the secondary node
3877
      for new_lv in dev.children:
3878
        if not _CreateBlockDevOnPrimary(cfg, new_node, instance, new_lv,
3879
                                        _GetInstanceInfoText(instance)):
3880
          raise errors.OpExecError("Failed to create new LV named '%s' on"
3881
                                   " node '%s'" %
3882
                                   (new_lv.logical_id[1], new_node))
3883

    
3884
      iv_names[dev.iv_name] = (dev, dev.children)
3885

    
3886
    self.proc.LogStep(4, steps_total, "changing drbd configuration")
3887
    for dev in instance.disks:
3888
      size = dev.size
3889
      info("activating a new drbd on %s for %s" % (new_node, dev.iv_name))
3890
      # create new devices on new_node
3891
      new_drbd = objects.Disk(dev_type=constants.LD_DRBD8,
3892
                              logical_id=(pri_node, new_node,
3893
                                          dev.logical_id[2]),
3894
                              children=dev.children)
3895
      if not _CreateBlockDevOnSecondary(cfg, new_node, instance,
3896
                                        new_drbd, False,
3897
                                      _GetInstanceInfoText(instance)):
3898
        raise errors.OpExecError("Failed to create new DRBD on"
3899
                                 " node '%s'" % new_node)
3900

    
3901
    for dev in instance.disks:
3902
      # we have new devices, shutdown the drbd on the old secondary
3903
      info("shutting down drbd for %s on old node" % dev.iv_name)
3904
      cfg.SetDiskID(dev, old_node)
3905
      if not rpc.call_blockdev_shutdown(old_node, dev):
3906
        warning("Failed to shutdown drbd for %s on old node" % dev.iv_name,
3907
                hint="Please cleanup this device manually as soon as possible")
3908

    
3909
    info("detaching primary drbds from the network (=> standalone)")
3910
    done = 0
3911
    for dev in instance.disks:
3912
      cfg.SetDiskID(dev, pri_node)
3913
      # set the physical (unique in bdev terms) id to None, meaning
3914
      # detach from network
3915
      dev.physical_id = (None,) * len(dev.physical_id)
3916
      # and 'find' the device, which will 'fix' it to match the
3917
      # standalone state
3918
      if rpc.call_blockdev_find(pri_node, dev):
3919
        done += 1
3920
      else:
3921
        warning("Failed to detach drbd %s from network, unusual case" %
3922
                dev.iv_name)
3923

    
3924
    if not done:
3925
      # no detaches succeeded (very unlikely)
3926
      raise errors.OpExecError("Can't detach at least one DRBD from old node")
3927

    
3928
    # if we managed to detach at least one, we update all the disks of
3929
    # the instance to point to the new secondary
3930
    info("updating instance configuration")
3931
    for dev in instance.disks:
3932
      dev.logical_id = (pri_node, new_node) + dev.logical_id[2:]
3933
      cfg.SetDiskID(dev, pri_node)
3934
    cfg.Update(instance)
3935

    
3936
    # and now perform the drbd attach
3937
    info("attaching primary drbds to new secondary (standalone => connected)")
3938
    failures = []
3939
    for dev in instance.disks:
3940
      info("attaching primary drbd for %s to new secondary node" % dev.iv_name)
3941
      # since the attach is smart, it's enough to 'find' the device,
3942
      # it will automatically activate the network, if the physical_id
3943
      # is correct
3944
      cfg.SetDiskID(dev, pri_node)
3945
      if not rpc.call_blockdev_find(pri_node, dev):
3946
        warning("can't attach drbd %s to new secondary!" % dev.iv_name,
3947
                "please do a gnt-instance info to see the status of disks")