Statistics
| Branch: | Tag: | Revision:

root / lib / cmdlib.py @ 243cdbcc

History | View | Annotate | Download (176.6 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 config
41
from ganeti import constants
42
from ganeti import objects
43
from ganeti import opcodes
44
from ganeti import ssconf
45
from ganeti import serializer
46

    
47

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

51
  Subclasses must follow these rules:
52
    - implement CheckPrereq which also fills in the opcode instance
53
      with all the fields (even if as None)
54
    - implement Exec
55
    - implement BuildHooksEnv
56
    - redefine HPATH and HTYPE
57
    - optionally redefine their run requirements (REQ_CLUSTER,
58
      REQ_MASTER); note that all commands require root permissions
59

60
  """
61
  HPATH = None
62
  HTYPE = None
63
  _OP_REQP = []
64
  REQ_CLUSTER = True
65
  REQ_MASTER = True
66

    
67
  def __init__(self, processor, op, cfg, sstore):
68
    """Constructor for LogicalUnit.
69

70
    This needs to be overriden in derived classes in order to check op
71
    validity.
72

73
    """
74
    self.proc = processor
75
    self.op = op
76
    self.cfg = cfg
77
    self.sstore = sstore
78
    self.__ssh = None
79

    
80
    for attr_name in self._OP_REQP:
81
      attr_val = getattr(op, attr_name, None)
82
      if attr_val is None:
83
        raise errors.OpPrereqError("Required parameter '%s' missing" %
84
                                   attr_name)
85
    if self.REQ_CLUSTER:
86
      if not cfg.IsCluster():
87
        raise errors.OpPrereqError("Cluster not initialized yet,"
88
                                   " use 'gnt-cluster init' first.")
89
      if self.REQ_MASTER:
90
        master = sstore.GetMasterNode()
91
        if master != utils.HostInfo().name:
92
          raise errors.OpPrereqError("Commands must be run on the master"
93
                                     " node %s" % master)
94

    
95
  def __GetSSH(self):
96
    """Returns the SshRunner object
97

98
    """
99
    if not self.__ssh:
100
      self.__ssh = ssh.SshRunner(self.sstore)
101
    return self.__ssh
102

    
103
  ssh = property(fget=__GetSSH)
104

    
105
  def CheckPrereq(self):
106
    """Check prerequisites for this LU.
107

108
    This method should check that the prerequisites for the execution
109
    of this LU are fulfilled. It can do internode communication, but
110
    it should be idempotent - no cluster or system changes are
111
    allowed.
112

113
    The method should raise errors.OpPrereqError in case something is
114
    not fulfilled. Its return value is ignored.
115

116
    This method should also update all the parameters of the opcode to
117
    their canonical form; e.g. a short node name must be fully
118
    expanded after this method has successfully completed (so that
119
    hooks, logging, etc. work correctly).
120

121
    """
122
    raise NotImplementedError
123

    
124
  def Exec(self, feedback_fn):
125
    """Execute the LU.
126

127
    This method should implement the actual work. It should raise
128
    errors.OpExecError for failures that are somewhat dealt with in
129
    code, or expected.
130

131
    """
132
    raise NotImplementedError
133

    
134
  def BuildHooksEnv(self):
135
    """Build hooks environment for this LU.
136

137
    This method should return a three-node tuple consisting of: a dict
138
    containing the environment that will be used for running the
139
    specific hook for this LU, a list of node names on which the hook
140
    should run before the execution, and a list of node names on which
141
    the hook should run after the execution.
142

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

148
    No nodes should be returned as an empty list (and not None).
149

150
    Note that if the HPATH for a LU class is None, this function will
151
    not be called.
152

153
    """
154
    raise NotImplementedError
155

    
156
  def HooksCallBack(self, phase, hook_results, feedback_fn, lu_result):
157
    """Notify the LU about the results of its hooks.
158

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

165
    Args:
166
      phase: the hooks phase that has just been run
167
      hooks_results: the results of the multi-node hooks rpc call
168
      feedback_fn: function to send feedback back to the caller
169
      lu_result: the previous result this LU had, or None in the PRE phase.
170

171
    """
172
    return lu_result
173

    
174

    
175
class NoHooksLU(LogicalUnit):
176
  """Simple LU which runs no hooks.
177

178
  This LU is intended as a parent for other LogicalUnits which will
179
  run no hooks, in order to reduce duplicate code.
180

181
  """
182
  HPATH = None
183
  HTYPE = None
184

    
185

    
186
def _AddHostToEtcHosts(hostname):
187
  """Wrapper around utils.SetEtcHostsEntry.
188

189
  """
190
  hi = utils.HostInfo(name=hostname)
191
  utils.SetEtcHostsEntry(constants.ETC_HOSTS, hi.ip, hi.name, [hi.ShortName()])
192

    
193

    
194
def _RemoveHostFromEtcHosts(hostname):
195
  """Wrapper around utils.RemoveEtcHostsEntry.
196

197
  """
198
  hi = utils.HostInfo(name=hostname)
199
  utils.RemoveEtcHostsEntry(constants.ETC_HOSTS, hi.name)
200
  utils.RemoveEtcHostsEntry(constants.ETC_HOSTS, hi.ShortName())
201

    
202

    
203
def _GetWantedNodes(lu, nodes):
204
  """Returns list of checked and expanded node names.
205

206
  Args:
207
    nodes: List of nodes (strings) or None for all
208

209
  """
210
  if not isinstance(nodes, list):
211
    raise errors.OpPrereqError("Invalid argument type 'nodes'")
212

    
213
  if nodes:
214
    wanted = []
215

    
216
    for name in nodes:
217
      node = lu.cfg.ExpandNodeName(name)
218
      if node is None:
219
        raise errors.OpPrereqError("No such node name '%s'" % name)
220
      wanted.append(node)
221

    
222
  else:
223
    wanted = lu.cfg.GetNodeList()
224
  return utils.NiceSort(wanted)
225

    
226

    
227
def _GetWantedInstances(lu, instances):
228
  """Returns list of checked and expanded instance names.
229

230
  Args:
231
    instances: List of instances (strings) or None for all
232

233
  """
234
  if not isinstance(instances, list):
235
    raise errors.OpPrereqError("Invalid argument type 'instances'")
236

    
237
  if instances:
238
    wanted = []
239

    
240
    for name in instances:
241
      instance = lu.cfg.ExpandInstanceName(name)
242
      if instance is None:
243
        raise errors.OpPrereqError("No such instance name '%s'" % name)
244
      wanted.append(instance)
245

    
246
  else:
247
    wanted = lu.cfg.GetInstanceList()
248
  return utils.NiceSort(wanted)
249

    
250

    
251
def _CheckOutputFields(static, dynamic, selected):
252
  """Checks whether all selected fields are valid.
253

254
  Args:
255
    static: Static fields
256
    dynamic: Dynamic fields
257

258
  """
259
  static_fields = frozenset(static)
260
  dynamic_fields = frozenset(dynamic)
261

    
262
  all_fields = static_fields | dynamic_fields
263

    
264
  if not all_fields.issuperset(selected):
265
    raise errors.OpPrereqError("Unknown output fields selected: %s"
266
                               % ",".join(frozenset(selected).
267
                                          difference(all_fields)))
268

    
269

    
270
def _BuildInstanceHookEnv(name, primary_node, secondary_nodes, os_type, status,
271
                          memory, vcpus, nics):
272
  """Builds instance related env variables for hooks from single variables.
273

274
  Args:
275
    secondary_nodes: List of secondary nodes as strings
276
  """
277
  env = {
278
    "OP_TARGET": name,
279
    "INSTANCE_NAME": name,
280
    "INSTANCE_PRIMARY": primary_node,
281
    "INSTANCE_SECONDARIES": " ".join(secondary_nodes),
282
    "INSTANCE_OS_TYPE": os_type,
283
    "INSTANCE_STATUS": status,
284
    "INSTANCE_MEMORY": memory,
285
    "INSTANCE_VCPUS": vcpus,
286
  }
287

    
288
  if nics:
289
    nic_count = len(nics)
290
    for idx, (ip, bridge, mac) in enumerate(nics):
291
      if ip is None:
292
        ip = ""
293
      env["INSTANCE_NIC%d_IP" % idx] = ip
294
      env["INSTANCE_NIC%d_BRIDGE" % idx] = bridge
295
      env["INSTANCE_NIC%d_HWADDR" % idx] = mac
296
  else:
297
    nic_count = 0
298

    
299
  env["INSTANCE_NIC_COUNT"] = nic_count
300

    
301
  return env
302

    
303

    
304
def _BuildInstanceHookEnvByObject(instance, override=None):
305
  """Builds instance related env variables for hooks from an object.
306

307
  Args:
308
    instance: objects.Instance object of instance
309
    override: dict of values to override
310
  """
311
  args = {
312
    'name': instance.name,
313
    'primary_node': instance.primary_node,
314
    'secondary_nodes': instance.secondary_nodes,
315
    'os_type': instance.os,
316
    'status': instance.os,
317
    'memory': instance.memory,
318
    'vcpus': instance.vcpus,
319
    'nics': [(nic.ip, nic.bridge, nic.mac) for nic in instance.nics],
320
  }
321
  if override:
322
    args.update(override)
323
  return _BuildInstanceHookEnv(**args)
324

    
325

    
326
def _HasValidVG(vglist, vgname):
327
  """Checks if the volume group list is valid.
328

329
  A non-None return value means there's an error, and the return value
330
  is the error message.
331

332
  """
333
  vgsize = vglist.get(vgname, None)
334
  if vgsize is None:
335
    return "volume group '%s' missing" % vgname
336
  elif vgsize < 20480:
337
    return ("volume group '%s' too small (20480MiB required, %dMib found)" %
338
            (vgname, vgsize))
339
  return None
340

    
341

    
342
def _InitSSHSetup(node):
343
  """Setup the SSH configuration for the cluster.
344

345

346
  This generates a dsa keypair for root, adds the pub key to the
347
  permitted hosts and adds the hostkey to its own known hosts.
348

349
  Args:
350
    node: the name of this host as a fqdn
351

352
  """
353
  priv_key, pub_key, auth_keys = ssh.GetUserFiles(constants.GANETI_RUNAS)
354

    
355
  for name in priv_key, pub_key:
356
    if os.path.exists(name):
357
      utils.CreateBackup(name)
358
    utils.RemoveFile(name)
359

    
360
  result = utils.RunCmd(["ssh-keygen", "-t", "dsa",
361
                         "-f", priv_key,
362
                         "-q", "-N", ""])
363
  if result.failed:
364
    raise errors.OpExecError("Could not generate ssh keypair, error %s" %
365
                             result.output)
366

    
367
  f = open(pub_key, 'r')
368
  try:
369
    utils.AddAuthorizedKey(auth_keys, f.read(8192))
370
  finally:
371
    f.close()
372

    
373

    
374
def _InitGanetiServerSetup(ss):
375
  """Setup the necessary configuration for the initial node daemon.
376

377
  This creates the nodepass file containing the shared password for
378
  the cluster and also generates the SSL certificate.
379

380
  """
381
  # Create pseudo random password
382
  randpass = sha.new(os.urandom(64)).hexdigest()
383
  # and write it into sstore
384
  ss.SetKey(ss.SS_NODED_PASS, randpass)
385

    
386
  result = utils.RunCmd(["openssl", "req", "-new", "-newkey", "rsa:1024",
387
                         "-days", str(365*5), "-nodes", "-x509",
388
                         "-keyout", constants.SSL_CERT_FILE,
389
                         "-out", constants.SSL_CERT_FILE, "-batch"])
390
  if result.failed:
391
    raise errors.OpExecError("could not generate server ssl cert, command"
392
                             " %s had exitcode %s and error message %s" %
393
                             (result.cmd, result.exit_code, result.output))
394

    
395
  os.chmod(constants.SSL_CERT_FILE, 0400)
396

    
397
  result = utils.RunCmd([constants.NODE_INITD_SCRIPT, "restart"])
398

    
399
  if result.failed:
400
    raise errors.OpExecError("Could not start the node daemon, command %s"
401
                             " had exitcode %s and error %s" %
402
                             (result.cmd, result.exit_code, result.output))
403

    
404

    
405
def _CheckInstanceBridgesExist(instance):
406
  """Check that the brigdes needed by an instance exist.
407

408
  """
409
  # check bridges existance
410
  brlist = [nic.bridge for nic in instance.nics]
411
  if not rpc.call_bridges_exist(instance.primary_node, brlist):
412
    raise errors.OpPrereqError("one or more target bridges %s does not"
413
                               " exist on destination node '%s'" %
414
                               (brlist, instance.primary_node))
415

    
416

    
417
class LUInitCluster(LogicalUnit):
418
  """Initialise the cluster.
419

420
  """
421
  HPATH = "cluster-init"
422
  HTYPE = constants.HTYPE_CLUSTER
423
  _OP_REQP = ["cluster_name", "hypervisor_type", "mac_prefix",
424
              "def_bridge", "master_netdev", "file_storage_dir"]
425
  REQ_CLUSTER = False
426

    
427
  def BuildHooksEnv(self):
428
    """Build hooks env.
429

430
    Notes: Since we don't require a cluster, we must manually add
431
    ourselves in the post-run node list.
432

433
    """
434
    env = {"OP_TARGET": self.op.cluster_name}
435
    return env, [], [self.hostname.name]
436

    
437
  def CheckPrereq(self):
438
    """Verify that the passed name is a valid one.
439

440
    """
441
    if config.ConfigWriter.IsCluster():
442
      raise errors.OpPrereqError("Cluster is already initialised")
443

    
444
    if self.op.hypervisor_type == constants.HT_XEN_HVM31:
445
      if not os.path.exists(constants.VNC_PASSWORD_FILE):
446
        raise errors.OpPrereqError("Please prepare the cluster VNC"
447
                                   "password file %s" %
448
                                   constants.VNC_PASSWORD_FILE)
449

    
450
    self.hostname = hostname = utils.HostInfo()
451

    
452
    if hostname.ip.startswith("127."):
453
      raise errors.OpPrereqError("This host's IP resolves to the private"
454
                                 " range (%s). Please fix DNS or %s." %
455
                                 (hostname.ip, constants.ETC_HOSTS))
456

    
457
    if not utils.TcpPing(hostname.ip, constants.DEFAULT_NODED_PORT,
458
                         source=constants.LOCALHOST_IP_ADDRESS):
459
      raise errors.OpPrereqError("Inconsistency: this host's name resolves"
460
                                 " to %s,\nbut this ip address does not"
461
                                 " belong to this host."
462
                                 " Aborting." % hostname.ip)
463

    
464
    self.clustername = clustername = utils.HostInfo(self.op.cluster_name)
465

    
466
    if utils.TcpPing(clustername.ip, constants.DEFAULT_NODED_PORT,
467
                     timeout=5):
468
      raise errors.OpPrereqError("Cluster IP already active. Aborting.")
469

    
470
    secondary_ip = getattr(self.op, "secondary_ip", None)
471
    if secondary_ip and not utils.IsValidIP(secondary_ip):
472
      raise errors.OpPrereqError("Invalid secondary ip given")
473
    if (secondary_ip and
474
        secondary_ip != hostname.ip and
475
        (not utils.TcpPing(secondary_ip, constants.DEFAULT_NODED_PORT,
476
                           source=constants.LOCALHOST_IP_ADDRESS))):
477
      raise errors.OpPrereqError("You gave %s as secondary IP,"
478
                                 " but it does not belong to this host." %
479
                                 secondary_ip)
480
    self.secondary_ip = secondary_ip
481

    
482
    if not hasattr(self.op, "vg_name"):
483
      self.op.vg_name = None
484
    # if vg_name not None, checks if volume group is valid
485
    if self.op.vg_name:
486
      vgstatus = _HasValidVG(utils.ListVolumeGroups(), self.op.vg_name)
487
      if vgstatus:
488
        raise errors.OpPrereqError("Error: %s\nspecify --no-lvm-storage if"
489
                                   " you are not using lvm" % vgstatus)
490

    
491
    self.op.file_storage_dir = os.path.normpath(self.op.file_storage_dir)
492

    
493
    if not os.path.isabs(self.op.file_storage_dir):
494
      raise errors.OpPrereqError("The file storage directory you have is"
495
                                 " not an absolute path.")
496

    
497
    if not os.path.exists(self.op.file_storage_dir):
498
      try:
499
        os.makedirs(self.op.file_storage_dir, 0750)
500
      except OSError, err:
501
        raise errors.OpPrereqError("Cannot create file storage directory"
502
                                   " '%s': %s" %
503
                                   (self.op.file_storage_dir, err))
504

    
505
    if not os.path.isdir(self.op.file_storage_dir):
506
      raise errors.OpPrereqError("The file storage directory '%s' is not"
507
                                 " a directory." % self.op.file_storage_dir)
508

    
509
    if not re.match("^[0-9a-z]{2}:[0-9a-z]{2}:[0-9a-z]{2}$",
510
                    self.op.mac_prefix):
511
      raise errors.OpPrereqError("Invalid mac prefix given '%s'" %
512
                                 self.op.mac_prefix)
513

    
514
    if self.op.hypervisor_type not in constants.HYPER_TYPES:
515
      raise errors.OpPrereqError("Invalid hypervisor type given '%s'" %
516
                                 self.op.hypervisor_type)
517

    
518
    result = utils.RunCmd(["ip", "link", "show", "dev", self.op.master_netdev])
519
    if result.failed:
520
      raise errors.OpPrereqError("Invalid master netdev given (%s): '%s'" %
521
                                 (self.op.master_netdev,
522
                                  result.output.strip()))
523

    
524
    if not (os.path.isfile(constants.NODE_INITD_SCRIPT) and
525
            os.access(constants.NODE_INITD_SCRIPT, os.X_OK)):
526
      raise errors.OpPrereqError("Init.d script '%s' missing or not"
527
                                 " executable." % constants.NODE_INITD_SCRIPT)
528

    
529
  def Exec(self, feedback_fn):
530
    """Initialize the cluster.
531

532
    """
533
    clustername = self.clustername
534
    hostname = self.hostname
535

    
536
    # set up the simple store
537
    self.sstore = ss = ssconf.SimpleStore()
538
    ss.SetKey(ss.SS_HYPERVISOR, self.op.hypervisor_type)
539
    ss.SetKey(ss.SS_MASTER_NODE, hostname.name)
540
    ss.SetKey(ss.SS_MASTER_IP, clustername.ip)
541
    ss.SetKey(ss.SS_MASTER_NETDEV, self.op.master_netdev)
542
    ss.SetKey(ss.SS_CLUSTER_NAME, clustername.name)
543
    ss.SetKey(ss.SS_FILE_STORAGE_DIR, self.op.file_storage_dir)
544
    ss.SetKey(ss.SS_CONFIG_VERSION, constants.CONFIG_VERSION)
545

    
546
    # set up the inter-node password and certificate
547
    _InitGanetiServerSetup(ss)
548

    
549
    # start the master ip
550
    rpc.call_node_start_master(hostname.name)
551

    
552
    # set up ssh config and /etc/hosts
553
    f = open(constants.SSH_HOST_RSA_PUB, 'r')
554
    try:
555
      sshline = f.read()
556
    finally:
557
      f.close()
558
    sshkey = sshline.split(" ")[1]
559

    
560
    _AddHostToEtcHosts(hostname.name)
561
    _InitSSHSetup(hostname.name)
562

    
563
    # init of cluster config file
564
    self.cfg = cfgw = config.ConfigWriter()
565
    cfgw.InitConfig(hostname.name, hostname.ip, self.secondary_ip,
566
                    sshkey, self.op.mac_prefix,
567
                    self.op.vg_name, self.op.def_bridge)
568

    
569
    ssh.WriteKnownHostsFile(cfgw, ss, constants.SSH_KNOWN_HOSTS_FILE)
570

    
571

    
572
class LUDestroyCluster(NoHooksLU):
573
  """Logical unit for destroying the cluster.
574

575
  """
576
  _OP_REQP = []
577

    
578
  def CheckPrereq(self):
579
    """Check prerequisites.
580

581
    This checks whether the cluster is empty.
582

583
    Any errors are signalled by raising errors.OpPrereqError.
584

585
    """
586
    master = self.sstore.GetMasterNode()
587

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

    
597
  def Exec(self, feedback_fn):
598
    """Destroys the cluster.
599

600
    """
601
    master = self.sstore.GetMasterNode()
602
    if not rpc.call_node_stop_master(master):
603
      raise errors.OpExecError("Could not disable the master role")
604
    priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
605
    utils.CreateBackup(priv_key)
606
    utils.CreateBackup(pub_key)
607
    rpc.call_node_leave_cluster(master)
608

    
609

    
610
class LUVerifyCluster(LogicalUnit):
611
  """Verifies the cluster status.
612

613
  """
614
  HPATH = "cluster-verify"
615
  HTYPE = constants.HTYPE_CLUSTER
616
  _OP_REQP = ["skip_checks"]
617

    
618
  def _VerifyNode(self, node, file_list, local_cksum, vglist, node_result,
619
                  remote_version, feedback_fn):
620
    """Run multiple tests against a node.
621

622
    Test list:
623
      - compares ganeti version
624
      - checks vg existance and size > 20G
625
      - checks config file checksum
626
      - checks ssh to other nodes
627

628
    Args:
629
      node: name of the node to check
630
      file_list: required list of files
631
      local_cksum: dictionary of local files and their checksums
632

633
    """
634
    # compares ganeti version
635
    local_version = constants.PROTOCOL_VERSION
636
    if not remote_version:
637
      feedback_fn("  - ERROR: connection to %s failed" % (node))
638
      return True
639

    
640
    if local_version != remote_version:
641
      feedback_fn("  - ERROR: sw version mismatch: master %s, node(%s) %s" %
642
                      (local_version, node, remote_version))
643
      return True
644

    
645
    # checks vg existance and size > 20G
646

    
647
    bad = False
648
    if not vglist:
649
      feedback_fn("  - ERROR: unable to check volume groups on node %s." %
650
                      (node,))
651
      bad = True
652
    else:
653
      vgstatus = _HasValidVG(vglist, self.cfg.GetVGName())
654
      if vgstatus:
655
        feedback_fn("  - ERROR: %s on node %s" % (vgstatus, node))
656
        bad = True
657

    
658
    # checks config file checksum
659
    # checks ssh to any
660

    
661
    if 'filelist' not in node_result:
662
      bad = True
663
      feedback_fn("  - ERROR: node hasn't returned file checksum data")
664
    else:
665
      remote_cksum = node_result['filelist']
666
      for file_name in file_list:
667
        if file_name not in remote_cksum:
668
          bad = True
669
          feedback_fn("  - ERROR: file '%s' missing" % file_name)
670
        elif remote_cksum[file_name] != local_cksum[file_name]:
671
          bad = True
672
          feedback_fn("  - ERROR: file '%s' has wrong checksum" % file_name)
673

    
674
    if 'nodelist' not in node_result:
675
      bad = True
676
      feedback_fn("  - ERROR: node hasn't returned node connectivity data")
677
    else:
678
      if node_result['nodelist']:
679
        bad = True
680
        for node in node_result['nodelist']:
681
          feedback_fn("  - ERROR: communication with node '%s': %s" %
682
                          (node, node_result['nodelist'][node]))
683
    hyp_result = node_result.get('hypervisor', None)
684
    if hyp_result is not None:
685
      feedback_fn("  - ERROR: hypervisor verify failure: '%s'" % hyp_result)
686
    return bad
687

    
688
  def _VerifyInstance(self, instance, instanceconfig, node_vol_is,
689
                      node_instance, feedback_fn):
690
    """Verify an instance.
691

692
    This function checks to see if the required block devices are
693
    available on the instance's node.
694

695
    """
696
    bad = False
697

    
698
    node_current = instanceconfig.primary_node
699

    
700
    node_vol_should = {}
701
    instanceconfig.MapLVsByNode(node_vol_should)
702

    
703
    for node in node_vol_should:
704
      for volume in node_vol_should[node]:
705
        if node not in node_vol_is or volume not in node_vol_is[node]:
706
          feedback_fn("  - ERROR: volume %s missing on node %s" %
707
                          (volume, node))
708
          bad = True
709

    
710
    if not instanceconfig.status == 'down':
711
      if (node_current not in node_instance or
712
          not instance in node_instance[node_current]):
713
        feedback_fn("  - ERROR: instance %s not running on node %s" %
714
                        (instance, node_current))
715
        bad = True
716

    
717
    for node in node_instance:
718
      if (not node == node_current):
719
        if instance in node_instance[node]:
720
          feedback_fn("  - ERROR: instance %s should not run on node %s" %
721
                          (instance, node))
722
          bad = True
723

    
724
    return bad
725

    
726
  def _VerifyOrphanVolumes(self, node_vol_should, node_vol_is, feedback_fn):
727
    """Verify if there are any unknown volumes in the cluster.
728

729
    The .os, .swap and backup volumes are ignored. All other volumes are
730
    reported as unknown.
731

732
    """
733
    bad = False
734

    
735
    for node in node_vol_is:
736
      for volume in node_vol_is[node]:
737
        if node not in node_vol_should or volume not in node_vol_should[node]:
738
          feedback_fn("  - ERROR: volume %s on node %s should not exist" %
739
                      (volume, node))
740
          bad = True
741
    return bad
742

    
743
  def _VerifyOrphanInstances(self, instancelist, node_instance, feedback_fn):
744
    """Verify the list of running instances.
745

746
    This checks what instances are running but unknown to the cluster.
747

748
    """
749
    bad = False
750
    for node in node_instance:
751
      for runninginstance in node_instance[node]:
752
        if runninginstance not in instancelist:
753
          feedback_fn("  - ERROR: instance %s on node %s should not exist" %
754
                          (runninginstance, node))
755
          bad = True
756
    return bad
757

    
758
  def _VerifyNPlusOneMemory(self, node_info, instance_cfg, feedback_fn):
759
    """Verify N+1 Memory Resilience.
760

761
    Check that if one single node dies we can still start all the instances it
762
    was primary for.
763

764
    """
765
    bad = False
766

    
767
    for node, nodeinfo in node_info.iteritems():
768
      # This code checks that every node which is now listed as secondary has
769
      # enough memory to host all instances it is supposed to should a single
770
      # other node in the cluster fail.
771
      # FIXME: not ready for failover to an arbitrary node
772
      # FIXME: does not support file-backed instances
773
      # WARNING: we currently take into account down instances as well as up
774
      # ones, considering that even if they're down someone might want to start
775
      # them even in the event of a node failure.
776
      for prinode, instances in nodeinfo['sinst-by-pnode'].iteritems():
777
        needed_mem = 0
778
        for instance in instances:
779
          needed_mem += instance_cfg[instance].memory
780
        if nodeinfo['mfree'] < needed_mem:
781
          feedback_fn("  - ERROR: not enough memory on node %s to accomodate"
782
                      " failovers should node %s fail" % (node, prinode))
783
          bad = True
784
    return bad
785

    
786
  def CheckPrereq(self):
787
    """Check prerequisites.
788

789
    Transform the list of checks we're going to skip into a set and check that
790
    all its members are valid.
791

792
    """
793
    self.skip_set = frozenset(self.op.skip_checks)
794
    if not constants.VERIFY_OPTIONAL_CHECKS.issuperset(self.skip_set):
795
      raise errors.OpPrereqError("Invalid checks to be skipped specified")
796

    
797
  def BuildHooksEnv(self):
798
    """Build hooks env.
799

800
    Cluster-Verify hooks just rone in the post phase and their failure makes
801
    the output be logged in the verify output and the verification to fail.
802

803
    """
804
    all_nodes = self.cfg.GetNodeList()
805
    # TODO: populate the environment with useful information for verify hooks
806
    env = {}
807
    return env, [], all_nodes
808

    
809
  def Exec(self, feedback_fn):
810
    """Verify integrity of cluster, performing various test on nodes.
811

812
    """
813
    bad = False
814
    feedback_fn("* Verifying global settings")
815
    for msg in self.cfg.VerifyConfig():
816
      feedback_fn("  - ERROR: %s" % msg)
817

    
818
    vg_name = self.cfg.GetVGName()
819
    nodelist = utils.NiceSort(self.cfg.GetNodeList())
820
    instancelist = utils.NiceSort(self.cfg.GetInstanceList())
821
    i_non_redundant = [] # Non redundant instances
822
    node_volume = {}
823
    node_instance = {}
824
    node_info = {}
825
    instance_cfg = {}
826

    
827
    # FIXME: verify OS list
828
    # do local checksums
829
    file_names = list(self.sstore.GetFileList())
830
    file_names.append(constants.SSL_CERT_FILE)
831
    file_names.append(constants.CLUSTER_CONF_FILE)
832
    local_checksums = utils.FingerprintFiles(file_names)
833

    
834
    feedback_fn("* Gathering data (%d nodes)" % len(nodelist))
835
    all_volumeinfo = rpc.call_volume_list(nodelist, vg_name)
836
    all_instanceinfo = rpc.call_instance_list(nodelist)
837
    all_vglist = rpc.call_vg_list(nodelist)
838
    node_verify_param = {
839
      'filelist': file_names,
840
      'nodelist': nodelist,
841
      'hypervisor': None,
842
      }
843
    all_nvinfo = rpc.call_node_verify(nodelist, node_verify_param)
844
    all_rversion = rpc.call_version(nodelist)
845
    all_ninfo = rpc.call_node_info(nodelist, self.cfg.GetVGName())
846

    
847
    for node in nodelist:
848
      feedback_fn("* Verifying node %s" % node)
849
      result = self._VerifyNode(node, file_names, local_checksums,
850
                                all_vglist[node], all_nvinfo[node],
851
                                all_rversion[node], feedback_fn)
852
      bad = bad or result
853

    
854
      # node_volume
855
      volumeinfo = all_volumeinfo[node]
856

    
857
      if isinstance(volumeinfo, basestring):
858
        feedback_fn("  - ERROR: LVM problem on node %s: %s" %
859
                    (node, volumeinfo[-400:].encode('string_escape')))
860
        bad = True
861
        node_volume[node] = {}
862
      elif not isinstance(volumeinfo, dict):
863
        feedback_fn("  - ERROR: connection to %s failed" % (node,))
864
        bad = True
865
        continue
866
      else:
867
        node_volume[node] = volumeinfo
868

    
869
      # node_instance
870
      nodeinstance = all_instanceinfo[node]
871
      if type(nodeinstance) != list:
872
        feedback_fn("  - ERROR: connection to %s failed" % (node,))
873
        bad = True
874
        continue
875

    
876
      node_instance[node] = nodeinstance
877

    
878
      # node_info
879
      nodeinfo = all_ninfo[node]
880
      if not isinstance(nodeinfo, dict):
881
        feedback_fn("  - ERROR: connection to %s failed" % (node,))
882
        bad = True
883
        continue
884

    
885
      try:
886
        node_info[node] = {
887
          "mfree": int(nodeinfo['memory_free']),
888
          "dfree": int(nodeinfo['vg_free']),
889
          "pinst": [],
890
          "sinst": [],
891
          # dictionary holding all instances this node is secondary for,
892
          # grouped by their primary node. Each key is a cluster node, and each
893
          # value is a list of instances which have the key as primary and the
894
          # current node as secondary.  this is handy to calculate N+1 memory
895
          # availability if you can only failover from a primary to its
896
          # secondary.
897
          "sinst-by-pnode": {},
898
        }
899
      except ValueError:
900
        feedback_fn("  - ERROR: invalid value returned from node %s" % (node,))
901
        bad = True
902
        continue
903

    
904
    node_vol_should = {}
905

    
906
    for instance in instancelist:
907
      feedback_fn("* Verifying instance %s" % instance)
908
      inst_config = self.cfg.GetInstanceInfo(instance)
909
      result =  self._VerifyInstance(instance, inst_config, node_volume,
910
                                     node_instance, feedback_fn)
911
      bad = bad or result
912

    
913
      inst_config.MapLVsByNode(node_vol_should)
914

    
915
      instance_cfg[instance] = inst_config
916

    
917
      pnode = inst_config.primary_node
918
      if pnode in node_info:
919
        node_info[pnode]['pinst'].append(instance)
920
      else:
921
        feedback_fn("  - ERROR: instance %s, connection to primary node"
922
                    " %s failed" % (instance, pnode))
923
        bad = True
924

    
925
      # If the instance is non-redundant we cannot survive losing its primary
926
      # node, so we are not N+1 compliant. On the other hand we have no disk
927
      # templates with more than one secondary so that situation is not well
928
      # supported either.
929
      # FIXME: does not support file-backed instances
930
      if len(inst_config.secondary_nodes) == 0:
931
        i_non_redundant.append(instance)
932
      elif len(inst_config.secondary_nodes) > 1:
933
        feedback_fn("  - WARNING: multiple secondaries for instance %s"
934
                    % instance)
935

    
936
      for snode in inst_config.secondary_nodes:
937
        if snode in node_info:
938
          node_info[snode]['sinst'].append(instance)
939
          if pnode not in node_info[snode]['sinst-by-pnode']:
940
            node_info[snode]['sinst-by-pnode'][pnode] = []
941
          node_info[snode]['sinst-by-pnode'][pnode].append(instance)
942
        else:
943
          feedback_fn("  - ERROR: instance %s, connection to secondary node"
944
                      " %s failed" % (instance, snode))
945

    
946
    feedback_fn("* Verifying orphan volumes")
947
    result = self._VerifyOrphanVolumes(node_vol_should, node_volume,
948
                                       feedback_fn)
949
    bad = bad or result
950

    
951
    feedback_fn("* Verifying remaining instances")
952
    result = self._VerifyOrphanInstances(instancelist, node_instance,
953
                                         feedback_fn)
954
    bad = bad or result
955

    
956
    if constants.VERIFY_NPLUSONE_MEM not in self.skip_set:
957
      feedback_fn("* Verifying N+1 Memory redundancy")
958
      result = self._VerifyNPlusOneMemory(node_info, instance_cfg, feedback_fn)
959
      bad = bad or result
960

    
961
    feedback_fn("* Other Notes")
962
    if i_non_redundant:
963
      feedback_fn("  - NOTICE: %d non-redundant instance(s) found."
964
                  % len(i_non_redundant))
965

    
966
    return int(bad)
967

    
968
  def HooksCallBack(self, phase, hooks_results, feedback_fn, lu_result):
969
    """Analize the post-hooks' result, handle it, and send some
970
    nicely-formatted feedback back to the user.
971

972
    Args:
973
      phase: the hooks phase that has just been run
974
      hooks_results: the results of the multi-node hooks rpc call
975
      feedback_fn: function to send feedback back to the caller
976
      lu_result: previous Exec result
977

978
    """
979
    # We only really run POST phase hooks, and are only interested in their results
980
    if phase == constants.HOOKS_PHASE_POST:
981
      # Used to change hooks' output to proper indentation
982
      indent_re = re.compile('^', re.M)
983
      feedback_fn("* Hooks Results")
984
      if not hooks_results:
985
        feedback_fn("  - ERROR: general communication failure")
986
        lu_result = 1
987
      else:
988
        for node_name in hooks_results:
989
          show_node_header = True
990
          res = hooks_results[node_name]
991
          if res is False or not isinstance(res, list):
992
            feedback_fn("    Communication failure")
993
            lu_result = 1
994
            continue
995
          for script, hkr, output in res:
996
            if hkr == constants.HKR_FAIL:
997
              # The node header is only shown once, if there are
998
              # failing hooks on that node
999
              if show_node_header:
1000
                feedback_fn("  Node %s:" % node_name)
1001
                show_node_header = False
1002
              feedback_fn("    ERROR: Script %s failed, output:" % script)
1003
              output = indent_re.sub('      ', output)
1004
              feedback_fn("%s" % output)
1005
              lu_result = 1
1006

    
1007
      return lu_result
1008

    
1009

    
1010
class LUVerifyDisks(NoHooksLU):
1011
  """Verifies the cluster disks status.
1012

1013
  """
1014
  _OP_REQP = []
1015

    
1016
  def CheckPrereq(self):
1017
    """Check prerequisites.
1018

1019
    This has no prerequisites.
1020

1021
    """
1022
    pass
1023

    
1024
  def Exec(self, feedback_fn):
1025
    """Verify integrity of cluster disks.
1026

1027
    """
1028
    result = res_nodes, res_nlvm, res_instances, res_missing = [], {}, [], {}
1029

    
1030
    vg_name = self.cfg.GetVGName()
1031
    nodes = utils.NiceSort(self.cfg.GetNodeList())
1032
    instances = [self.cfg.GetInstanceInfo(name)
1033
                 for name in self.cfg.GetInstanceList()]
1034

    
1035
    nv_dict = {}
1036
    for inst in instances:
1037
      inst_lvs = {}
1038
      if (inst.status != "up" or
1039
          inst.disk_template not in constants.DTS_NET_MIRROR):
1040
        continue
1041
      inst.MapLVsByNode(inst_lvs)
1042
      # transform { iname: {node: [vol,],},} to {(node, vol): iname}
1043
      for node, vol_list in inst_lvs.iteritems():
1044
        for vol in vol_list:
1045
          nv_dict[(node, vol)] = inst
1046

    
1047
    if not nv_dict:
1048
      return result
1049

    
1050
    node_lvs = rpc.call_volume_list(nodes, vg_name)
1051

    
1052
    to_act = set()
1053
    for node in nodes:
1054
      # node_volume
1055
      lvs = node_lvs[node]
1056

    
1057
      if isinstance(lvs, basestring):
1058
        logger.Info("error enumerating LVs on node %s: %s" % (node, lvs))
1059
        res_nlvm[node] = lvs
1060
      elif not isinstance(lvs, dict):
1061
        logger.Info("connection to node %s failed or invalid data returned" %
1062
                    (node,))
1063
        res_nodes.append(node)
1064
        continue
1065

    
1066
      for lv_name, (_, lv_inactive, lv_online) in lvs.iteritems():
1067
        inst = nv_dict.pop((node, lv_name), None)
1068
        if (not lv_online and inst is not None
1069
            and inst.name not in res_instances):
1070
          res_instances.append(inst.name)
1071

    
1072
    # any leftover items in nv_dict are missing LVs, let's arrange the
1073
    # data better
1074
    for key, inst in nv_dict.iteritems():
1075
      if inst.name not in res_missing:
1076
        res_missing[inst.name] = []
1077
      res_missing[inst.name].append(key)
1078

    
1079
    return result
1080

    
1081

    
1082
class LURenameCluster(LogicalUnit):
1083
  """Rename the cluster.
1084

1085
  """
1086
  HPATH = "cluster-rename"
1087
  HTYPE = constants.HTYPE_CLUSTER
1088
  _OP_REQP = ["name"]
1089

    
1090
  def BuildHooksEnv(self):
1091
    """Build hooks env.
1092

1093
    """
1094
    env = {
1095
      "OP_TARGET": self.sstore.GetClusterName(),
1096
      "NEW_NAME": self.op.name,
1097
      }
1098
    mn = self.sstore.GetMasterNode()
1099
    return env, [mn], [mn]
1100

    
1101
  def CheckPrereq(self):
1102
    """Verify that the passed name is a valid one.
1103

1104
    """
1105
    hostname = utils.HostInfo(self.op.name)
1106

    
1107
    new_name = hostname.name
1108
    self.ip = new_ip = hostname.ip
1109
    old_name = self.sstore.GetClusterName()
1110
    old_ip = self.sstore.GetMasterIP()
1111
    if new_name == old_name and new_ip == old_ip:
1112
      raise errors.OpPrereqError("Neither the name nor the IP address of the"
1113
                                 " cluster has changed")
1114
    if new_ip != old_ip:
1115
      result = utils.RunCmd(["fping", "-q", new_ip])
1116
      if not result.failed:
1117
        raise errors.OpPrereqError("The given cluster IP address (%s) is"
1118
                                   " reachable on the network. Aborting." %
1119
                                   new_ip)
1120

    
1121
    self.op.name = new_name
1122

    
1123
  def Exec(self, feedback_fn):
1124
    """Rename the cluster.
1125

1126
    """
1127
    clustername = self.op.name
1128
    ip = self.ip
1129
    ss = self.sstore
1130

    
1131
    # shutdown the master IP
1132
    master = ss.GetMasterNode()
1133
    if not rpc.call_node_stop_master(master):
1134
      raise errors.OpExecError("Could not disable the master role")
1135

    
1136
    try:
1137
      # modify the sstore
1138
      ss.SetKey(ss.SS_MASTER_IP, ip)
1139
      ss.SetKey(ss.SS_CLUSTER_NAME, clustername)
1140

    
1141
      # Distribute updated ss config to all nodes
1142
      myself = self.cfg.GetNodeInfo(master)
1143
      dist_nodes = self.cfg.GetNodeList()
1144
      if myself.name in dist_nodes:
1145
        dist_nodes.remove(myself.name)
1146

    
1147
      logger.Debug("Copying updated ssconf data to all nodes")
1148
      for keyname in [ss.SS_CLUSTER_NAME, ss.SS_MASTER_IP]:
1149
        fname = ss.KeyToFilename(keyname)
1150
        result = rpc.call_upload_file(dist_nodes, fname)
1151
        for to_node in dist_nodes:
1152
          if not result[to_node]:
1153
            logger.Error("copy of file %s to node %s failed" %
1154
                         (fname, to_node))
1155
    finally:
1156
      if not rpc.call_node_start_master(master):
1157
        logger.Error("Could not re-enable the master role on the master,"
1158
                     " please restart manually.")
1159

    
1160

    
1161
def _RecursiveCheckIfLVMBased(disk):
1162
  """Check if the given disk or its children are lvm-based.
1163

1164
  Args:
1165
    disk: ganeti.objects.Disk object
1166

1167
  Returns:
1168
    boolean indicating whether a LD_LV dev_type was found or not
1169

1170
  """
1171
  if disk.children:
1172
    for chdisk in disk.children:
1173
      if _RecursiveCheckIfLVMBased(chdisk):
1174
        return True
1175
  return disk.dev_type == constants.LD_LV
1176

    
1177

    
1178
class LUSetClusterParams(LogicalUnit):
1179
  """Change the parameters of the cluster.
1180

1181
  """
1182
  HPATH = "cluster-modify"
1183
  HTYPE = constants.HTYPE_CLUSTER
1184
  _OP_REQP = []
1185

    
1186
  def BuildHooksEnv(self):
1187
    """Build hooks env.
1188

1189
    """
1190
    env = {
1191
      "OP_TARGET": self.sstore.GetClusterName(),
1192
      "NEW_VG_NAME": self.op.vg_name,
1193
      }
1194
    mn = self.sstore.GetMasterNode()
1195
    return env, [mn], [mn]
1196

    
1197
  def CheckPrereq(self):
1198
    """Check prerequisites.
1199

1200
    This checks whether the given params don't conflict and
1201
    if the given volume group is valid.
1202

1203
    """
1204
    if not self.op.vg_name:
1205
      instances = [self.cfg.GetInstanceInfo(name)
1206
                   for name in self.cfg.GetInstanceList()]
1207
      for inst in instances:
1208
        for disk in inst.disks:
1209
          if _RecursiveCheckIfLVMBased(disk):
1210
            raise errors.OpPrereqError("Cannot disable lvm storage while"
1211
                                       " lvm-based instances exist")
1212

    
1213
    # if vg_name not None, checks given volume group on all nodes
1214
    if self.op.vg_name:
1215
      node_list = self.cfg.GetNodeList()
1216
      vglist = rpc.call_vg_list(node_list)
1217
      for node in node_list:
1218
        vgstatus = _HasValidVG(vglist[node], self.op.vg_name)
1219
        if vgstatus:
1220
          raise errors.OpPrereqError("Error on node '%s': %s" %
1221
                                     (node, vgstatus))
1222

    
1223
  def Exec(self, feedback_fn):
1224
    """Change the parameters of the cluster.
1225

1226
    """
1227
    if self.op.vg_name != self.cfg.GetVGName():
1228
      self.cfg.SetVGName(self.op.vg_name)
1229
    else:
1230
      feedback_fn("Cluster LVM configuration already in desired"
1231
                  " state, not changing")
1232

    
1233

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

1237
  """
1238
  if not instance.disks:
1239
    return True
1240

    
1241
  if not oneshot:
1242
    proc.LogInfo("Waiting for instance %s to sync disks." % instance.name)
1243

    
1244
  node = instance.primary_node
1245

    
1246
  for dev in instance.disks:
1247
    cfgw.SetDiskID(dev, node)
1248

    
1249
  retries = 0
1250
  while True:
1251
    max_time = 0
1252
    done = True
1253
    cumul_degraded = False
1254
    rstats = rpc.call_blockdev_getmirrorstatus(node, instance.disks)
1255
    if not rstats:
1256
      proc.LogWarning("Can't get any data from node %s" % node)
1257
      retries += 1
1258
      if retries >= 10:
1259
        raise errors.RemoteError("Can't contact node %s for mirror data,"
1260
                                 " aborting." % node)
1261
      time.sleep(6)
1262
      continue
1263
    retries = 0
1264
    for i in range(len(rstats)):
1265
      mstat = rstats[i]
1266
      if mstat is None:
1267
        proc.LogWarning("Can't compute data for node %s/%s" %
1268
                        (node, instance.disks[i].iv_name))
1269
        continue
1270
      # we ignore the ldisk parameter
1271
      perc_done, est_time, is_degraded, _ = mstat
1272
      cumul_degraded = cumul_degraded or (is_degraded and perc_done is None)
1273
      if perc_done is not None:
1274
        done = False
1275
        if est_time is not None:
1276
          rem_time = "%d estimated seconds remaining" % est_time
1277
          max_time = est_time
1278
        else:
1279
          rem_time = "no time estimate"
1280
        proc.LogInfo("- device %s: %5.2f%% done, %s" %
1281
                     (instance.disks[i].iv_name, perc_done, rem_time))
1282
    if done or oneshot:
1283
      break
1284

    
1285
    if unlock:
1286
      #utils.Unlock('cmd')
1287
      pass
1288
    try:
1289
      time.sleep(min(60, max_time))
1290
    finally:
1291
      if unlock:
1292
        #utils.Lock('cmd')
1293
        pass
1294

    
1295
  if done:
1296
    proc.LogInfo("Instance %s's disks are in sync." % instance.name)
1297
  return not cumul_degraded
1298

    
1299

    
1300
def _CheckDiskConsistency(cfgw, dev, node, on_primary, ldisk=False):
1301
  """Check that mirrors are not degraded.
1302

1303
  The ldisk parameter, if True, will change the test from the
1304
  is_degraded attribute (which represents overall non-ok status for
1305
  the device(s)) to the ldisk (representing the local storage status).
1306

1307
  """
1308
  cfgw.SetDiskID(dev, node)
1309
  if ldisk:
1310
    idx = 6
1311
  else:
1312
    idx = 5
1313

    
1314
  result = True
1315
  if on_primary or dev.AssembleOnSecondary():
1316
    rstats = rpc.call_blockdev_find(node, dev)
1317
    if not rstats:
1318
      logger.ToStderr("Node %s: Disk degraded, not found or node down" % node)
1319
      result = False
1320
    else:
1321
      result = result and (not rstats[idx])
1322
  if dev.children:
1323
    for child in dev.children:
1324
      result = result and _CheckDiskConsistency(cfgw, child, node, on_primary)
1325

    
1326
  return result
1327

    
1328

    
1329
class LUDiagnoseOS(NoHooksLU):
1330
  """Logical unit for OS diagnose/query.
1331

1332
  """
1333
  _OP_REQP = ["output_fields", "names"]
1334

    
1335
  def CheckPrereq(self):
1336
    """Check prerequisites.
1337

1338
    This always succeeds, since this is a pure query LU.
1339

1340
    """
1341
    if self.op.names:
1342
      raise errors.OpPrereqError("Selective OS query not supported")
1343

    
1344
    self.dynamic_fields = frozenset(["name", "valid", "node_status"])
1345
    _CheckOutputFields(static=[],
1346
                       dynamic=self.dynamic_fields,
1347
                       selected=self.op.output_fields)
1348

    
1349
  @staticmethod
1350
  def _DiagnoseByOS(node_list, rlist):
1351
    """Remaps a per-node return list into an a per-os per-node dictionary
1352

1353
      Args:
1354
        node_list: a list with the names of all nodes
1355
        rlist: a map with node names as keys and OS objects as values
1356

1357
      Returns:
1358
        map: a map with osnames as keys and as value another map, with
1359
             nodes as
1360
             keys and list of OS objects as values
1361
             e.g. {"debian-etch": {"node1": [<object>,...],
1362
                                   "node2": [<object>,]}
1363
                  }
1364

1365
    """
1366
    all_os = {}
1367
    for node_name, nr in rlist.iteritems():
1368
      if not nr:
1369
        continue
1370
      for os_obj in nr:
1371
        if os_obj.name not in all_os:
1372
          # build a list of nodes for this os containing empty lists
1373
          # for each node in node_list
1374
          all_os[os_obj.name] = {}
1375
          for nname in node_list:
1376
            all_os[os_obj.name][nname] = []
1377
        all_os[os_obj.name][node_name].append(os_obj)
1378
    return all_os
1379

    
1380
  def Exec(self, feedback_fn):
1381
    """Compute the list of OSes.
1382

1383
    """
1384
    node_list = self.cfg.GetNodeList()
1385
    node_data = rpc.call_os_diagnose(node_list)
1386
    if node_data == False:
1387
      raise errors.OpExecError("Can't gather the list of OSes")
1388
    pol = self._DiagnoseByOS(node_list, node_data)
1389
    output = []
1390
    for os_name, os_data in pol.iteritems():
1391
      row = []
1392
      for field in self.op.output_fields:
1393
        if field == "name":
1394
          val = os_name
1395
        elif field == "valid":
1396
          val = utils.all([osl and osl[0] for osl in os_data.values()])
1397
        elif field == "node_status":
1398
          val = {}
1399
          for node_name, nos_list in os_data.iteritems():
1400
            val[node_name] = [(v.status, v.path) for v in nos_list]
1401
        else:
1402
          raise errors.ParameterError(field)
1403
        row.append(val)
1404
      output.append(row)
1405

    
1406
    return output
1407

    
1408

    
1409
class LURemoveNode(LogicalUnit):
1410
  """Logical unit for removing a node.
1411

1412
  """
1413
  HPATH = "node-remove"
1414
  HTYPE = constants.HTYPE_NODE
1415
  _OP_REQP = ["node_name"]
1416

    
1417
  def BuildHooksEnv(self):
1418
    """Build hooks env.
1419

1420
    This doesn't run on the target node in the pre phase as a failed
1421
    node would not allows itself to run.
1422

1423
    """
1424
    env = {
1425
      "OP_TARGET": self.op.node_name,
1426
      "NODE_NAME": self.op.node_name,
1427
      }
1428
    all_nodes = self.cfg.GetNodeList()
1429
    all_nodes.remove(self.op.node_name)
1430
    return env, all_nodes, all_nodes
1431

    
1432
  def CheckPrereq(self):
1433
    """Check prerequisites.
1434

1435
    This checks:
1436
     - the node exists in the configuration
1437
     - it does not have primary or secondary instances
1438
     - it's not the master
1439

1440
    Any errors are signalled by raising errors.OpPrereqError.
1441

1442
    """
1443
    node = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.node_name))
1444
    if node is None:
1445
      raise errors.OpPrereqError, ("Node '%s' is unknown." % self.op.node_name)
1446

    
1447
    instance_list = self.cfg.GetInstanceList()
1448

    
1449
    masternode = self.sstore.GetMasterNode()
1450
    if node.name == masternode:
1451
      raise errors.OpPrereqError("Node is the master node,"
1452
                                 " you need to failover first.")
1453

    
1454
    for instance_name in instance_list:
1455
      instance = self.cfg.GetInstanceInfo(instance_name)
1456
      if node.name == instance.primary_node:
1457
        raise errors.OpPrereqError("Instance %s still running on the node,"
1458
                                   " please remove first." % instance_name)
1459
      if node.name in instance.secondary_nodes:
1460
        raise errors.OpPrereqError("Instance %s has node as a secondary,"
1461
                                   " please remove first." % instance_name)
1462
    self.op.node_name = node.name
1463
    self.node = node
1464

    
1465
  def Exec(self, feedback_fn):
1466
    """Removes the node from the cluster.
1467

1468
    """
1469
    node = self.node
1470
    logger.Info("stopping the node daemon and removing configs from node %s" %
1471
                node.name)
1472

    
1473
    rpc.call_node_leave_cluster(node.name)
1474

    
1475
    self.ssh.Run(node.name, 'root', "%s stop" % constants.NODE_INITD_SCRIPT)
1476

    
1477
    logger.Info("Removing node %s from config" % node.name)
1478

    
1479
    self.cfg.RemoveNode(node.name)
1480

    
1481
    _RemoveHostFromEtcHosts(node.name)
1482

    
1483

    
1484
class LUQueryNodes(NoHooksLU):
1485
  """Logical unit for querying nodes.
1486

1487
  """
1488
  _OP_REQP = ["output_fields", "names"]
1489

    
1490
  def CheckPrereq(self):
1491
    """Check prerequisites.
1492

1493
    This checks that the fields required are valid output fields.
1494

1495
    """
1496
    self.dynamic_fields = frozenset([
1497
      "dtotal", "dfree",
1498
      "mtotal", "mnode", "mfree",
1499
      "bootid",
1500
      "ctotal",
1501
      ])
1502

    
1503
    _CheckOutputFields(static=["name", "pinst_cnt", "sinst_cnt",
1504
                               "pinst_list", "sinst_list",
1505
                               "pip", "sip"],
1506
                       dynamic=self.dynamic_fields,
1507
                       selected=self.op.output_fields)
1508

    
1509
    self.wanted = _GetWantedNodes(self, self.op.names)
1510

    
1511
  def Exec(self, feedback_fn):
1512
    """Computes the list of nodes and their attributes.
1513

1514
    """
1515
    nodenames = self.wanted
1516
    nodelist = [self.cfg.GetNodeInfo(name) for name in nodenames]
1517

    
1518
    # begin data gathering
1519

    
1520
    if self.dynamic_fields.intersection(self.op.output_fields):
1521
      live_data = {}
1522
      node_data = rpc.call_node_info(nodenames, self.cfg.GetVGName())
1523
      for name in nodenames:
1524
        nodeinfo = node_data.get(name, None)
1525
        if nodeinfo:
1526
          live_data[name] = {
1527
            "mtotal": utils.TryConvert(int, nodeinfo['memory_total']),
1528
            "mnode": utils.TryConvert(int, nodeinfo['memory_dom0']),
1529
            "mfree": utils.TryConvert(int, nodeinfo['memory_free']),
1530
            "dtotal": utils.TryConvert(int, nodeinfo['vg_size']),
1531
            "dfree": utils.TryConvert(int, nodeinfo['vg_free']),
1532
            "ctotal": utils.TryConvert(int, nodeinfo['cpu_total']),
1533
            "bootid": nodeinfo['bootid'],
1534
            }
1535
        else:
1536
          live_data[name] = {}
1537
    else:
1538
      live_data = dict.fromkeys(nodenames, {})
1539

    
1540
    node_to_primary = dict([(name, set()) for name in nodenames])
1541
    node_to_secondary = dict([(name, set()) for name in nodenames])
1542

    
1543
    inst_fields = frozenset(("pinst_cnt", "pinst_list",
1544
                             "sinst_cnt", "sinst_list"))
1545
    if inst_fields & frozenset(self.op.output_fields):
1546
      instancelist = self.cfg.GetInstanceList()
1547

    
1548
      for instance_name in instancelist:
1549
        inst = self.cfg.GetInstanceInfo(instance_name)
1550
        if inst.primary_node in node_to_primary:
1551
          node_to_primary[inst.primary_node].add(inst.name)
1552
        for secnode in inst.secondary_nodes:
1553
          if secnode in node_to_secondary:
1554
            node_to_secondary[secnode].add(inst.name)
1555

    
1556
    # end data gathering
1557

    
1558
    output = []
1559
    for node in nodelist:
1560
      node_output = []
1561
      for field in self.op.output_fields:
1562
        if field == "name":
1563
          val = node.name
1564
        elif field == "pinst_list":
1565
          val = list(node_to_primary[node.name])
1566
        elif field == "sinst_list":
1567
          val = list(node_to_secondary[node.name])
1568
        elif field == "pinst_cnt":
1569
          val = len(node_to_primary[node.name])
1570
        elif field == "sinst_cnt":
1571
          val = len(node_to_secondary[node.name])
1572
        elif field == "pip":
1573
          val = node.primary_ip
1574
        elif field == "sip":
1575
          val = node.secondary_ip
1576
        elif field in self.dynamic_fields:
1577
          val = live_data[node.name].get(field, None)
1578
        else:
1579
          raise errors.ParameterError(field)
1580
        node_output.append(val)
1581
      output.append(node_output)
1582

    
1583
    return output
1584

    
1585

    
1586
class LUQueryNodeVolumes(NoHooksLU):
1587
  """Logical unit for getting volumes on node(s).
1588

1589
  """
1590
  _OP_REQP = ["nodes", "output_fields"]
1591

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

1595
    This checks that the fields required are valid output fields.
1596

1597
    """
1598
    self.nodes = _GetWantedNodes(self, self.op.nodes)
1599

    
1600
    _CheckOutputFields(static=["node"],
1601
                       dynamic=["phys", "vg", "name", "size", "instance"],
1602
                       selected=self.op.output_fields)
1603

    
1604

    
1605
  def Exec(self, feedback_fn):
1606
    """Computes the list of nodes and their attributes.
1607

1608
    """
1609
    nodenames = self.nodes
1610
    volumes = rpc.call_node_volumes(nodenames)
1611

    
1612
    ilist = [self.cfg.GetInstanceInfo(iname) for iname
1613
             in self.cfg.GetInstanceList()]
1614

    
1615
    lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
1616

    
1617
    output = []
1618
    for node in nodenames:
1619
      if node not in volumes or not volumes[node]:
1620
        continue
1621

    
1622
      node_vols = volumes[node][:]
1623
      node_vols.sort(key=lambda vol: vol['dev'])
1624

    
1625
      for vol in node_vols:
1626
        node_output = []
1627
        for field in self.op.output_fields:
1628
          if field == "node":
1629
            val = node
1630
          elif field == "phys":
1631
            val = vol['dev']
1632
          elif field == "vg":
1633
            val = vol['vg']
1634
          elif field == "name":
1635
            val = vol['name']
1636
          elif field == "size":
1637
            val = int(float(vol['size']))
1638
          elif field == "instance":
1639
            for inst in ilist:
1640
              if node not in lv_by_node[inst]:
1641
                continue
1642
              if vol['name'] in lv_by_node[inst][node]:
1643
                val = inst.name
1644
                break
1645
            else:
1646
              val = '-'
1647
          else:
1648
            raise errors.ParameterError(field)
1649
          node_output.append(str(val))
1650

    
1651
        output.append(node_output)
1652

    
1653
    return output
1654

    
1655

    
1656
class LUAddNode(LogicalUnit):
1657
  """Logical unit for adding node to the cluster.
1658

1659
  """
1660
  HPATH = "node-add"
1661
  HTYPE = constants.HTYPE_NODE
1662
  _OP_REQP = ["node_name"]
1663

    
1664
  def BuildHooksEnv(self):
1665
    """Build hooks env.
1666

1667
    This will run on all nodes before, and on all nodes + the new node after.
1668

1669
    """
1670
    env = {
1671
      "OP_TARGET": self.op.node_name,
1672
      "NODE_NAME": self.op.node_name,
1673
      "NODE_PIP": self.op.primary_ip,
1674
      "NODE_SIP": self.op.secondary_ip,
1675
      }
1676
    nodes_0 = self.cfg.GetNodeList()
1677
    nodes_1 = nodes_0 + [self.op.node_name, ]
1678
    return env, nodes_0, nodes_1
1679

    
1680
  def CheckPrereq(self):
1681
    """Check prerequisites.
1682

1683
    This checks:
1684
     - the new node is not already in the config
1685
     - it is resolvable
1686
     - its parameters (single/dual homed) matches the cluster
1687

1688
    Any errors are signalled by raising errors.OpPrereqError.
1689

1690
    """
1691
    node_name = self.op.node_name
1692
    cfg = self.cfg
1693

    
1694
    dns_data = utils.HostInfo(node_name)
1695

    
1696
    node = dns_data.name
1697
    primary_ip = self.op.primary_ip = dns_data.ip
1698
    secondary_ip = getattr(self.op, "secondary_ip", None)
1699
    if secondary_ip is None:
1700
      secondary_ip = primary_ip
1701
    if not utils.IsValidIP(secondary_ip):
1702
      raise errors.OpPrereqError("Invalid secondary IP given")
1703
    self.op.secondary_ip = secondary_ip
1704

    
1705
    node_list = cfg.GetNodeList()
1706
    if not self.op.readd and node in node_list:
1707
      raise errors.OpPrereqError("Node %s is already in the configuration" %
1708
                                 node)
1709
    elif self.op.readd and node not in node_list:
1710
      raise errors.OpPrereqError("Node %s is not in the configuration" % node)
1711

    
1712
    for existing_node_name in node_list:
1713
      existing_node = cfg.GetNodeInfo(existing_node_name)
1714

    
1715
      if self.op.readd and node == existing_node_name:
1716
        if (existing_node.primary_ip != primary_ip or
1717
            existing_node.secondary_ip != secondary_ip):
1718
          raise errors.OpPrereqError("Readded node doesn't have the same IP"
1719
                                     " address configuration as before")
1720
        continue
1721

    
1722
      if (existing_node.primary_ip == primary_ip or
1723
          existing_node.secondary_ip == primary_ip or
1724
          existing_node.primary_ip == secondary_ip or
1725
          existing_node.secondary_ip == secondary_ip):
1726
        raise errors.OpPrereqError("New node ip address(es) conflict with"
1727
                                   " existing node %s" % existing_node.name)
1728

    
1729
    # check that the type of the node (single versus dual homed) is the
1730
    # same as for the master
1731
    myself = cfg.GetNodeInfo(self.sstore.GetMasterNode())
1732
    master_singlehomed = myself.secondary_ip == myself.primary_ip
1733
    newbie_singlehomed = secondary_ip == primary_ip
1734
    if master_singlehomed != newbie_singlehomed:
1735
      if master_singlehomed:
1736
        raise errors.OpPrereqError("The master has no private ip but the"
1737
                                   " new node has one")
1738
      else:
1739
        raise errors.OpPrereqError("The master has a private ip but the"
1740
                                   " new node doesn't have one")
1741

    
1742
    # checks reachablity
1743
    if not utils.TcpPing(primary_ip, constants.DEFAULT_NODED_PORT):
1744
      raise errors.OpPrereqError("Node not reachable by ping")
1745

    
1746
    if not newbie_singlehomed:
1747
      # check reachability from my secondary ip to newbie's secondary ip
1748
      if not utils.TcpPing(secondary_ip, constants.DEFAULT_NODED_PORT,
1749
                           source=myself.secondary_ip):
1750
        raise errors.OpPrereqError("Node secondary ip not reachable by TCP"
1751
                                   " based ping to noded port")
1752

    
1753
    self.new_node = objects.Node(name=node,
1754
                                 primary_ip=primary_ip,
1755
                                 secondary_ip=secondary_ip)
1756

    
1757
    if self.sstore.GetHypervisorType() == constants.HT_XEN_HVM31:
1758
      if not os.path.exists(constants.VNC_PASSWORD_FILE):
1759
        raise errors.OpPrereqError("Cluster VNC password file %s missing" %
1760
                                   constants.VNC_PASSWORD_FILE)
1761

    
1762
  def Exec(self, feedback_fn):
1763
    """Adds the new node to the cluster.
1764

1765
    """
1766
    new_node = self.new_node
1767
    node = new_node.name
1768

    
1769
    # set up inter-node password and certificate and restarts the node daemon
1770
    gntpass = self.sstore.GetNodeDaemonPassword()
1771
    if not re.match('^[a-zA-Z0-9.]{1,64}$', gntpass):
1772
      raise errors.OpExecError("ganeti password corruption detected")
1773
    f = open(constants.SSL_CERT_FILE)
1774
    try:
1775
      gntpem = f.read(8192)
1776
    finally:
1777
      f.close()
1778
    # in the base64 pem encoding, neither '!' nor '.' are valid chars,
1779
    # so we use this to detect an invalid certificate; as long as the
1780
    # cert doesn't contain this, the here-document will be correctly
1781
    # parsed by the shell sequence below
1782
    if re.search('^!EOF\.', gntpem, re.MULTILINE):
1783
      raise errors.OpExecError("invalid PEM encoding in the SSL certificate")
1784
    if not gntpem.endswith("\n"):
1785
      raise errors.OpExecError("PEM must end with newline")
1786
    logger.Info("copy cluster pass to %s and starting the node daemon" % node)
1787

    
1788
    # and then connect with ssh to set password and start ganeti-noded
1789
    # note that all the below variables are sanitized at this point,
1790
    # either by being constants or by the checks above
1791
    ss = self.sstore
1792
    mycommand = ("umask 077 && "
1793
                 "echo '%s' > '%s' && "
1794
                 "cat > '%s' << '!EOF.' && \n"
1795
                 "%s!EOF.\n%s restart" %
1796
                 (gntpass, ss.KeyToFilename(ss.SS_NODED_PASS),
1797
                  constants.SSL_CERT_FILE, gntpem,
1798
                  constants.NODE_INITD_SCRIPT))
1799

    
1800
    result = self.ssh.Run(node, 'root', mycommand, batch=False, ask_key=True)
1801
    if result.failed:
1802
      raise errors.OpExecError("Remote command on node %s, error: %s,"
1803
                               " output: %s" %
1804
                               (node, result.fail_reason, result.output))
1805

    
1806
    # check connectivity
1807
    time.sleep(4)
1808

    
1809
    result = rpc.call_version([node])[node]
1810
    if result:
1811
      if constants.PROTOCOL_VERSION == result:
1812
        logger.Info("communication to node %s fine, sw version %s match" %
1813
                    (node, result))
1814
      else:
1815
        raise errors.OpExecError("Version mismatch master version %s,"
1816
                                 " node version %s" %
1817
                                 (constants.PROTOCOL_VERSION, result))
1818
    else:
1819
      raise errors.OpExecError("Cannot get version from the new node")
1820

    
1821
    # setup ssh on node
1822
    logger.Info("copy ssh key to node %s" % node)
1823
    priv_key, pub_key, _ = ssh.GetUserFiles(constants.GANETI_RUNAS)
1824
    keyarray = []
1825
    keyfiles = [constants.SSH_HOST_DSA_PRIV, constants.SSH_HOST_DSA_PUB,
1826
                constants.SSH_HOST_RSA_PRIV, constants.SSH_HOST_RSA_PUB,
1827
                priv_key, pub_key]
1828

    
1829
    for i in keyfiles:
1830
      f = open(i, 'r')
1831
      try:
1832
        keyarray.append(f.read())
1833
      finally:
1834
        f.close()
1835

    
1836
    result = rpc.call_node_add(node, keyarray[0], keyarray[1], keyarray[2],
1837
                               keyarray[3], keyarray[4], keyarray[5])
1838

    
1839
    if not result:
1840
      raise errors.OpExecError("Cannot transfer ssh keys to the new node")
1841

    
1842
    # Add node to our /etc/hosts, and add key to known_hosts
1843
    _AddHostToEtcHosts(new_node.name)
1844

    
1845
    if new_node.secondary_ip != new_node.primary_ip:
1846
      if not rpc.call_node_tcp_ping(new_node.name,
1847
                                    constants.LOCALHOST_IP_ADDRESS,
1848
                                    new_node.secondary_ip,
1849
                                    constants.DEFAULT_NODED_PORT,
1850
                                    10, False):
1851
        raise errors.OpExecError("Node claims it doesn't have the secondary ip"
1852
                                 " you gave (%s). Please fix and re-run this"
1853
                                 " command." % new_node.secondary_ip)
1854

    
1855
    success, msg = self.ssh.VerifyNodeHostname(node)
1856
    if not success:
1857
      raise errors.OpExecError("Node '%s' claims it has a different hostname"
1858
                               " than the one the resolver gives: %s."
1859
                               " Please fix and re-run this command." %
1860
                               (node, msg))
1861

    
1862
    # Distribute updated /etc/hosts and known_hosts to all nodes,
1863
    # including the node just added
1864
    myself = self.cfg.GetNodeInfo(self.sstore.GetMasterNode())
1865
    dist_nodes = self.cfg.GetNodeList() + [node]
1866
    if myself.name in dist_nodes:
1867
      dist_nodes.remove(myself.name)
1868

    
1869
    logger.Debug("Copying hosts and known_hosts to all nodes")
1870
    for fname in (constants.ETC_HOSTS, constants.SSH_KNOWN_HOSTS_FILE):
1871
      result = rpc.call_upload_file(dist_nodes, fname)
1872
      for to_node in dist_nodes:
1873
        if not result[to_node]:
1874
          logger.Error("copy of file %s to node %s failed" %
1875
                       (fname, to_node))
1876

    
1877
    to_copy = ss.GetFileList()
1878
    if self.sstore.GetHypervisorType() == constants.HT_XEN_HVM31:
1879
      to_copy.append(constants.VNC_PASSWORD_FILE)
1880
    for fname in to_copy:
1881
      if not self.ssh.CopyFileToNode(node, fname):
1882
        logger.Error("could not copy file %s to node %s" % (fname, node))
1883

    
1884
    if not self.op.readd:
1885
      logger.Info("adding node %s to cluster.conf" % node)
1886
      self.cfg.AddNode(new_node)
1887

    
1888

    
1889
class LUMasterFailover(LogicalUnit):
1890
  """Failover the master node to the current node.
1891

1892
  This is a special LU in that it must run on a non-master node.
1893

1894
  """
1895
  HPATH = "master-failover"
1896
  HTYPE = constants.HTYPE_CLUSTER
1897
  REQ_MASTER = False
1898
  _OP_REQP = []
1899

    
1900
  def BuildHooksEnv(self):
1901
    """Build hooks env.
1902

1903
    This will run on the new master only in the pre phase, and on all
1904
    the nodes in the post phase.
1905

1906
    """
1907
    env = {
1908
      "OP_TARGET": self.new_master,
1909
      "NEW_MASTER": self.new_master,
1910
      "OLD_MASTER": self.old_master,
1911
      }
1912
    return env, [self.new_master], self.cfg.GetNodeList()
1913

    
1914
  def CheckPrereq(self):
1915
    """Check prerequisites.
1916

1917
    This checks that we are not already the master.
1918

1919
    """
1920
    self.new_master = utils.HostInfo().name
1921
    self.old_master = self.sstore.GetMasterNode()
1922

    
1923
    if self.old_master == self.new_master:
1924
      raise errors.OpPrereqError("This commands must be run on the node"
1925
                                 " where you want the new master to be."
1926
                                 " %s is already the master" %
1927
                                 self.old_master)
1928

    
1929
  def Exec(self, feedback_fn):
1930
    """Failover the master node.
1931

1932
    This command, when run on a non-master node, will cause the current
1933
    master to cease being master, and the non-master to become new
1934
    master.
1935

1936
    """
1937
    #TODO: do not rely on gethostname returning the FQDN
1938
    logger.Info("setting master to %s, old master: %s" %
1939
                (self.new_master, self.old_master))
1940

    
1941
    if not rpc.call_node_stop_master(self.old_master):
1942
      logger.Error("could disable the master role on the old master"
1943
                   " %s, please disable manually" % self.old_master)
1944

    
1945
    ss = self.sstore
1946
    ss.SetKey(ss.SS_MASTER_NODE, self.new_master)
1947
    if not rpc.call_upload_file(self.cfg.GetNodeList(),
1948
                                ss.KeyToFilename(ss.SS_MASTER_NODE)):
1949
      logger.Error("could not distribute the new simple store master file"
1950
                   " to the other nodes, please check.")
1951

    
1952
    if not rpc.call_node_start_master(self.new_master):
1953
      logger.Error("could not start the master role on the new master"
1954
                   " %s, please check" % self.new_master)
1955
      feedback_fn("Error in activating the master IP on the new master,"
1956
                  " please fix manually.")
1957

    
1958

    
1959

    
1960
class LUQueryClusterInfo(NoHooksLU):
1961
  """Query cluster configuration.
1962

1963
  """
1964
  _OP_REQP = []
1965
  REQ_MASTER = False
1966

    
1967
  def CheckPrereq(self):
1968
    """No prerequsites needed for this LU.
1969

1970
    """
1971
    pass
1972

    
1973
  def Exec(self, feedback_fn):
1974
    """Return cluster config.
1975

1976
    """
1977
    result = {
1978
      "name": self.sstore.GetClusterName(),
1979
      "software_version": constants.RELEASE_VERSION,
1980
      "protocol_version": constants.PROTOCOL_VERSION,
1981
      "config_version": constants.CONFIG_VERSION,
1982
      "os_api_version": constants.OS_API_VERSION,
1983
      "export_version": constants.EXPORT_VERSION,
1984
      "master": self.sstore.GetMasterNode(),
1985
      "architecture": (platform.architecture()[0], platform.machine()),
1986
      }
1987

    
1988
    return result
1989

    
1990

    
1991
class LUClusterCopyFile(NoHooksLU):
1992
  """Copy file to cluster.
1993

1994
  """
1995
  _OP_REQP = ["nodes", "filename"]
1996

    
1997
  def CheckPrereq(self):
1998
    """Check prerequisites.
1999

2000
    It should check that the named file exists and that the given list
2001
    of nodes is valid.
2002

2003
    """
2004
    if not os.path.exists(self.op.filename):
2005
      raise errors.OpPrereqError("No such filename '%s'" % self.op.filename)
2006

    
2007
    self.nodes = _GetWantedNodes(self, self.op.nodes)
2008

    
2009
  def Exec(self, feedback_fn):
2010
    """Copy a file from master to some nodes.
2011

2012
    Args:
2013
      opts - class with options as members
2014
      args - list containing a single element, the file name
2015
    Opts used:
2016
      nodes - list containing the name of target nodes; if empty, all nodes
2017

2018
    """
2019
    filename = self.op.filename
2020

    
2021
    myname = utils.HostInfo().name
2022

    
2023
    for node in self.nodes:
2024
      if node == myname:
2025
        continue
2026
      if not self.ssh.CopyFileToNode(node, filename):
2027
        logger.Error("Copy of file %s to node %s failed" % (filename, node))
2028

    
2029

    
2030
class LUDumpClusterConfig(NoHooksLU):
2031
  """Return a text-representation of the cluster-config.
2032

2033
  """
2034
  _OP_REQP = []
2035

    
2036
  def CheckPrereq(self):
2037
    """No prerequisites.
2038

2039
    """
2040
    pass
2041

    
2042
  def Exec(self, feedback_fn):
2043
    """Dump a representation of the cluster config to the standard output.
2044

2045
    """
2046
    return self.cfg.DumpConfig()
2047

    
2048

    
2049
class LURunClusterCommand(NoHooksLU):
2050
  """Run a command on some nodes.
2051

2052
  """
2053
  _OP_REQP = ["command", "nodes"]
2054

    
2055
  def CheckPrereq(self):
2056
    """Check prerequisites.
2057

2058
    It checks that the given list of nodes is valid.
2059

2060
    """
2061
    self.nodes = _GetWantedNodes(self, self.op.nodes)
2062

    
2063
  def Exec(self, feedback_fn):
2064
    """Run a command on some nodes.
2065

2066
    """
2067
    # put the master at the end of the nodes list
2068
    master_node = self.sstore.GetMasterNode()
2069
    if master_node in self.nodes:
2070
      self.nodes.remove(master_node)
2071
      self.nodes.append(master_node)
2072

    
2073
    data = []
2074
    for node in self.nodes:
2075
      result = self.ssh.Run(node, "root", self.op.command)
2076
      data.append((node, result.output, result.exit_code))
2077

    
2078
    return data
2079

    
2080

    
2081
class LUActivateInstanceDisks(NoHooksLU):
2082
  """Bring up an instance's disks.
2083

2084
  """
2085
  _OP_REQP = ["instance_name"]
2086

    
2087
  def CheckPrereq(self):
2088
    """Check prerequisites.
2089

2090
    This checks that the instance is in the cluster.
2091

2092
    """
2093
    instance = self.cfg.GetInstanceInfo(
2094
      self.cfg.ExpandInstanceName(self.op.instance_name))
2095
    if instance is None:
2096
      raise errors.OpPrereqError("Instance '%s' not known" %
2097
                                 self.op.instance_name)
2098
    self.instance = instance
2099

    
2100

    
2101
  def Exec(self, feedback_fn):
2102
    """Activate the disks.
2103

2104
    """
2105
    disks_ok, disks_info = _AssembleInstanceDisks(self.instance, self.cfg)
2106
    if not disks_ok:
2107
      raise errors.OpExecError("Cannot activate block devices")
2108

    
2109
    return disks_info
2110

    
2111

    
2112
def _AssembleInstanceDisks(instance, cfg, ignore_secondaries=False):
2113
  """Prepare the block devices for an instance.
2114

2115
  This sets up the block devices on all nodes.
2116

2117
  Args:
2118
    instance: a ganeti.objects.Instance object
2119
    ignore_secondaries: if true, errors on secondary nodes won't result
2120
                        in an error return from the function
2121

2122
  Returns:
2123
    false if the operation failed
2124
    list of (host, instance_visible_name, node_visible_name) if the operation
2125
         suceeded with the mapping from node devices to instance devices
2126
  """
2127
  device_info = []
2128
  disks_ok = True
2129
  iname = instance.name
2130
  # With the two passes mechanism we try to reduce the window of
2131
  # opportunity for the race condition of switching DRBD to primary
2132
  # before handshaking occured, but we do not eliminate it
2133

    
2134
  # The proper fix would be to wait (with some limits) until the
2135
  # connection has been made and drbd transitions from WFConnection
2136
  # into any other network-connected state (Connected, SyncTarget,
2137
  # SyncSource, etc.)
2138

    
2139
  # 1st pass, assemble on all nodes in secondary mode
2140
  for inst_disk in instance.disks:
2141
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
2142
      cfg.SetDiskID(node_disk, node)
2143
      result = rpc.call_blockdev_assemble(node, node_disk, iname, False)
2144
      if not result:
2145
        logger.Error("could not prepare block device %s on node %s"
2146
                     " (is_primary=False, pass=1)" % (inst_disk.iv_name, node))
2147
        if not ignore_secondaries:
2148
          disks_ok = False
2149

    
2150
  # FIXME: race condition on drbd migration to primary
2151

    
2152
  # 2nd pass, do only the primary node
2153
  for inst_disk in instance.disks:
2154
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
2155
      if node != instance.primary_node:
2156
        continue
2157
      cfg.SetDiskID(node_disk, node)
2158
      result = rpc.call_blockdev_assemble(node, node_disk, iname, True)
2159
      if not result:
2160
        logger.Error("could not prepare block device %s on node %s"
2161
                     " (is_primary=True, pass=2)" % (inst_disk.iv_name, node))
2162
        disks_ok = False
2163
    device_info.append((instance.primary_node, inst_disk.iv_name, result))
2164

    
2165
  # leave the disks configured for the primary node
2166
  # this is a workaround that would be fixed better by
2167
  # improving the logical/physical id handling
2168
  for disk in instance.disks:
2169
    cfg.SetDiskID(disk, instance.primary_node)
2170

    
2171
  return disks_ok, device_info
2172

    
2173

    
2174
def _StartInstanceDisks(cfg, instance, force):
2175
  """Start the disks of an instance.
2176

2177
  """
2178
  disks_ok, dummy = _AssembleInstanceDisks(instance, cfg,
2179
                                           ignore_secondaries=force)
2180
  if not disks_ok:
2181
    _ShutdownInstanceDisks(instance, cfg)
2182
    if force is not None and not force:
2183
      logger.Error("If the message above refers to a secondary node,"
2184
                   " you can retry the operation using '--force'.")
2185
    raise errors.OpExecError("Disk consistency error")
2186

    
2187

    
2188
class LUDeactivateInstanceDisks(NoHooksLU):
2189
  """Shutdown an instance's disks.
2190

2191
  """
2192
  _OP_REQP = ["instance_name"]
2193

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

2197
    This checks that the instance is in the cluster.
2198

2199
    """
2200
    instance = self.cfg.GetInstanceInfo(
2201
      self.cfg.ExpandInstanceName(self.op.instance_name))
2202
    if instance is None:
2203
      raise errors.OpPrereqError("Instance '%s' not known" %
2204
                                 self.op.instance_name)
2205
    self.instance = instance
2206

    
2207
  def Exec(self, feedback_fn):
2208
    """Deactivate the disks
2209

2210
    """
2211
    instance = self.instance
2212
    ins_l = rpc.call_instance_list([instance.primary_node])
2213
    ins_l = ins_l[instance.primary_node]
2214
    if not type(ins_l) is list:
2215
      raise errors.OpExecError("Can't contact node '%s'" %
2216
                               instance.primary_node)
2217

    
2218
    if self.instance.name in ins_l:
2219
      raise errors.OpExecError("Instance is running, can't shutdown"
2220
                               " block devices.")
2221

    
2222
    _ShutdownInstanceDisks(instance, self.cfg)
2223

    
2224

    
2225
def _ShutdownInstanceDisks(instance, cfg, ignore_primary=False):
2226
  """Shutdown block devices of an instance.
2227

2228
  This does the shutdown on all nodes of the instance.
2229

2230
  If the ignore_primary is false, errors on the primary node are
2231
  ignored.
2232

2233
  """
2234
  result = True
2235
  for disk in instance.disks:
2236
    for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
2237
      cfg.SetDiskID(top_disk, node)
2238
      if not rpc.call_blockdev_shutdown(node, top_disk):
2239
        logger.Error("could not shutdown block device %s on node %s" %
2240
                     (disk.iv_name, node))
2241
        if not ignore_primary or node != instance.primary_node:
2242
          result = False
2243
  return result
2244

    
2245

    
2246
def _CheckNodeFreeMemory(cfg, node, reason, requested):
2247
  """Checks if a node has enough free memory.
2248

2249
  This function check if a given node has the needed amount of free
2250
  memory. In case the node has less memory or we cannot get the
2251
  information from the node, this function raise an OpPrereqError
2252
  exception.
2253

2254
  Args:
2255
    - cfg: a ConfigWriter instance
2256
    - node: the node name
2257
    - reason: string to use in the error message
2258
    - requested: the amount of memory in MiB
2259

2260
  """
2261
  nodeinfo = rpc.call_node_info([node], cfg.GetVGName())
2262
  if not nodeinfo or not isinstance(nodeinfo, dict):
2263
    raise errors.OpPrereqError("Could not contact node %s for resource"
2264
                             " information" % (node,))
2265

    
2266
  free_mem = nodeinfo[node].get('memory_free')
2267
  if not isinstance(free_mem, int):
2268
    raise errors.OpPrereqError("Can't compute free memory on node %s, result"
2269
                             " was '%s'" % (node, free_mem))
2270
  if requested > free_mem:
2271
    raise errors.OpPrereqError("Not enough memory on node %s for %s:"
2272
                             " needed %s MiB, available %s MiB" %
2273
                             (node, reason, requested, free_mem))
2274

    
2275

    
2276
class LUStartupInstance(LogicalUnit):
2277
  """Starts an instance.
2278

2279
  """
2280
  HPATH = "instance-start"
2281
  HTYPE = constants.HTYPE_INSTANCE
2282
  _OP_REQP = ["instance_name", "force"]
2283

    
2284
  def BuildHooksEnv(self):
2285
    """Build hooks env.
2286

2287
    This runs on master, primary and secondary nodes of the instance.
2288

2289
    """
2290
    env = {
2291
      "FORCE": self.op.force,
2292
      }
2293
    env.update(_BuildInstanceHookEnvByObject(self.instance))
2294
    nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
2295
          list(self.instance.secondary_nodes))
2296
    return env, nl, nl
2297

    
2298
  def CheckPrereq(self):
2299
    """Check prerequisites.
2300

2301
    This checks that the instance is in the cluster.
2302

2303
    """
2304
    instance = self.cfg.GetInstanceInfo(
2305
      self.cfg.ExpandInstanceName(self.op.instance_name))
2306
    if instance is None:
2307
      raise errors.OpPrereqError("Instance '%s' not known" %
2308
                                 self.op.instance_name)
2309

    
2310
    # check bridges existance
2311
    _CheckInstanceBridgesExist(instance)
2312

    
2313
    _CheckNodeFreeMemory(self.cfg, instance.primary_node,
2314
                         "starting instance %s" % instance.name,
2315
                         instance.memory)
2316

    
2317
    self.instance = instance
2318
    self.op.instance_name = instance.name
2319

    
2320
  def Exec(self, feedback_fn):
2321
    """Start the instance.
2322

2323
    """
2324
    instance = self.instance
2325
    force = self.op.force
2326
    extra_args = getattr(self.op, "extra_args", "")
2327

    
2328
    self.cfg.MarkInstanceUp(instance.name)
2329

    
2330
    node_current = instance.primary_node
2331

    
2332
    _StartInstanceDisks(self.cfg, instance, force)
2333

    
2334
    if not rpc.call_instance_start(node_current, instance, extra_args):
2335
      _ShutdownInstanceDisks(instance, self.cfg)
2336
      raise errors.OpExecError("Could not start instance")
2337

    
2338

    
2339
class LURebootInstance(LogicalUnit):
2340
  """Reboot an instance.
2341

2342
  """
2343
  HPATH = "instance-reboot"
2344
  HTYPE = constants.HTYPE_INSTANCE
2345
  _OP_REQP = ["instance_name", "ignore_secondaries", "reboot_type"]
2346

    
2347
  def BuildHooksEnv(self):
2348
    """Build hooks env.
2349

2350
    This runs on master, primary and secondary nodes of the instance.
2351

2352
    """
2353
    env = {
2354
      "IGNORE_SECONDARIES": self.op.ignore_secondaries,
2355
      }
2356
    env.update(_BuildInstanceHookEnvByObject(self.instance))
2357
    nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
2358
          list(self.instance.secondary_nodes))
2359
    return env, nl, nl
2360

    
2361
  def CheckPrereq(self):
2362
    """Check prerequisites.
2363

2364
    This checks that the instance is in the cluster.
2365

2366
    """
2367
    instance = self.cfg.GetInstanceInfo(
2368
      self.cfg.ExpandInstanceName(self.op.instance_name))
2369
    if instance is None:
2370
      raise errors.OpPrereqError("Instance '%s' not known" %
2371
                                 self.op.instance_name)
2372

    
2373
    # check bridges existance
2374
    _CheckInstanceBridgesExist(instance)
2375

    
2376
    self.instance = instance
2377
    self.op.instance_name = instance.name
2378

    
2379
  def Exec(self, feedback_fn):
2380
    """Reboot the instance.
2381

2382
    """
2383
    instance = self.instance
2384
    ignore_secondaries = self.op.ignore_secondaries
2385
    reboot_type = self.op.reboot_type
2386
    extra_args = getattr(self.op, "extra_args", "")
2387

    
2388
    node_current = instance.primary_node
2389

    
2390
    if reboot_type not in [constants.INSTANCE_REBOOT_SOFT,
2391
                           constants.INSTANCE_REBOOT_HARD,
2392
                           constants.INSTANCE_REBOOT_FULL]:
2393
      raise errors.ParameterError("reboot type not in [%s, %s, %s]" %
2394
                                  (constants.INSTANCE_REBOOT_SOFT,
2395
                                   constants.INSTANCE_REBOOT_HARD,
2396
                                   constants.INSTANCE_REBOOT_FULL))
2397

    
2398
    if reboot_type in [constants.INSTANCE_REBOOT_SOFT,
2399
                       constants.INSTANCE_REBOOT_HARD]:
2400
      if not rpc.call_instance_reboot(node_current, instance,
2401
                                      reboot_type, extra_args):
2402
        raise errors.OpExecError("Could not reboot instance")
2403
    else:
2404
      if not rpc.call_instance_shutdown(node_current, instance):
2405
        raise errors.OpExecError("could not shutdown instance for full reboot")
2406
      _ShutdownInstanceDisks(instance, self.cfg)
2407
      _StartInstanceDisks(self.cfg, instance, ignore_secondaries)
2408
      if not rpc.call_instance_start(node_current, instance, extra_args):
2409
        _ShutdownInstanceDisks(instance, self.cfg)
2410
        raise errors.OpExecError("Could not start instance for full reboot")
2411

    
2412
    self.cfg.MarkInstanceUp(instance.name)
2413

    
2414

    
2415
class LUShutdownInstance(LogicalUnit):
2416
  """Shutdown an instance.
2417

2418
  """
2419
  HPATH = "instance-stop"
2420
  HTYPE = constants.HTYPE_INSTANCE
2421
  _OP_REQP = ["instance_name"]
2422

    
2423
  def BuildHooksEnv(self):
2424
    """Build hooks env.
2425

2426
    This runs on master, primary and secondary nodes of the instance.
2427

2428
    """
2429
    env = _BuildInstanceHookEnvByObject(self.instance)
2430
    nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
2431
          list(self.instance.secondary_nodes))
2432
    return env, nl, nl
2433

    
2434
  def CheckPrereq(self):
2435
    """Check prerequisites.
2436

2437
    This checks that the instance is in the cluster.
2438

2439
    """
2440
    instance = self.cfg.GetInstanceInfo(
2441
      self.cfg.ExpandInstanceName(self.op.instance_name))
2442
    if instance is None:
2443
      raise errors.OpPrereqError("Instance '%s' not known" %
2444
                                 self.op.instance_name)
2445
    self.instance = instance
2446

    
2447
  def Exec(self, feedback_fn):
2448
    """Shutdown the instance.
2449

2450
    """
2451
    instance = self.instance
2452
    node_current = instance.primary_node
2453
    self.cfg.MarkInstanceDown(instance.name)
2454
    if not rpc.call_instance_shutdown(node_current, instance):
2455
      logger.Error("could not shutdown instance")
2456

    
2457
    _ShutdownInstanceDisks(instance, self.cfg)
2458

    
2459

    
2460
class LUReinstallInstance(LogicalUnit):
2461
  """Reinstall an instance.
2462

2463
  """
2464
  HPATH = "instance-reinstall"
2465
  HTYPE = constants.HTYPE_INSTANCE
2466
  _OP_REQP = ["instance_name"]
2467

    
2468
  def BuildHooksEnv(self):
2469
    """Build hooks env.
2470

2471
    This runs on master, primary and secondary nodes of the instance.
2472

2473
    """
2474
    env = _BuildInstanceHookEnvByObject(self.instance)
2475
    nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
2476
          list(self.instance.secondary_nodes))
2477
    return env, nl, nl
2478

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

2482
    This checks that the instance is in the cluster and is not running.
2483

2484
    """
2485
    instance = self.cfg.GetInstanceInfo(
2486
      self.cfg.ExpandInstanceName(self.op.instance_name))
2487
    if instance is None:
2488
      raise errors.OpPrereqError("Instance '%s' not known" %
2489
                                 self.op.instance_name)
2490
    if instance.disk_template == constants.DT_DISKLESS:
2491
      raise errors.OpPrereqError("Instance '%s' has no disks" %
2492
                                 self.op.instance_name)
2493
    if instance.status != "down":
2494
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2495
                                 self.op.instance_name)
2496
    remote_info = rpc.call_instance_info(instance.primary_node, instance.name)
2497
    if remote_info:
2498
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
2499
                                 (self.op.instance_name,
2500
                                  instance.primary_node))
2501

    
2502
    self.op.os_type = getattr(self.op, "os_type", None)
2503
    if self.op.os_type is not None:
2504
      # OS verification
2505
      pnode = self.cfg.GetNodeInfo(
2506
        self.cfg.ExpandNodeName(instance.primary_node))
2507
      if pnode is None:
2508
        raise errors.OpPrereqError("Primary node '%s' is unknown" %
2509
                                   self.op.pnode)
2510
      os_obj = rpc.call_os_get(pnode.name, self.op.os_type)
2511
      if not os_obj:
2512
        raise errors.OpPrereqError("OS '%s' not in supported OS list for"
2513
                                   " primary node"  % self.op.os_type)
2514

    
2515
    self.instance = instance
2516

    
2517
  def Exec(self, feedback_fn):
2518
    """Reinstall the instance.
2519

2520
    """
2521
    inst = self.instance
2522

    
2523
    if self.op.os_type is not None:
2524
      feedback_fn("Changing OS to '%s'..." % self.op.os_type)
2525
      inst.os = self.op.os_type
2526
      self.cfg.AddInstance(inst)
2527

    
2528
    _StartInstanceDisks(self.cfg, inst, None)
2529
    try:
2530
      feedback_fn("Running the instance OS create scripts...")
2531
      if not rpc.call_instance_os_add(inst.primary_node, inst, "sda", "sdb"):
2532
        raise errors.OpExecError("Could not install OS for instance %s"
2533
                                 " on node %s" %
2534
                                 (inst.name, inst.primary_node))
2535
    finally:
2536
      _ShutdownInstanceDisks(inst, self.cfg)
2537

    
2538

    
2539
class LURenameInstance(LogicalUnit):
2540
  """Rename an instance.
2541

2542
  """
2543
  HPATH = "instance-rename"
2544
  HTYPE = constants.HTYPE_INSTANCE
2545
  _OP_REQP = ["instance_name", "new_name"]
2546

    
2547
  def BuildHooksEnv(self):
2548
    """Build hooks env.
2549

2550
    This runs on master, primary and secondary nodes of the instance.
2551

2552
    """
2553
    env = _BuildInstanceHookEnvByObject(self.instance)
2554
    env["INSTANCE_NEW_NAME"] = self.op.new_name
2555
    nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
2556
          list(self.instance.secondary_nodes))
2557
    return env, nl, nl
2558

    
2559
  def CheckPrereq(self):
2560
    """Check prerequisites.
2561

2562
    This checks that the instance is in the cluster and is not running.
2563

2564
    """
2565
    instance = self.cfg.GetInstanceInfo(
2566
      self.cfg.ExpandInstanceName(self.op.instance_name))
2567
    if instance is None:
2568
      raise errors.OpPrereqError("Instance '%s' not known" %
2569
                                 self.op.instance_name)
2570
    if instance.status != "down":
2571
      raise errors.OpPrereqError("Instance '%s' is marked to be up" %
2572
                                 self.op.instance_name)
2573
    remote_info = rpc.call_instance_info(instance.primary_node, instance.name)
2574
    if remote_info:
2575
      raise errors.OpPrereqError("Instance '%s' is running on the node %s" %
2576
                                 (self.op.instance_name,
2577
                                  instance.primary_node))
2578
    self.instance = instance
2579

    
2580
    # new name verification
2581
    name_info = utils.HostInfo(self.op.new_name)
2582

    
2583
    self.op.new_name = new_name = name_info.name
2584
    instance_list = self.cfg.GetInstanceList()
2585
    if new_name in instance_list:
2586
      raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
2587
                                 new_name)
2588

    
2589
    if not getattr(self.op, "ignore_ip", False):
2590
      command = ["fping", "-q", name_info.ip]
2591
      result = utils.RunCmd(command)
2592
      if not result.failed:
2593
        raise errors.OpPrereqError("IP %s of instance %s already in use" %
2594
                                   (name_info.ip, new_name))
2595

    
2596

    
2597
  def Exec(self, feedback_fn):
2598
    """Reinstall the instance.
2599

2600
    """
2601
    inst = self.instance
2602
    old_name = inst.name
2603

    
2604
    if inst.disk_template == constants.DT_FILE:
2605
      old_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
2606

    
2607
    self.cfg.RenameInstance(inst.name, self.op.new_name)
2608

    
2609
    # re-read the instance from the configuration after rename
2610
    inst = self.cfg.GetInstanceInfo(self.op.new_name)
2611

    
2612
    if inst.disk_template == constants.DT_FILE:
2613
      new_file_storage_dir = os.path.dirname(inst.disks[0].logical_id[1])
2614
      result = rpc.call_file_storage_dir_rename(inst.primary_node,
2615
                                                old_file_storage_dir,
2616
                                                new_file_storage_dir)
2617

    
2618
      if not result:
2619
        raise errors.OpExecError("Could not connect to node '%s' to rename"
2620
                                 " directory '%s' to '%s' (but the instance"
2621
                                 " has been renamed in Ganeti)" % (
2622
                                 inst.primary_node, old_file_storage_dir,
2623
                                 new_file_storage_dir))
2624

    
2625
      if not result[0]:
2626
        raise errors.OpExecError("Could not rename directory '%s' to '%s'"
2627
                                 " (but the instance has been renamed in"
2628
                                 " Ganeti)" % (old_file_storage_dir,
2629
                                               new_file_storage_dir))
2630

    
2631
    _StartInstanceDisks(self.cfg, inst, None)
2632
    try:
2633
      if not rpc.call_instance_run_rename(inst.primary_node, inst, old_name,
2634
                                          "sda", "sdb"):
2635
        msg = ("Could run OS rename script for instance %s on node %s (but the"
2636
               " instance has been renamed in Ganeti)" %
2637
               (inst.name, inst.primary_node))
2638
        logger.Error(msg)
2639
    finally:
2640
      _ShutdownInstanceDisks(inst, self.cfg)
2641

    
2642

    
2643
class LURemoveInstance(LogicalUnit):
2644
  """Remove an instance.
2645

2646
  """
2647
  HPATH = "instance-remove"
2648
  HTYPE = constants.HTYPE_INSTANCE
2649
  _OP_REQP = ["instance_name"]
2650

    
2651
  def BuildHooksEnv(self):
2652
    """Build hooks env.
2653

2654
    This runs on master, primary and secondary nodes of the instance.
2655

2656
    """
2657
    env = _BuildInstanceHookEnvByObject(self.instance)
2658
    nl = [self.sstore.GetMasterNode()]
2659
    return env, nl, nl
2660

    
2661
  def CheckPrereq(self):
2662
    """Check prerequisites.
2663

2664
    This checks that the instance is in the cluster.
2665

2666
    """
2667
    instance = self.cfg.GetInstanceInfo(
2668
      self.cfg.ExpandInstanceName(self.op.instance_name))
2669
    if instance is None:
2670
      raise errors.OpPrereqError("Instance '%s' not known" %
2671
                                 self.op.instance_name)
2672
    self.instance = instance
2673

    
2674
  def Exec(self, feedback_fn):
2675
    """Remove the instance.
2676

2677
    """
2678
    instance = self.instance
2679
    logger.Info("shutting down instance %s on node %s" %
2680
                (instance.name, instance.primary_node))
2681

    
2682
    if not rpc.call_instance_shutdown(instance.primary_node, instance):
2683
      if self.op.ignore_failures:
2684
        feedback_fn("Warning: can't shutdown instance")
2685
      else:
2686
        raise errors.OpExecError("Could not shutdown instance %s on node %s" %
2687
                                 (instance.name, instance.primary_node))
2688

    
2689
    logger.Info("removing block devices for instance %s" % instance.name)
2690

    
2691
    if not _RemoveDisks(instance, self.cfg):
2692
      if self.op.ignore_failures:
2693
        feedback_fn("Warning: can't remove instance's disks")
2694
      else:
2695
        raise errors.OpExecError("Can't remove instance's disks")
2696

    
2697
    logger.Info("removing instance %s out of cluster config" % instance.name)
2698

    
2699
    self.cfg.RemoveInstance(instance.name)
2700

    
2701

    
2702
class LUQueryInstances(NoHooksLU):
2703
  """Logical unit for querying instances.
2704

2705
  """
2706
  _OP_REQP = ["output_fields", "names"]
2707

    
2708
  def CheckPrereq(self):
2709
    """Check prerequisites.
2710

2711
    This checks that the fields required are valid output fields.
2712

2713
    """
2714
    self.dynamic_fields = frozenset(["oper_state", "oper_ram", "status"])
2715
    _CheckOutputFields(static=["name", "os", "pnode", "snodes",
2716
                               "admin_state", "admin_ram",
2717
                               "disk_template", "ip", "mac", "bridge",
2718
                               "sda_size", "sdb_size", "vcpus"],
2719
                       dynamic=self.dynamic_fields,
2720
                       selected=self.op.output_fields)
2721

    
2722
    self.wanted = _GetWantedInstances(self, self.op.names)
2723

    
2724
  def Exec(self, feedback_fn):
2725
    """Computes the list of nodes and their attributes.
2726

2727
    """
2728
    instance_names = self.wanted
2729
    instance_list = [self.cfg.GetInstanceInfo(iname) for iname
2730
                     in instance_names]
2731

    
2732
    # begin data gathering
2733

    
2734
    nodes = frozenset([inst.primary_node for inst in instance_list])
2735

    
2736
    bad_nodes = []
2737
    if self.dynamic_fields.intersection(self.op.output_fields):
2738
      live_data = {}
2739
      node_data = rpc.call_all_instances_info(nodes)
2740
      for name in nodes:
2741
        result = node_data[name]
2742
        if result:
2743
          live_data.update(result)
2744
        elif result == False:
2745
          bad_nodes.append(name)
2746
        # else no instance is alive
2747
    else:
2748
      live_data = dict([(name, {}) for name in instance_names])
2749

    
2750
    # end data gathering
2751

    
2752
    output = []
2753
    for instance in instance_list:
2754
      iout = []
2755
      for field in self.op.output_fields:
2756
        if field == "name":
2757
          val = instance.name
2758
        elif field == "os":
2759
          val = instance.os
2760
        elif field == "pnode":
2761
          val = instance.primary_node
2762
        elif field == "snodes":
2763
          val = list(instance.secondary_nodes)
2764
        elif field == "admin_state":
2765
          val = (instance.status != "down")
2766
        elif field == "oper_state":
2767
          if instance.primary_node in bad_nodes:
2768
            val = None
2769
          else:
2770
            val = bool(live_data.get(instance.name))
2771
        elif field == "status":
2772
          if instance.primary_node in bad_nodes:
2773
            val = "ERROR_nodedown"
2774
          else:
2775
            running = bool(live_data.get(instance.name))
2776
            if running:
2777
              if instance.status != "down":
2778
                val = "running"
2779
              else:
2780
                val = "ERROR_up"
2781
            else:
2782
              if instance.status != "down":
2783
                val = "ERROR_down"
2784
              else:
2785
                val = "ADMIN_down"
2786
        elif field == "admin_ram":
2787
          val = instance.memory
2788
        elif field == "oper_ram":
2789
          if instance.primary_node in bad_nodes:
2790
            val = None
2791
          elif instance.name in live_data:
2792
            val = live_data[instance.name].get("memory", "?")
2793
          else:
2794
            val = "-"
2795
        elif field == "disk_template":
2796
          val = instance.disk_template
2797
        elif field == "ip":
2798
          val = instance.nics[0].ip
2799
        elif field == "bridge":
2800
          val = instance.nics[0].bridge
2801
        elif field == "mac":
2802
          val = instance.nics[0].mac
2803
        elif field == "sda_size" or field == "sdb_size":
2804
          disk = instance.FindDisk(field[:3])
2805
          if disk is None:
2806
            val = None
2807
          else:
2808
            val = disk.size
2809
        elif field == "vcpus":
2810
          val = instance.vcpus
2811
        else:
2812
          raise errors.ParameterError(field)
2813
        iout.append(val)
2814
      output.append(iout)
2815

    
2816
    return output
2817

    
2818

    
2819
class LUFailoverInstance(LogicalUnit):
2820
  """Failover an instance.
2821

2822
  """
2823
  HPATH = "instance-failover"
2824
  HTYPE = constants.HTYPE_INSTANCE
2825
  _OP_REQP = ["instance_name", "ignore_consistency"]
2826

    
2827
  def BuildHooksEnv(self):
2828
    """Build hooks env.
2829

2830
    This runs on master, primary and secondary nodes of the instance.
2831

2832
    """
2833
    env = {
2834
      "IGNORE_CONSISTENCY": self.op.ignore_consistency,
2835
      }
2836
    env.update(_BuildInstanceHookEnvByObject(self.instance))
2837
    nl = [self.sstore.GetMasterNode()] + list(self.instance.secondary_nodes)
2838
    return env, nl, nl
2839

    
2840
  def CheckPrereq(self):
2841
    """Check prerequisites.
2842

2843
    This checks that the instance is in the cluster.
2844

2845
    """
2846
    instance = self.cfg.GetInstanceInfo(
2847
      self.cfg.ExpandInstanceName(self.op.instance_name))
2848
    if instance is None:
2849
      raise errors.OpPrereqError("Instance '%s' not known" %
2850
                                 self.op.instance_name)
2851

    
2852
    if instance.disk_template not in constants.DTS_NET_MIRROR:
2853
      raise errors.OpPrereqError("Instance's disk layout is not"
2854
                                 " network mirrored, cannot failover.")
2855

    
2856
    secondary_nodes = instance.secondary_nodes
2857
    if not secondary_nodes:
2858
      raise errors.ProgrammerError("no secondary node but using "
2859
                                   "DT_REMOTE_RAID1 template")
2860

    
2861
    target_node = secondary_nodes[0]
2862
    # check memory requirements on the secondary node
2863
    _CheckNodeFreeMemory(self.cfg, target_node, "failing over instance %s" %
2864
                         instance.name, instance.memory)
2865

    
2866
    # check bridge existance
2867
    brlist = [nic.bridge for nic in instance.nics]
2868
    if not rpc.call_bridges_exist(target_node, brlist):
2869
      raise errors.OpPrereqError("One or more target bridges %s does not"
2870
                                 " exist on destination node '%s'" %
2871
                                 (brlist, target_node))
2872

    
2873
    self.instance = instance
2874

    
2875
  def Exec(self, feedback_fn):
2876
    """Failover an instance.
2877

2878
    The failover is done by shutting it down on its present node and
2879
    starting it on the secondary.
2880

2881
    """
2882
    instance = self.instance
2883

    
2884
    source_node = instance.primary_node
2885
    target_node = instance.secondary_nodes[0]
2886

    
2887
    feedback_fn("* checking disk consistency between source and target")
2888
    for dev in instance.disks:
2889
      # for remote_raid1, these are md over drbd
2890
      if not _CheckDiskConsistency(self.cfg, dev, target_node, False):
2891
        if instance.status == "up" and not self.op.ignore_consistency:
2892
          raise errors.OpExecError("Disk %s is degraded on target node,"
2893
                                   " aborting failover." % dev.iv_name)
2894

    
2895
    feedback_fn("* shutting down instance on source node")
2896
    logger.Info("Shutting down instance %s on node %s" %
2897
                (instance.name, source_node))
2898

    
2899
    if not rpc.call_instance_shutdown(source_node, instance):
2900
      if self.op.ignore_consistency:
2901
        logger.Error("Could not shutdown instance %s on node %s. Proceeding"
2902
                     " anyway. Please make sure node %s is down"  %
2903
                     (instance.name, source_node, source_node))
2904
      else:
2905
        raise errors.OpExecError("Could not shutdown instance %s on node %s" %
2906
                                 (instance.name, source_node))
2907

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

    
2912
    instance.primary_node = target_node
2913
    # distribute new instance config to the other nodes
2914
    self.cfg.AddInstance(instance)
2915

    
2916
    # Only start the instance if it's marked as up
2917
    if instance.status == "up":
2918
      feedback_fn("* activating the instance's disks on target node")
2919
      logger.Info("Starting instance %s on node %s" %
2920
                  (instance.name, target_node))
2921

    
2922
      disks_ok, dummy = _AssembleInstanceDisks(instance, self.cfg,
2923
                                               ignore_secondaries=True)
2924
      if not disks_ok:
2925
        _ShutdownInstanceDisks(instance, self.cfg)
2926
        raise errors.OpExecError("Can't activate the instance's disks")
2927

    
2928
      feedback_fn("* starting the instance on the target node")
2929
      if not rpc.call_instance_start(target_node, instance, None):
2930
        _ShutdownInstanceDisks(instance, self.cfg)
2931
        raise errors.OpExecError("Could not start instance %s on node %s." %
2932
                                 (instance.name, target_node))
2933

    
2934

    
2935
def _CreateBlockDevOnPrimary(cfg, node, instance, device, info):
2936
  """Create a tree of block devices on the primary node.
2937

2938
  This always creates all devices.
2939

2940
  """
2941
  if device.children:
2942
    for child in device.children:
2943
      if not _CreateBlockDevOnPrimary(cfg, node, instance, child, info):
2944
        return False
2945

    
2946
  cfg.SetDiskID(device, node)
2947
  new_id = rpc.call_blockdev_create(node, device, device.size,
2948
                                    instance.name, True, info)
2949
  if not new_id:
2950
    return False
2951
  if device.physical_id is None:
2952
    device.physical_id = new_id
2953
  return True
2954

    
2955

    
2956
def _CreateBlockDevOnSecondary(cfg, node, instance, device, force, info):
2957
  """Create a tree of block devices on a secondary node.
2958

2959
  If this device type has to be created on secondaries, create it and
2960
  all its children.
2961

2962
  If not, just recurse to children keeping the same 'force' value.
2963

2964
  """
2965
  if device.CreateOnSecondary():
2966
    force = True
2967
  if device.children:
2968
    for child in device.children:
2969
      if not _CreateBlockDevOnSecondary(cfg, node, instance,
2970
                                        child, force, info):
2971
        return False
2972

    
2973
  if not force:
2974
    return True
2975
  cfg.SetDiskID(device, node)
2976
  new_id = rpc.call_blockdev_create(node, device, device.size,
2977
                                    instance.name, False, info)
2978
  if not new_id:
2979
    return False
2980
  if device.physical_id is None:
2981
    device.physical_id = new_id
2982
  return True
2983

    
2984

    
2985
def _GenerateUniqueNames(cfg, exts):
2986
  """Generate a suitable LV name.
2987

2988
  This will generate a logical volume name for the given instance.
2989

2990
  """
2991
  results = []
2992
  for val in exts:
2993
    new_id = cfg.GenerateUniqueID()
2994
    results.append("%s%s" % (new_id, val))
2995
  return results
2996

    
2997

    
2998
def _GenerateMDDRBDBranch(cfg, primary, secondary, size, names):
2999
  """Generate a drbd device complete with its children.
3000

3001
  """
3002
  port = cfg.AllocatePort()
3003
  vgname = cfg.GetVGName()
3004
  dev_data = objects.Disk(dev_type=constants.LD_LV, size=size,
3005
                          logical_id=(vgname, names[0]))
3006
  dev_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
3007
                          logical_id=(vgname, names[1]))
3008
  drbd_dev = objects.Disk(dev_type=constants.LD_DRBD7, size=size,
3009
                          logical_id = (primary, secondary, port),
3010
                          children = [dev_data, dev_meta])
3011
  return drbd_dev
3012

    
3013

    
3014
def _GenerateDRBD8Branch(cfg, primary, secondary, size, names, iv_name):
3015
  """Generate a drbd8 device complete with its children.
3016

3017
  """
3018
  port = cfg.AllocatePort()
3019
  vgname = cfg.GetVGName()
3020
  dev_data = objects.Disk(dev_type=constants.LD_LV, size=size,
3021
                          logical_id=(vgname, names[0]))
3022
  dev_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
3023
                          logical_id=(vgname, names[1]))
3024
  drbd_dev = objects.Disk(dev_type=constants.LD_DRBD8, size=size,
3025
                          logical_id = (primary, secondary, port),
3026
                          children = [dev_data, dev_meta],
3027
                          iv_name=iv_name)
3028
  return drbd_dev
3029

    
3030

    
3031
def _GenerateDiskTemplate(cfg, template_name,
3032
                          instance_name, primary_node,
3033
                          secondary_nodes, disk_sz, swap_sz,
3034
                          file_storage_dir, file_driver):
3035
  """Generate the entire disk layout for a given template type.
3036

3037
  """
3038
  #TODO: compute space requirements
3039

    
3040
  vgname = cfg.GetVGName()
3041
  if template_name == constants.DT_DISKLESS:
3042
    disks = []
3043
  elif template_name == constants.DT_PLAIN:
3044
    if len(secondary_nodes) != 0:
3045
      raise errors.ProgrammerError("Wrong template configuration")
3046

    
3047
    names = _GenerateUniqueNames(cfg, [".sda", ".sdb"])
3048
    sda_dev = objects.Disk(dev_type=constants.LD_LV, size=disk_sz,
3049
                           logical_id=(vgname, names[0]),
3050
                           iv_name = "sda")
3051
    sdb_dev = objects.Disk(dev_type=constants.LD_LV, size=swap_sz,
3052
                           logical_id=(vgname, names[1]),
3053
                           iv_name = "sdb")
3054
    disks = [sda_dev, sdb_dev]
3055
  elif template_name == constants.DT_DRBD8:
3056
    if len(secondary_nodes) != 1:
3057
      raise errors.ProgrammerError("Wrong template configuration")
3058
    remote_node = secondary_nodes[0]
3059
    names = _GenerateUniqueNames(cfg, [".sda_data", ".sda_meta",
3060
                                       ".sdb_data", ".sdb_meta"])
3061
    drbd_sda_dev = _GenerateDRBD8Branch(cfg, primary_node, remote_node,
3062
                                         disk_sz, names[0:2], "sda")
3063
    drbd_sdb_dev = _GenerateDRBD8Branch(cfg, primary_node, remote_node,
3064
                                         swap_sz, names[2:4], "sdb")
3065
    disks = [drbd_sda_dev, drbd_sdb_dev]
3066
  elif template_name == constants.DT_FILE:
3067
    if len(secondary_nodes) != 0:
3068
      raise errors.ProgrammerError("Wrong template configuration")
3069

    
3070
    file_sda_dev = objects.Disk(dev_type=constants.LD_FILE, size=disk_sz,
3071
                                iv_name="sda", logical_id=(file_driver,
3072
                                "%s/sda" % file_storage_dir))
3073
    file_sdb_dev = objects.Disk(dev_type=constants.LD_FILE, size=swap_sz,
3074
                                iv_name="sdb", logical_id=(file_driver,
3075
                                "%s/sdb" % file_storage_dir))
3076
    disks = [file_sda_dev, file_sdb_dev]
3077
  else:
3078
    raise errors.ProgrammerError("Invalid disk template '%s'" % template_name)
3079
  return disks
3080

    
3081

    
3082
def _GetInstanceInfoText(instance):
3083
  """Compute that text that should be added to the disk's metadata.
3084

3085
  """
3086
  return "originstname+%s" % instance.name
3087

    
3088

    
3089
def _CreateDisks(cfg, instance):
3090
  """Create all disks for an instance.
3091

3092
  This abstracts away some work from AddInstance.
3093

3094
  Args:
3095
    instance: the instance object
3096

3097
  Returns:
3098
    True or False showing the success of the creation process
3099

3100
  """
3101
  info = _GetInstanceInfoText(instance)
3102

    
3103
  if instance.disk_template == constants.DT_FILE:
3104
    file_storage_dir = os.path.dirname(instance.disks[0].logical_id[1])
3105
    result = rpc.call_file_storage_dir_create(instance.primary_node,
3106
                                              file_storage_dir)
3107

    
3108
    if not result:
3109
      logger.Error("Could not connect to node '%s'" % instance.primary_node)
3110
      return False
3111

    
3112
    if not result[0]:
3113
      logger.Error("failed to create directory '%s'" % file_storage_dir)
3114
      return False
3115

    
3116
  for device in instance.disks:
3117
    logger.Info("creating volume %s for instance %s" %
3118
                (device.iv_name, instance.name))
3119
    #HARDCODE
3120
    for secondary_node in instance.secondary_nodes:
3121
      if not _CreateBlockDevOnSecondary(cfg, secondary_node, instance,
3122
                                        device, False, info):
3123
        logger.Error("failed to create volume %s (%s) on secondary node %s!" %
3124
                     (device.iv_name, device, secondary_node))
3125
        return False
3126
    #HARDCODE
3127
    if not _CreateBlockDevOnPrimary(cfg, instance.primary_node,
3128
                                    instance, device, info):
3129
      logger.Error("failed to create volume %s on primary!" %
3130
                   device.iv_name)
3131
      return False
3132

    
3133
  return True
3134

    
3135

    
3136
def _RemoveDisks(instance, cfg):
3137
  """Remove all disks for an instance.
3138

3139
  This abstracts away some work from `AddInstance()` and
3140
  `RemoveInstance()`. Note that in case some of the devices couldn't
3141
  be removed, the removal will continue with the other ones (compare
3142
  with `_CreateDisks()`).
3143

3144
  Args:
3145
    instance: the instance object
3146

3147
  Returns:
3148
    True or False showing the success of the removal proces
3149

3150
  """
3151
  logger.Info("removing block devices for instance %s" % instance.name)
3152

    
3153
  result = True
3154
  for device in instance.disks:
3155
    for node, disk in device.ComputeNodeTree(instance.primary_node):
3156
      cfg.SetDiskID(disk, node)
3157
      if not rpc.call_blockdev_remove(node, disk):
3158
        logger.Error("could not remove block device %s on node %s,"
3159
                     " continuing anyway" %
3160
                     (device.iv_name, node))
3161
        result = False
3162

    
3163
  if instance.disk_template == constants.DT_FILE:
3164
    file_storage_dir = os.path.dirname(instance.disks[0].logical_id[1])
3165
    if not rpc.call_file_storage_dir_remove(instance.primary_node,
3166
                                            file_storage_dir):
3167
      logger.Error("could not remove directory '%s'" % file_storage_dir)
3168
      result = False
3169

    
3170
  return result
3171

    
3172

    
3173
def _ComputeDiskSize(disk_template, disk_size, swap_size):
3174
  """Compute disk size requirements in the volume group
3175

3176
  This is currently hard-coded for the two-drive layout.
3177

3178
  """
3179
  # Required free disk space as a function of disk and swap space
3180
  req_size_dict = {
3181
    constants.DT_DISKLESS: None,
3182
    constants.DT_PLAIN: disk_size + swap_size,
3183
    # 256 MB are added for drbd metadata, 128MB for each drbd device
3184
    constants.DT_DRBD8: disk_size + swap_size + 256,
3185
    constants.DT_FILE: None,
3186
  }
3187

    
3188
  if disk_template not in req_size_dict:
3189
    raise errors.ProgrammerError("Disk template '%s' size requirement"
3190
                                 " is unknown" %  disk_template)
3191

    
3192
  return req_size_dict[disk_template]
3193

    
3194

    
3195
class LUCreateInstance(LogicalUnit):
3196
  """Create an instance.
3197

3198
  """
3199
  HPATH = "instance-add"
3200
  HTYPE = constants.HTYPE_INSTANCE
3201
  _OP_REQP = ["instance_name", "mem_size", "disk_size",
3202
              "disk_template", "swap_size", "mode", "start", "vcpus",
3203
              "wait_for_sync", "ip_check", "mac"]
3204

    
3205
  def _RunAllocator(self):
3206
    """Run the allocator based on input opcode.
3207

3208
    """
3209
    disks = [{"size": self.op.disk_size, "mode": "w"},
3210
             {"size": self.op.swap_size, "mode": "w"}]
3211
    nics = [{"mac": self.op.mac, "ip": getattr(self.op, "ip", None),
3212
             "bridge": self.op.bridge}]
3213
    ial = IAllocator(self.cfg, self.sstore,
3214
                     mode=constants.IALLOCATOR_MODE_ALLOC,
3215
                     name=self.op.instance_name,
3216
                     disk_template=self.op.disk_template,
3217
                     tags=[],
3218
                     os=self.op.os_type,
3219
                     vcpus=self.op.vcpus,
3220
                     mem_size=self.op.mem_size,
3221
                     disks=disks,
3222
                     nics=nics,
3223
                     )
3224

    
3225
    ial.Run(self.op.iallocator)
3226

    
3227
    if not ial.success:
3228
      raise errors.OpPrereqError("Can't compute nodes using"
3229
                                 " iallocator '%s': %s" % (self.op.iallocator,
3230
                                                           ial.info))
3231
    if len(ial.nodes) != ial.required_nodes:
3232
      raise errors.OpPrereqError("iallocator '%s' returned invalid number"
3233
                                 " of nodes (%s), required %s" %
3234
                                 (len(ial.nodes), ial.required_nodes))
3235
    self.op.pnode = ial.nodes[0]
3236
    logger.ToStdout("Selected nodes for the instance: %s" %
3237
                    (", ".join(ial.nodes),))
3238
    logger.Info("Selected nodes for instance %s via iallocator %s: %s" %
3239
                (self.op.instance_name, self.op.iallocator, ial.nodes))
3240
    if ial.required_nodes == 2:
3241
      self.op.snode = ial.nodes[1]
3242

    
3243
  def BuildHooksEnv(self):
3244
    """Build hooks env.
3245

3246
    This runs on master, primary and secondary nodes of the instance.
3247

3248
    """
3249
    env = {
3250
      "INSTANCE_DISK_TEMPLATE": self.op.disk_template,
3251
      "INSTANCE_DISK_SIZE": self.op.disk_size,
3252
      "INSTANCE_SWAP_SIZE": self.op.swap_size,
3253
      "INSTANCE_ADD_MODE": self.op.mode,
3254
      }
3255
    if self.op.mode == constants.INSTANCE_IMPORT:
3256
      env["INSTANCE_SRC_NODE"] = self.op.src_node
3257
      env["INSTANCE_SRC_PATH"] = self.op.src_path
3258
      env["INSTANCE_SRC_IMAGE"] = self.src_image
3259

    
3260
    env.update(_BuildInstanceHookEnv(name=self.op.instance_name,
3261
      primary_node=self.op.pnode,
3262
      secondary_nodes=self.secondaries,
3263
      status=self.instance_status,
3264
      os_type=self.op.os_type,
3265
      memory=self.op.mem_size,
3266
      vcpus=self.op.vcpus,
3267
      nics=[(self.inst_ip, self.op.bridge, self.op.mac)],
3268
    ))
3269

    
3270
    nl = ([self.sstore.GetMasterNode(), self.op.pnode] +
3271
          self.secondaries)
3272
    return env, nl, nl
3273

    
3274

    
3275
  def CheckPrereq(self):
3276
    """Check prerequisites.
3277

3278
    """
3279
    # set optional parameters to none if they don't exist
3280
    for attr in ["kernel_path", "initrd_path", "hvm_boot_order", "pnode",
3281
                 "iallocator"]:
3282
      if not hasattr(self.op, attr):
3283
        setattr(self.op, attr, None)
3284

    
3285
    if self.op.mode not in (constants.INSTANCE_CREATE,
3286
                            constants.INSTANCE_IMPORT):
3287
      raise errors.OpPrereqError("Invalid instance creation mode '%s'" %
3288
                                 self.op.mode)
3289

    
3290
    if (not self.cfg.GetVGName() and
3291
        self.op.disk_template not in constants.DTS_NOT_LVM):
3292
      raise errors.OpPrereqError("Cluster does not support lvm-based"
3293
                                 " instances")
3294

    
3295
    if self.op.mode == constants.INSTANCE_IMPORT:
3296
      src_node = getattr(self.op, "src_node", None)
3297
      src_path = getattr(self.op, "src_path", None)
3298
      if src_node is None or src_path is None:
3299
        raise errors.OpPrereqError("Importing an instance requires source"
3300
                                   " node and path options")
3301
      src_node_full = self.cfg.ExpandNodeName(src_node)
3302
      if src_node_full is None:
3303
        raise errors.OpPrereqError("Unknown source node '%s'" % src_node)
3304
      self.op.src_node = src_node = src_node_full
3305

    
3306
      if not os.path.isabs(src_path):
3307
        raise errors.OpPrereqError("The source path must be absolute")
3308

    
3309
      export_info = rpc.call_export_info(src_node, src_path)
3310

    
3311
      if not export_info:
3312
        raise errors.OpPrereqError("No export found in dir %s" % src_path)
3313

    
3314
      if not export_info.has_section(constants.INISECT_EXP):
3315
        raise errors.ProgrammerError("Corrupted export config")
3316

    
3317
      ei_version = export_info.get(constants.INISECT_EXP, 'version')
3318
      if (int(ei_version) != constants.EXPORT_VERSION):
3319
        raise errors.OpPrereqError("Wrong export version %s (wanted %d)" %
3320
                                   (ei_version, constants.EXPORT_VERSION))
3321

    
3322
      if int(export_info.get(constants.INISECT_INS, 'disk_count')) > 1:
3323
        raise errors.OpPrereqError("Can't import instance with more than"
3324
                                   " one data disk")
3325

    
3326
      # FIXME: are the old os-es, disk sizes, etc. useful?
3327
      self.op.os_type = export_info.get(constants.INISECT_EXP, 'os')
3328
      diskimage = os.path.join(src_path, export_info.get(constants.INISECT_INS,
3329
                                                         'disk0_dump'))
3330
      self.src_image = diskimage
3331
    else: # INSTANCE_CREATE
3332
      if getattr(self.op, "os_type", None) is None:
3333
        raise errors.OpPrereqError("No guest OS specified")
3334

    
3335
    #### instance parameters check
3336

    
3337
    # disk template and mirror node verification
3338
    if self.op.disk_template not in constants.DISK_TEMPLATES:
3339
      raise errors.OpPrereqError("Invalid disk template name")
3340

    
3341
    # instance name verification
3342
    hostname1 = utils.HostInfo(self.op.instance_name)
3343

    
3344
    self.op.instance_name = instance_name = hostname1.name
3345
    instance_list = self.cfg.GetInstanceList()
3346
    if instance_name in instance_list:
3347
      raise errors.OpPrereqError("Instance '%s' is already in the cluster" %
3348
                                 instance_name)
3349

    
3350
    # ip validity checks
3351
    ip = getattr(self.op, "ip", None)
3352
    if ip is None or ip.lower() == "none":
3353
      inst_ip = None
3354
    elif ip.lower() == "auto":
3355
      inst_ip = hostname1.ip
3356
    else:
3357
      if not utils.IsValidIP(ip):
3358
        raise errors.OpPrereqError("given IP address '%s' doesn't look"
3359
                                   " like a valid IP" % ip)
3360
      inst_ip = ip
3361
    self.inst_ip = self.op.ip = inst_ip
3362

    
3363
    if self.op.start and not self.op.ip_check:
3364
      raise errors.OpPrereqError("Cannot ignore IP address conflicts when"
3365
                                 " adding an instance in start mode")
3366

    
3367
    if self.op.ip_check:
3368
      if utils.TcpPing(hostname1.ip, constants.DEFAULT_NODED_PORT):
3369
        raise errors.OpPrereqError("IP %s of instance %s already in use" %
3370
                                   (hostname1.ip, instance_name))
3371

    
3372
    # MAC address verification
3373
    if self.op.mac != "auto":
3374
      if not utils.IsValidMac(self.op.mac.lower()):
3375
        raise errors.OpPrereqError("invalid MAC address specified: %s" %
3376
                                   self.op.mac)
3377

    
3378
    # bridge verification
3379
    bridge = getattr(self.op, "bridge", None)
3380
    if bridge is None:
3381
      self.op.bridge = self.cfg.GetDefBridge()
3382
    else:
3383
      self.op.bridge = bridge
3384

    
3385
    # boot order verification
3386
    if self.op.hvm_boot_order is not None:
3387
      if len(self.op.hvm_boot_order.strip("acdn")) != 0:
3388
        raise errors.OpPrereqError("invalid boot order specified,"
3389
                                   " must be one or more of [acdn]")
3390
    # file storage checks
3391
    if (self.op.file_driver and
3392
        not self.op.file_driver in constants.FILE_DRIVER):
3393
      raise errors.OpPrereqError("Invalid file driver name '%s'" %
3394
                                 self.op.file_driver)
3395

    
3396
    if self.op.file_storage_dir and os.path.isabs(self.op.file_storage_dir):
3397
      raise errors.OpPrereqError("File storage directory not a relative"
3398
                                 " path")
3399
    #### allocator run
3400

    
3401
    if [self.op.iallocator, self.op.pnode].count(None) != 1:
3402
      raise errors.OpPrereqError("One and only one of iallocator and primary"
3403
                                 " node must be given")
3404

    
3405
    if self.op.iallocator is not None:
3406
      self._RunAllocator()
3407

    
3408
    #### node related checks
3409

    
3410
    # check primary node
3411
    pnode = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.pnode))
3412
    if pnode is None:
3413
      raise errors.OpPrereqError("Primary node '%s' is unknown" %
3414
                                 self.op.pnode)
3415
    self.op.pnode = pnode.name
3416
    self.pnode = pnode
3417
    self.secondaries = []
3418

    
3419
    # mirror node verification
3420
    if self.op.disk_template in constants.DTS_NET_MIRROR:
3421
      if getattr(self.op, "snode", None) is None:
3422
        raise errors.OpPrereqError("The networked disk templates need"
3423
                                   " a mirror node")
3424

    
3425
      snode_name = self.cfg.ExpandNodeName(self.op.snode)
3426
      if snode_name is None:
3427
        raise errors.OpPrereqError("Unknown secondary node '%s'" %
3428
                                   self.op.snode)
3429
      elif snode_name == pnode.name:
3430
        raise errors.OpPrereqError("The secondary node cannot be"
3431
                                   " the primary node.")
3432
      self.secondaries.append(snode_name)
3433

    
3434
    req_size = _ComputeDiskSize(self.op.disk_template,
3435
                                self.op.disk_size, self.op.swap_size)
3436

    
3437
    # Check lv size requirements
3438
    if req_size is not None:
3439
      nodenames = [pnode.name] + self.secondaries
3440
      nodeinfo = rpc.call_node_info(nodenames, self.cfg.GetVGName())
3441
      for node in nodenames:
3442
        info = nodeinfo.get(node, None)
3443
        if not info:
3444
          raise errors.OpPrereqError("Cannot get current information"
3445
                                     " from node '%s'" % nodeinfo)
3446
        vg_free = info.get('vg_free', None)
3447
        if not isinstance(vg_free, int):
3448
          raise errors.OpPrereqError("Can't compute free disk space on"
3449
                                     " node %s" % node)
3450
        if req_size > info['vg_free']:
3451
          raise errors.OpPrereqError("Not enough disk space on target node %s."
3452
                                     " %d MB available, %d MB required" %
3453
                                     (node, info['vg_free'], req_size))
3454

    
3455
    # os verification
3456
    os_obj = rpc.call_os_get(pnode.name, self.op.os_type)
3457
    if not os_obj:
3458
      raise errors.OpPrereqError("OS '%s' not in supported os list for"
3459
                                 " primary node"  % self.op.os_type)
3460

    
3461
    if self.op.kernel_path == constants.VALUE_NONE:
3462
      raise errors.OpPrereqError("Can't set instance kernel to none")
3463

    
3464

    
3465
    # bridge check on primary node
3466
    if not rpc.call_bridges_exist(self.pnode.name, [self.op.bridge]):
3467
      raise errors.OpPrereqError("target bridge '%s' does not exist on"
3468
                                 " destination node '%s'" %
3469
                                 (self.op.bridge, pnode.name))
3470

    
3471
    if self.op.start:
3472
      self.instance_status = 'up'
3473
    else:
3474
      self.instance_status = 'down'
3475

    
3476
  def Exec(self, feedback_fn):
3477
    """Create and add the instance to the cluster.
3478

3479
    """
3480
    instance = self.op.instance_name
3481
    pnode_name = self.pnode.name
3482

    
3483
    if self.op.mac == "auto":
3484
      mac_address = self.cfg.GenerateMAC()
3485
    else:
3486
      mac_address = self.op.mac
3487

    
3488
    nic = objects.NIC(bridge=self.op.bridge, mac=mac_address)
3489
    if self.inst_ip is not None:
3490
      nic.ip = self.inst_ip
3491

    
3492
    ht_kind = self.sstore.GetHypervisorType()
3493
    if ht_kind in constants.HTS_REQ_PORT:
3494
      network_port = self.cfg.AllocatePort()
3495
    else:
3496
      network_port = None
3497

    
3498
    # this is needed because os.path.join does not accept None arguments
3499
    if self.op.file_storage_dir is None:
3500
      string_file_storage_dir = ""
3501
    else:
3502
      string_file_storage_dir = self.op.file_storage_dir
3503

    
3504
    # build the full file storage dir path
3505
    file_storage_dir = os.path.normpath(os.path.join(
3506
                                        self.sstore.GetFileStorageDir(),
3507
                                        string_file_storage_dir, instance))
3508

    
3509

    
3510
    disks = _GenerateDiskTemplate(self.cfg,
3511
                                  self.op.disk_template,
3512
                                  instance, pnode_name,
3513
                                  self.secondaries, self.op.disk_size,
3514
                                  self.op.swap_size,
3515
                                  file_storage_dir,
3516
                                  self.op.file_driver)
3517

    
3518
    iobj = objects.Instance(name=instance, os=self.op.os_type,
3519
                            primary_node=pnode_name,
3520
                            memory=self.op.mem_size,
3521
                            vcpus=self.op.vcpus,
3522
                            nics=[nic], disks=disks,
3523
                            disk_template=self.op.disk_template,
3524
                            status=self.instance_status,
3525
                            network_port=network_port,
3526
                            kernel_path=self.op.kernel_path,
3527
                            initrd_path=self.op.initrd_path,
3528
                            hvm_boot_order=self.op.hvm_boot_order,
3529
                            )
3530

    
3531
    feedback_fn("* creating instance disks...")
3532
    if not _CreateDisks(self.cfg, iobj):
3533
      _RemoveDisks(iobj, self.cfg)
3534
      raise errors.OpExecError("Device creation failed, reverting...")
3535

    
3536
    feedback_fn("adding instance %s to cluster config" % instance)
3537

    
3538
    self.cfg.AddInstance(iobj)
3539

    
3540
    if self.op.wait_for_sync:
3541
      disk_abort = not _WaitForSync(self.cfg, iobj, self.proc)
3542
    elif iobj.disk_template in constants.DTS_NET_MIRROR:
3543
      # make sure the disks are not degraded (still sync-ing is ok)
3544
      time.sleep(15)
3545
      feedback_fn("* checking mirrors status")
3546
      disk_abort = not _WaitForSync(self.cfg, iobj, self.proc, oneshot=True)
3547
    else:
3548
      disk_abort = False
3549

    
3550
    if disk_abort:
3551
      _RemoveDisks(iobj, self.cfg)
3552
      self.cfg.RemoveInstance(iobj.name)
3553
      raise errors.OpExecError("There are some degraded disks for"
3554
                               " this instance")
3555

    
3556
    feedback_fn("creating os for instance %s on node %s" %
3557
                (instance, pnode_name))
3558

    
3559
    if iobj.disk_template != constants.DT_DISKLESS:
3560
      if self.op.mode == constants.INSTANCE_CREATE:
3561
        feedback_fn("* running the instance OS create scripts...")
3562
        if not rpc.call_instance_os_add(pnode_name, iobj, "sda", "sdb"):
3563
          raise errors.OpExecError("could not add os for instance %s"
3564
                                   " on node %s" %
3565
                                   (instance, pnode_name))
3566

    
3567
      elif self.op.mode == constants.INSTANCE_IMPORT:
3568
        feedback_fn("* running the instance OS import scripts...")
3569
        src_node = self.op.src_node
3570
        src_image = self.src_image
3571
        if not rpc.call_instance_os_import(pnode_name, iobj, "sda", "sdb",
3572
                                                src_node, src_image):
3573
          raise errors.OpExecError("Could not import os for instance"
3574
                                   " %s on node %s" %
3575
                                   (instance, pnode_name))
3576
      else:
3577
        # also checked in the prereq part
3578
        raise errors.ProgrammerError("Unknown OS initialization mode '%s'"
3579
                                     % self.op.mode)
3580

    
3581
    if self.op.start:
3582
      logger.Info("starting instance %s on node %s" % (instance, pnode_name))
3583
      feedback_fn("* starting instance...")
3584
      if not rpc.call_instance_start(pnode_name, iobj, None):
3585
        raise errors.OpExecError("Could not start instance")
3586

    
3587

    
3588
class LUConnectConsole(NoHooksLU):
3589
  """Connect to an instance's console.
3590

3591
  This is somewhat special in that it returns the command line that
3592
  you need to run on the master node in order to connect to the
3593
  console.
3594

3595
  """
3596
  _OP_REQP = ["instance_name"]
3597

    
3598
  def CheckPrereq(self):
3599
    """Check prerequisites.
3600

3601
    This checks that the instance is in the cluster.
3602

3603
    """
3604
    instance = self.cfg.GetInstanceInfo(
3605
      self.cfg.ExpandInstanceName(self.op.instance_name))
3606
    if instance is None:
3607
      raise errors.OpPrereqError("Instance '%s' not known" %
3608
                                 self.op.instance_name)
3609
    self.instance = instance
3610

    
3611
  def Exec(self, feedback_fn):
3612
    """Connect to the console of an instance
3613

3614
    """
3615
    instance = self.instance
3616
    node = instance.primary_node
3617

    
3618
    node_insts = rpc.call_instance_list([node])[node]
3619
    if node_insts is False:
3620
      raise errors.OpExecError("Can't connect to node %s." % node)
3621

    
3622
    if instance.name not in node_insts:
3623
      raise errors.OpExecError("Instance %s is not running." % instance.name)
3624

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

    
3627
    hyper = hypervisor.GetHypervisor()
3628
    console_cmd = hyper.GetShellCommandForConsole(instance)
3629

    
3630
    # build ssh cmdline
3631
    return self.ssh.BuildCmd(node, "root", console_cmd, batch=True, tty=True)
3632

    
3633

    
3634
class LUReplaceDisks(LogicalUnit):
3635
  """Replace the disks of an instance.
3636

3637
  """
3638
  HPATH = "mirrors-replace"
3639
  HTYPE = constants.HTYPE_INSTANCE
3640
  _OP_REQP = ["instance_name", "mode", "disks"]
3641

    
3642
  def _RunAllocator(self):
3643
    """Compute a new secondary node using an IAllocator.
3644

3645
    """
3646
    ial = IAllocator(self.cfg, self.sstore,
3647
                     mode=constants.IALLOCATOR_MODE_RELOC,
3648
                     name=self.op.instance_name,
3649
                     relocate_from=[self.sec_node])
3650

    
3651
    ial.Run(self.op.iallocator)
3652

    
3653
    if not ial.success:
3654
      raise errors.OpPrereqError("Can't compute nodes using"
3655
                                 " iallocator '%s': %s" % (self.op.iallocator,
3656
                                                           ial.info))
3657
    if len(ial.nodes) != ial.required_nodes:
3658
      raise errors.OpPrereqError("iallocator '%s' returned invalid number"
3659
                                 " of nodes (%s), required %s" %
3660
                                 (len(ial.nodes), ial.required_nodes))
3661
    self.op.remote_node = ial.nodes[0]
3662
    logger.ToStdout("Selected new secondary for the instance: %s" %
3663
                    self.op.remote_node)
3664

    
3665
  def BuildHooksEnv(self):
3666
    """Build hooks env.
3667

3668
    This runs on the master, the primary and all the secondaries.
3669

3670
    """
3671
    env = {
3672
      "MODE": self.op.mode,
3673
      "NEW_SECONDARY": self.op.remote_node,
3674
      "OLD_SECONDARY": self.instance.secondary_nodes[0],
3675
      }
3676
    env.update(_BuildInstanceHookEnvByObject(self.instance))
3677
    nl = [
3678
      self.sstore.GetMasterNode(),
3679
      self.instance.primary_node,
3680
      ]
3681
    if self.op.remote_node is not None:
3682
      nl.append(self.op.remote_node)
3683
    return env, nl, nl
3684

    
3685
  def CheckPrereq(self):
3686
    """Check prerequisites.
3687

3688
    This checks that the instance is in the cluster.
3689

3690
    """
3691
    if not hasattr(self.op, "remote_node"):
3692
      self.op.remote_node = None
3693

    
3694
    instance = self.cfg.GetInstanceInfo(
3695
      self.cfg.ExpandInstanceName(self.op.instance_name))
3696
    if instance is None:
3697
      raise errors.OpPrereqError("Instance '%s' not known" %
3698
                                 self.op.instance_name)
3699
    self.instance = instance
3700
    self.op.instance_name = instance.name
3701

    
3702
    if instance.disk_template not in constants.DTS_NET_MIRROR:
3703
      raise errors.OpPrereqError("Instance's disk layout is not"
3704
                                 " network mirrored.")
3705

    
3706
    if len(instance.secondary_nodes) != 1:
3707
      raise errors.OpPrereqError("The instance has a strange layout,"
3708
                                 " expected one secondary but found %d" %
3709
                                 len(instance.secondary_nodes))
3710

    
3711
    self.sec_node = instance.secondary_nodes[0]
3712

    
3713
    ia_name = getattr(self.op, "iallocator", None)
3714
    if ia_name is not None:
3715
      if self.op.remote_node is not None:
3716
        raise errors.OpPrereqError("Give either the iallocator or the new"
3717
                                   " secondary, not both")
3718
      self.op.remote_node = self._RunAllocator()
3719

    
3720
    remote_node = self.op.remote_node
3721
    if remote_node is not None:
3722
      remote_node = self.cfg.ExpandNodeName(remote_node)
3723
      if remote_node is None:
3724
        raise errors.OpPrereqError("Node '%s' not known" %
3725
                                   self.op.remote_node)
3726
      self.remote_node_info = self.cfg.GetNodeInfo(remote_node)
3727
    else:
3728
      self.remote_node_info = None
3729
    if remote_node == instance.primary_node:
3730
      raise errors.OpPrereqError("The specified node is the primary node of"
3731
                                 " the instance.")
3732
    elif remote_node == self.sec_node:
3733
      if self.op.mode == constants.REPLACE_DISK_SEC:
3734
        # this is for DRBD8, where we can't execute the same mode of
3735
        # replacement as for drbd7 (no different port allocated)
3736
        raise errors.OpPrereqError("Same secondary given, cannot execute"
3737
                                   " replacement")
3738
      # the user gave the current secondary, switch to
3739
      # 'no-replace-secondary' mode for drbd7
3740
      remote_node = None
3741
    if (instance.disk_template == constants.DT_REMOTE_RAID1 and
3742
        self.op.mode != constants.REPLACE_DISK_ALL):
3743
      raise errors.OpPrereqError("Template 'remote_raid1' only allows all"
3744
                                 " disks replacement, not individual ones")
3745
    if instance.disk_template == constants.DT_DRBD8:
3746
      if (self.op.mode == constants.REPLACE_DISK_ALL and
3747
          remote_node is not None):
3748
        # switch to replace secondary mode
3749
        self.op.mode = constants.REPLACE_DISK_SEC
3750

    
3751
      if self.op.mode == constants.REPLACE_DISK_ALL:
3752
        raise errors.OpPrereqError("Template 'drbd' only allows primary or"
3753
                                   " secondary disk replacement, not"
3754
                                   " both at once")
3755
      elif self.op.mode == constants.REPLACE_DISK_PRI:
3756
        if remote_node is not None:
3757
          raise errors.OpPrereqError("Template 'drbd' does not allow changing"
3758
                                     " the secondary while doing a primary"
3759
                                     " node disk replacement")
3760
        self.tgt_node = instance.primary_node
3761
        self.oth_node = instance.secondary_nodes[0]
3762
      elif self.op.mode == constants.REPLACE_DISK_SEC:
3763
        self.new_node = remote_node # this can be None, in which case
3764
                                    # we don't change the secondary
3765
        self.tgt_node = instance.secondary_nodes[0]
3766
        self.oth_node = instance.primary_node
3767
      else:
3768
        raise errors.ProgrammerError("Unhandled disk replace mode")
3769

    
3770
    for name in self.op.disks:
3771
      if instance.FindDisk(name) is None:
3772
        raise errors.OpPrereqError("Disk '%s' not found for instance '%s'" %
3773
                                   (name, instance.name))
3774
    self.op.remote_node = remote_node
3775

    
3776
  def _ExecRR1(self, feedback_fn):
3777
    """Replace the disks of an instance.
3778

3779
    """
3780
    instance = self.instance
3781
    iv_names = {}
3782
    # start of work
3783
    if self.op.remote_node is None:
3784
      remote_node = self.sec_node
3785
    else:
3786
      remote_node = self.op.remote_node
3787
    cfg = self.cfg
3788
    for dev in instance.disks:
3789
      size = dev.size
3790
      lv_names = [".%s_%s" % (dev.iv_name, suf) for suf in ["data", "meta"]]
3791
      names = _GenerateUniqueNames(cfg, lv_names)
3792
      new_drbd = _GenerateMDDRBDBranch(cfg, instance.primary_node,
3793
                                       remote_node, size, names)
3794
      iv_names[dev.iv_name] = (dev, dev.children[0], new_drbd)
3795
      logger.Info("adding new mirror component on secondary for %s" %
3796
                  dev.iv_name)
3797
      #HARDCODE
3798
      if not _CreateBlockDevOnSecondary(cfg, remote_node, instance,
3799
                                        new_drbd, False,
3800
                                        _GetInstanceInfoText(instance)):
3801
        raise errors.OpExecError("Failed to create new component on secondary"
3802
                                 " node %s. Full abort, cleanup manually!" %
3803
                                 remote_node)
3804

    
3805
      logger.Info("adding new mirror component on primary")
3806
      #HARDCODE
3807
      if not _CreateBlockDevOnPrimary(cfg, instance.primary_node,
3808
                                      instance, new_drbd,
3809
                                      _GetInstanceInfoText(instance)):
3810
        # remove secondary dev
3811
        cfg.SetDiskID(new_drbd, remote_node)
3812
        rpc.call_blockdev_remove(remote_node, new_drbd)
3813
        raise errors.OpExecError("Failed to create volume on primary!"
3814
                                 " Full abort, cleanup manually!!")
3815

    
3816
      # the device exists now
3817
      # call the primary node to add the mirror to md
3818
      logger.Info("adding new mirror component to md")
3819
      if not rpc.call_blockdev_addchildren(instance.primary_node, dev,
3820
                                           [new_drbd]):
3821
        logger.Error("Can't add mirror compoment to md!")
3822
        cfg.SetDiskID(new_drbd, remote_node)
3823
        if not rpc.call_blockdev_remove(remote_node, new_drbd):
3824
          logger.Error("Can't rollback on secondary")
3825
        cfg.SetDiskID(new_drbd, instance.primary_node)
3826
        if not rpc.call_blockdev_remove(instance.primary_node, new_drbd):
3827
          logger.Error("Can't rollback on primary")
3828
        raise errors.OpExecError("Full abort, cleanup manually!!")
3829

    
3830
      dev.children.append(new_drbd)
3831
      cfg.AddInstance(instance)
3832

    
3833
    # this can fail as the old devices are degraded and _WaitForSync
3834
    # does a combined result over all disks, so we don't check its
3835
    # return value
3836
    _WaitForSync(cfg, instance, self.proc, unlock=True)
3837

    
3838
    # so check manually all the devices
3839
    for name in iv_names:
3840
      dev, child, new_drbd = iv_names[name]
3841
      cfg.SetDiskID(dev, instance.primary_node)
3842
      is_degr = rpc.call_blockdev_find(instance.primary_node, dev)[5]
3843
      if is_degr:
3844
        raise errors.OpExecError("MD device %s is degraded!" % name)
3845
      cfg.SetDiskID(new_drbd, instance.primary_node)
3846
      is_degr = rpc.call_blockdev_find(instance.primary_node, new_drbd)[5]
3847
      if is_degr:
3848
        raise errors.OpExecError("New drbd device %s is degraded!" % name)
3849

    
3850
    for name in iv_names:
3851
      dev, child, new_drbd = iv_names[name]
3852
      logger.Info("remove mirror %s component" % name)
3853
      cfg.SetDiskID(dev, instance.primary_node)
3854
      if not rpc.call_blockdev_removechildren(instance.primary_node,
3855
                                              dev, [child]):
3856
        logger.Error("Can't remove child from mirror, aborting"
3857
                     " *this device cleanup*.\nYou need to cleanup manually!!")
3858
        continue
3859

    
3860
      for node in child.logical_id[:2]:
3861
        logger.Info("remove child device on %s" % node)
3862
        cfg.SetDiskID(child, node)
3863
        if not rpc.call_blockdev_remove(node, child):
3864
          logger.Error("Warning: failed to remove device from node %s,"
3865
                       " continuing operation." % node)
3866

    
3867
      dev.children.remove(child)
3868

    
3869
      cfg.AddInstance(instance)
3870

    
3871
  def _ExecD8DiskOnly(self, feedback_fn):
3872
    """Replace a disk on the primary or secondary for dbrd8.
3873

3874
    The algorithm for replace is quite complicated:
3875
      - for each disk to be replaced:
3876
        - create new LVs on the target node with unique names
3877
        - detach old LVs from the drbd device
3878
        - rename old LVs to name_replaced.<time_t>
3879
        - rename new LVs to old LVs
3880
        - attach the new LVs (with the old names now) to the drbd device
3881
      - wait for sync across all devices
3882
      - for each modified disk:
3883
        - remove old LVs (which have the name name_replaces.<time_t>)
3884

3885
    Failures are not very well handled.
3886

3887
    """
3888
    steps_total = 6
3889
    warning, info = (self.proc.LogWarning, self.proc.LogInfo)
3890
    instance = self.instance
3891
    iv_names = {}
3892
    vgname = self.cfg.GetVGName()
3893
    # start of work
3894
    cfg = self.cfg
3895
    tgt_node = self.tgt_node
3896
    oth_node = self.oth_node
3897

    
3898
    # Step: check device activation
3899
    self.proc.LogStep(1, steps_total, "check device existence")
3900
    info("checking volume groups")
3901
    my_vg = cfg.GetVGName()
3902
    results = rpc.call_vg_list([oth_node, tgt_node])
3903
    if not results:
3904
      raise errors.OpExecError("Can't list volume groups on the nodes")
3905
    for node in oth_node, tgt_node:
3906
      res = results.get(node, False)
3907
      if not res or my_vg not in res:
3908
        raise errors.OpExecError("Volume group '%s' not found on %s" %
3909
                                 (my_vg, node))
3910
    for dev in instance.disks:
3911
      if not dev.iv_name in self.op.disks:
3912
        continue
3913
      for node in tgt_node, oth_node:
3914
        info("checking %s on %s" % (dev.iv_name, node))
3915
        cfg.SetDiskID(dev, node)
3916
        if not rpc.call_blockdev_find(node, dev):
3917
          raise errors.OpExecError("Can't find device %s on node %s" %
3918
                                   (dev.iv_name, node))
3919

    
3920
    # Step: check other node consistency
3921
    self.proc.LogStep(2, steps_total, "check peer consistency")
3922
    for dev in instance.disks:
3923
      if not dev.iv_name in self.op.disks:
3924
        continue
3925
      info("checking %s consistency on %s" % (dev.iv_name, oth_node))
3926
      if not _CheckDiskConsistency(self.cfg, dev, oth_node,
3927
                                   oth_node==instance.primary_node):
3928
        raise errors.OpExecError("Peer node (%s) has degraded storage, unsafe"
3929
                                 " to replace disks on this node (%s)" %
3930
                                 (oth_node, tgt_node))
3931

    
3932
    # Step: create new storage
3933
    self.proc.LogStep(3, steps_total, "allocate new storage")
3934
    for dev in instance.disks:
3935
      if not dev.iv_name in self.op.disks:
3936
        continue
3937
      size = dev.size
3938
      cfg.SetDiskID(dev, tgt_node)
3939
      lv_names = [".%s_%s" % (dev.iv_name, suf) for suf in ["data", "meta"]]
3940
      names = _GenerateUniqueNames(cfg, lv_names)
3941
      lv_data = objects.Disk(dev_type=constants.LD_LV, size=size,
3942
                             logical_id=(vgname, names[0]))
3943
      lv_meta = objects.Disk(dev_type=constants.LD_LV, size=128,
3944
                             logical_id=(vgname, names[1]))
3945
      new_lvs = [lv_data, lv_meta]
3946
      old_lvs = dev.children
3947
      iv_names[dev.iv_name] = (dev, old_lvs, new_lvs)
3948
      info("creating new local storage on %s for %s" %
3949
           (tgt_node, dev.iv_name))
3950
      # since we *always* want to create this LV, we use the
3951
      # _Create...OnPrimary (which forces the creation), even if we
3952
      # are talking about the secondary node
3953
      for new_lv in new_lvs:
3954
        if not _CreateBlockDevOnPrimary(cfg, tgt_node, instance, new_lv,
3955
                                        _GetInstanceInfoText(instance)):
3956
          raise errors.OpExecError("Failed to create new LV named '%s' on"
3957
                                   " node '%s'" %
3958
                                   (new_lv.logical_id[1], tgt_node))
3959

    
3960
    # Step: for each lv, detach+rename*2+attach
3961
    self.proc.LogStep(4, steps_total, "change drbd configuration")
3962
    for dev, old_lvs, new_lvs in iv_names.itervalues():
3963
      info("detaching %s drbd from local storage" % dev.iv_name)
3964
      if not rpc.call_blockdev_removechildren(tgt_node, dev, old_lvs):
3965
        raise errors.OpExecError("Can't detach drbd from local storage on node"
3966
                                 " %s for device %s" % (tgt_node, dev.iv_name))
3967
      #dev.children = []
3968
      #cfg.Update(instance)
3969

    
3970
      # ok, we created the new LVs, so now we know we have the needed
3971
      # storage; as such, we proceed on the target node to rename
3972
      # old_lv to _old, and new_lv to old_lv; note that we rename LVs
3973
      # using the assumption that logical_id == physical_id (which in
3974
      # turn is the unique_id on that node)
3975

    
3976
      # FIXME(iustin): use a better name for the replaced LVs
3977
      temp_suffix = int(time.time())
3978
      ren_fn = lambda d, suff: (d.physical_id[0],
3979
                                d.physical_id[1] + "_replaced-%s" % suff)
3980
      # build the rename list based on what LVs exist on the node
3981
      rlist = []
3982
      for to_ren in old_lvs:
3983
        find_res = rpc.call_blockdev_find(tgt_node, to_ren)
3984
        if find_res is not None: # device exists
3985
          rlist.append((to_ren, ren_fn(to_ren, temp_suffix)))
3986

    
3987
      info("renaming the old LVs on the target node")
3988
      if not rpc.call_blockdev_rename(tgt_node, rlist):
3989
        raise errors.OpExecError("Can't rename old LVs on node %s" % tgt_node)
3990
      # now we rename the new LVs to the old LVs
3991