Statistics
| Branch: | Tag: | Revision:

root / lib / cmdlib.py @ 83120a01

History | View | Annotate | Download (109.5 kB)

1
#!/usr/bin/python
2
#
3

    
4
# Copyright (C) 2006, 2007 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 socket
30
import time
31
import tempfile
32
import re
33
import platform
34

    
35
from ganeti import rpc
36
from ganeti import ssh
37
from ganeti import logger
38
from ganeti import utils
39
from ganeti import errors
40
from ganeti import hypervisor
41
from ganeti import config
42
from ganeti import constants
43
from ganeti import objects
44
from ganeti import opcodes
45
from ganeti import ssconf
46

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

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

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

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

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

72
    """
73
    self.processor = processor
74
    self.op = op
75
    self.cfg = cfg
76
    self.sstore = sstore
77
    for attr_name in self._OP_REQP:
78
      attr_val = getattr(op, attr_name, None)
79
      if attr_val is None:
80
        raise errors.OpPrereqError, ("Required parameter '%s' missing" %
81
                                     attr_name)
82
    if self.REQ_CLUSTER:
83
      if not cfg.IsCluster():
84
        raise errors.OpPrereqError, ("Cluster not initialized yet,"
85
                                     " use 'gnt-cluster init' first.")
86
      if self.REQ_MASTER:
87
        master = sstore.GetMasterNode()
88
        if master != socket.gethostname():
89
          raise errors.OpPrereqError, ("Commands must be run on the master"
90
                                       " node %s" % master)
91

    
92
  def CheckPrereq(self):
93
    """Check prerequisites for this LU.
94

95
    This method should check that the prerequisites for the execution
96
    of this LU are fulfilled. It can do internode communication, but
97
    it should be idempotent - no cluster or system changes are
98
    allowed.
99

100
    The method should raise errors.OpPrereqError in case something is
101
    not fulfilled. Its return value is ignored.
102

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

108
    """
109
    raise NotImplementedError
110

    
111
  def Exec(self, feedback_fn):
112
    """Execute the LU.
113

114
    This method should implement the actual work. It should raise
115
    errors.OpExecError for failures that are somewhat dealt with in
116
    code, or expected.
117

118
    """
119
    raise NotImplementedError
120

    
121
  def BuildHooksEnv(self):
122
    """Build hooks environment for this LU.
123

124
    This method should return a three-node tuple consisting of: a dict
125
    containing the environment that will be used for running the
126
    specific hook for this LU, a list of node names on which the hook
127
    should run before the execution, and a list of node names on which
128
    the hook should run after the execution.
129

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

135
    As for the node lists, the master should not be included in the
136
    them, as it will be added by the hooks runner in case this LU
137
    requires a cluster to run on (otherwise we don't have a node
138
    list). No nodes should be returned as an empty list (and not
139
    None).
140

141
    Note that if the HPATH for a LU class is None, this function will
142
    not be called.
143

144
    """
145
    raise NotImplementedError
146

    
147

    
148
class NoHooksLU(LogicalUnit):
149
  """Simple LU which runs no hooks.
150

151
  This LU is intended as a parent for other LogicalUnits which will
152
  run no hooks, in order to reduce duplicate code.
153

154
  """
155
  HPATH = None
156
  HTYPE = None
157

    
158
  def BuildHooksEnv(self):
159
    """Build hooks env.
160

161
    This is a no-op, since we don't run hooks.
162

163
    """
164
    return
165

    
166

    
167
def _GetWantedNodes(lu, nodes):
168
  """Returns list of checked and expanded nodes.
169

170
  Args:
171
    nodes: List of nodes (strings) or None for all
172

173
  """
174
  if nodes is not None and not isinstance(nodes, list):
175
    raise errors.OpPrereqError, "Invalid argument type 'nodes'"
176

    
177
  if nodes:
178
    wanted_nodes = []
179

    
180
    for name in nodes:
181
      node = lu.cfg.GetNodeInfo(lu.cfg.ExpandNodeName(name))
182
      if node is None:
183
        raise errors.OpPrereqError, ("No such node name '%s'" % name)
184
    wanted_nodes.append(node)
185

    
186
    return wanted_nodes
187
  else:
188
    return [lu.cfg.GetNodeInfo(name) for name in lu.cfg.GetNodeList()]
189

    
190

    
191
def _CheckOutputFields(static, dynamic, selected):
192
  """Checks whether all selected fields are valid.
193

194
  Args:
195
    static: Static fields
196
    dynamic: Dynamic fields
197

198
  """
199
  static_fields = frozenset(static)
200
  dynamic_fields = frozenset(dynamic)
201

    
202
  all_fields = static_fields | dynamic_fields
203

    
204
  if not all_fields.issuperset(selected):
205
    raise errors.OpPrereqError, ("Unknown output fields selected: %s"
206
                                 % ",".join(frozenset(selected).
207
                                            difference(all_fields)))
208

    
209

    
210
def _UpdateEtcHosts(fullnode, ip):
211
  """Ensure a node has a correct entry in /etc/hosts.
212

213
  Args:
214
    fullnode - Fully qualified domain name of host. (str)
215
    ip       - IPv4 address of host (str)
216

217
  """
218
  node = fullnode.split(".", 1)[0]
219

    
220
  f = open('/etc/hosts', 'r+')
221

    
222
  inthere = False
223

    
224
  save_lines = []
225
  add_lines = []
226
  removed = False
227

    
228
  while True:
229
    rawline = f.readline()
230

    
231
    if not rawline:
232
      # End of file
233
      break
234

    
235
    line = rawline.split('\n')[0]
236

    
237
    # Strip off comments
238
    line = line.split('#')[0]
239

    
240
    if not line:
241
      # Entire line was comment, skip
242
      save_lines.append(rawline)
243
      continue
244

    
245
    fields = line.split()
246

    
247
    haveall = True
248
    havesome = False
249
    for spec in [ ip, fullnode, node ]:
250
      if spec not in fields:
251
        haveall = False
252
      if spec in fields:
253
        havesome = True
254

    
255
    if haveall:
256
      inthere = True
257
      save_lines.append(rawline)
258
      continue
259

    
260
    if havesome and not haveall:
261
      # Line (old, or manual?) which is missing some.  Remove.
262
      removed = True
263
      continue
264

    
265
    save_lines.append(rawline)
266

    
267
  if not inthere:
268
    add_lines.append('%s\t%s %s\n' % (ip, fullnode, node))
269

    
270
  if removed:
271
    if add_lines:
272
      save_lines = save_lines + add_lines
273

    
274
    # We removed a line, write a new file and replace old.
275
    fd, tmpname = tempfile.mkstemp('tmp', 'hosts_', '/etc')
276
    newfile = os.fdopen(fd, 'w')
277
    newfile.write(''.join(save_lines))
278
    newfile.close()
279
    os.rename(tmpname, '/etc/hosts')
280

    
281
  elif add_lines:
282
    # Simply appending a new line will do the trick.
283
    f.seek(0, 2)
284
    for add in add_lines:
285
      f.write(add)
286

    
287
  f.close()
288

    
289

    
290
def _UpdateKnownHosts(fullnode, ip, pubkey):
291
  """Ensure a node has a correct known_hosts entry.
292

293
  Args:
294
    fullnode - Fully qualified domain name of host. (str)
295
    ip       - IPv4 address of host (str)
296
    pubkey   - the public key of the cluster
297

298
  """
299
  if os.path.exists('/etc/ssh/ssh_known_hosts'):
300
    f = open('/etc/ssh/ssh_known_hosts', 'r+')
301
  else:
302
    f = open('/etc/ssh/ssh_known_hosts', 'w+')
303

    
304
  inthere = False
305

    
306
  save_lines = []
307
  add_lines = []
308
  removed = False
309

    
310
  while True:
311
    rawline = f.readline()
312
    logger.Debug('read %s' % (repr(rawline),))
313

    
314
    if not rawline:
315
      # End of file
316
      break
317

    
318
    line = rawline.split('\n')[0]
319

    
320
    parts = line.split(' ')
321
    fields = parts[0].split(',')
322
    key = parts[2]
323

    
324
    haveall = True
325
    havesome = False
326
    for spec in [ ip, fullnode ]:
327
      if spec not in fields:
328
        haveall = False
329
      if spec in fields:
330
        havesome = True
331

    
332
    logger.Debug("key, pubkey = %s." % (repr((key, pubkey)),))
333
    if haveall and key == pubkey:
334
      inthere = True
335
      save_lines.append(rawline)
336
      logger.Debug("Keeping known_hosts '%s'." % (repr(rawline),))
337
      continue
338

    
339
    if havesome and (not haveall or key != pubkey):
340
      removed = True
341
      logger.Debug("Discarding known_hosts '%s'." % (repr(rawline),))
342
      continue
343

    
344
    save_lines.append(rawline)
345

    
346
  if not inthere:
347
    add_lines.append('%s,%s ssh-rsa %s\n' % (fullnode, ip, pubkey))
348
    logger.Debug("Adding known_hosts '%s'." % (repr(add_lines[-1]),))
349

    
350
  if removed:
351
    save_lines = save_lines + add_lines
352

    
353
    # Write a new file and replace old.
354
    fd, tmpname = tempfile.mkstemp('tmp', 'ssh_known_hosts_', '/etc/ssh')
355
    newfile = os.fdopen(fd, 'w')
356
    newfile.write(''.join(save_lines))
357
    newfile.close()
358
    logger.Debug("Wrote new known_hosts.")
359
    os.rename(tmpname, '/etc/ssh/ssh_known_hosts')
360

    
361
  elif add_lines:
362
    # Simply appending a new line will do the trick.
363
    f.seek(0, 2)
364
    for add in add_lines:
365
      f.write(add)
366

    
367
  f.close()
368

    
369

    
370
def _HasValidVG(vglist, vgname):
371
  """Checks if the volume group list is valid.
372

373
  A non-None return value means there's an error, and the return value
374
  is the error message.
375

376
  """
377
  vgsize = vglist.get(vgname, None)
378
  if vgsize is None:
379
    return "volume group '%s' missing" % vgname
380
  elif vgsize < 20480:
381
    return ("volume group '%s' too small (20480MiB required, %dMib found)" %
382
            (vgname, vgsize))
383
  return None
384

    
385

    
386
def _InitSSHSetup(node):
387
  """Setup the SSH configuration for the cluster.
388

389

390
  This generates a dsa keypair for root, adds the pub key to the
391
  permitted hosts and adds the hostkey to its own known hosts.
392

393
  Args:
394
    node: the name of this host as a fqdn
395

396
  """
397
  utils.RemoveFile('/root/.ssh/known_hosts')
398

    
399
  if os.path.exists('/root/.ssh/id_dsa'):
400
    utils.CreateBackup('/root/.ssh/id_dsa')
401
  if os.path.exists('/root/.ssh/id_dsa.pub'):
402
    utils.CreateBackup('/root/.ssh/id_dsa.pub')
403

    
404
  utils.RemoveFile('/root/.ssh/id_dsa')
405
  utils.RemoveFile('/root/.ssh/id_dsa.pub')
406

    
407
  result = utils.RunCmd(["ssh-keygen", "-t", "dsa",
408
                         "-f", "/root/.ssh/id_dsa",
409
                         "-q", "-N", ""])
410
  if result.failed:
411
    raise errors.OpExecError, ("could not generate ssh keypair, error %s" %
412
                               result.output)
413

    
414
  f = open('/root/.ssh/id_dsa.pub', 'r')
415
  try:
416
    utils.AddAuthorizedKey('/root/.ssh/authorized_keys', f.read(8192))
417
  finally:
418
    f.close()
419

    
420

    
421
def _InitGanetiServerSetup(ss):
422
  """Setup the necessary configuration for the initial node daemon.
423

424
  This creates the nodepass file containing the shared password for
425
  the cluster and also generates the SSL certificate.
426

427
  """
428
  # Create pseudo random password
429
  randpass = sha.new(os.urandom(64)).hexdigest()
430
  # and write it into sstore
431
  ss.SetKey(ss.SS_NODED_PASS, randpass)
432

    
433
  result = utils.RunCmd(["openssl", "req", "-new", "-newkey", "rsa:1024",
434
                         "-days", str(365*5), "-nodes", "-x509",
435
                         "-keyout", constants.SSL_CERT_FILE,
436
                         "-out", constants.SSL_CERT_FILE, "-batch"])
437
  if result.failed:
438
    raise errors.OpExecError, ("could not generate server ssl cert, command"
439
                               " %s had exitcode %s and error message %s" %
440
                               (result.cmd, result.exit_code, result.output))
441

    
442
  os.chmod(constants.SSL_CERT_FILE, 0400)
443

    
444
  result = utils.RunCmd([constants.NODE_INITD_SCRIPT, "restart"])
445

    
446
  if result.failed:
447
    raise errors.OpExecError, ("could not start the node daemon, command %s"
448
                               " had exitcode %s and error %s" %
449
                               (result.cmd, result.exit_code, result.output))
450

    
451

    
452
class LUInitCluster(LogicalUnit):
453
  """Initialise the cluster.
454

455
  """
456
  HPATH = "cluster-init"
457
  HTYPE = constants.HTYPE_CLUSTER
458
  _OP_REQP = ["cluster_name", "hypervisor_type", "vg_name", "mac_prefix",
459
              "def_bridge", "master_netdev"]
460
  REQ_CLUSTER = False
461

    
462
  def BuildHooksEnv(self):
463
    """Build hooks env.
464

465
    Notes: Since we don't require a cluster, we must manually add
466
    ourselves in the post-run node list.
467

468
    """
469

    
470
    env = {"CLUSTER": self.op.cluster_name,
471
           "MASTER": self.hostname['hostname_full']}
472
    return env, [], [self.hostname['hostname_full']]
473

    
474
  def CheckPrereq(self):
475
    """Verify that the passed name is a valid one.
476

477
    """
478
    if config.ConfigWriter.IsCluster():
479
      raise errors.OpPrereqError, ("Cluster is already initialised")
480

    
481
    hostname_local = socket.gethostname()
482
    self.hostname = hostname = utils.LookupHostname(hostname_local)
483
    if not hostname:
484
      raise errors.OpPrereqError, ("Cannot resolve my own hostname ('%s')" %
485
                                   hostname_local)
486

    
487
    self.clustername = clustername = utils.LookupHostname(self.op.cluster_name)
488
    if not clustername:
489
      raise errors.OpPrereqError, ("Cannot resolve given cluster name ('%s')"
490
                                   % self.op.cluster_name)
491

    
492
    result = utils.RunCmd(["fping", "-S127.0.0.1", "-q", hostname['ip']])
493
    if result.failed:
494
      raise errors.OpPrereqError, ("Inconsistency: this host's name resolves"
495
                                   " to %s,\nbut this ip address does not"
496
                                   " belong to this host."
497
                                   " Aborting." % hostname['ip'])
498

    
499
    secondary_ip = getattr(self.op, "secondary_ip", None)
500
    if secondary_ip and not utils.IsValidIP(secondary_ip):
501
      raise errors.OpPrereqError, ("Invalid secondary ip given")
502
    if secondary_ip and secondary_ip != hostname['ip']:
503
      result = utils.RunCmd(["fping", "-S127.0.0.1", "-q", secondary_ip])
504
      if result.failed:
505
        raise errors.OpPrereqError, ("You gave %s as secondary IP,\n"
506
                                     "but it does not belong to this host." %
507
                                     secondary_ip)
508
    self.secondary_ip = secondary_ip
509

    
510
    # checks presence of the volume group given
511
    vgstatus = _HasValidVG(utils.ListVolumeGroups(), self.op.vg_name)
512

    
513
    if vgstatus:
514
      raise errors.OpPrereqError, ("Error: %s" % vgstatus)
515

    
516
    if not re.match("^[0-9a-z]{2}:[0-9a-z]{2}:[0-9a-z]{2}$",
517
                    self.op.mac_prefix):
518
      raise errors.OpPrereqError, ("Invalid mac prefix given '%s'" %
519
                                   self.op.mac_prefix)
520

    
521
    if self.op.hypervisor_type not in hypervisor.VALID_HTYPES:
522
      raise errors.OpPrereqError, ("Invalid hypervisor type given '%s'" %
523
                                   self.op.hypervisor_type)
524

    
525
    result = utils.RunCmd(["ip", "link", "show", "dev", self.op.master_netdev])
526
    if result.failed:
527
      raise errors.OpPrereqError, ("Invalid master netdev given (%s): '%s'" %
528
                                   (self.op.master_netdev, result.output))
529

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

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

    
537
    # set up the simple store
538
    ss = ssconf.SimpleStore()
539
    ss.SetKey(ss.SS_HYPERVISOR, self.op.hypervisor_type)
540
    ss.SetKey(ss.SS_MASTER_NODE, hostname['hostname_full'])
541
    ss.SetKey(ss.SS_MASTER_IP, clustername['ip'])
542
    ss.SetKey(ss.SS_MASTER_NETDEV, self.op.master_netdev)
543

    
544
    # set up the inter-node password and certificate
545
    _InitGanetiServerSetup(ss)
546

    
547
    # start the master ip
548
    rpc.call_node_start_master(hostname['hostname_full'])
549

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

    
558
    _UpdateEtcHosts(hostname['hostname_full'],
559
                    hostname['ip'],
560
                    )
561

    
562
    _UpdateKnownHosts(hostname['hostname_full'],
563
                      hostname['ip'],
564
                      sshkey,
565
                      )
566

    
567
    _InitSSHSetup(hostname['hostname'])
568

    
569
    # init of cluster config file
570
    cfgw = config.ConfigWriter()
571
    cfgw.InitConfig(hostname['hostname'], hostname['ip'], self.secondary_ip,
572
                    clustername['hostname'], sshkey, self.op.mac_prefix,
573
                    self.op.vg_name, self.op.def_bridge)
574

    
575

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

579
  """
580
  _OP_REQP = []
581

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

585
    This checks whether the cluster is empty.
586

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

589
    """
590
    master = self.sstore.GetMasterNode()
591

    
592
    nodelist = self.cfg.GetNodeList()
593
    if len(nodelist) > 0 and nodelist != [master]:
594
      raise errors.OpPrereqError, ("There are still %d node(s) in "
595
                                   "this cluster." % (len(nodelist) - 1))
596

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

600
    """
601
    utils.CreateBackup('/root/.ssh/id_dsa')
602
    utils.CreateBackup('/root/.ssh/id_dsa.pub')
603
    rpc.call_node_leave_cluster(self.sstore.GetMasterNode())
604

    
605

    
606
class LUVerifyCluster(NoHooksLU):
607
  """Verifies the cluster status.
608

609
  """
610
  _OP_REQP = []
611

    
612
  def _VerifyNode(self, node, file_list, local_cksum, vglist, node_result,
613
                  remote_version, feedback_fn):
614
    """Run multiple tests against a node.
615

616
    Test list:
617
      - compares ganeti version
618
      - checks vg existance and size > 20G
619
      - checks config file checksum
620
      - checks ssh to other nodes
621

622
    Args:
623
      node: name of the node to check
624
      file_list: required list of files
625
      local_cksum: dictionary of local files and their checksums
626
    """
627
    # compares ganeti version
628
    local_version = constants.PROTOCOL_VERSION
629
    if not remote_version:
630
      feedback_fn(" - ERROR: connection to %s failed" % (node))
631
      return True
632

    
633
    if local_version != remote_version:
634
      feedback_fn("  - ERROR: sw version mismatch: master %s, node(%s) %s" %
635
                      (local_version, node, remote_version))
636
      return True
637

    
638
    # checks vg existance and size > 20G
639

    
640
    bad = False
641
    if not vglist:
642
      feedback_fn("  - ERROR: unable to check volume groups on node %s." %
643
                      (node,))
644
      bad = True
645
    else:
646
      vgstatus = _HasValidVG(vglist, self.cfg.GetVGName())
647
      if vgstatus:
648
        feedback_fn("  - ERROR: %s on node %s" % (vgstatus, node))
649
        bad = True
650

    
651
    # checks config file checksum
652
    # checks ssh to any
653

    
654
    if 'filelist' not in node_result:
655
      bad = True
656
      feedback_fn("  - ERROR: node hasn't returned file checksum data")
657
    else:
658
      remote_cksum = node_result['filelist']
659
      for file_name in file_list:
660
        if file_name not in remote_cksum:
661
          bad = True
662
          feedback_fn("  - ERROR: file '%s' missing" % file_name)
663
        elif remote_cksum[file_name] != local_cksum[file_name]:
664
          bad = True
665
          feedback_fn("  - ERROR: file '%s' has wrong checksum" % file_name)
666

    
667
    if 'nodelist' not in node_result:
668
      bad = True
669
      feedback_fn("  - ERROR: node hasn't returned node connectivity data")
670
    else:
671
      if node_result['nodelist']:
672
        bad = True
673
        for node in node_result['nodelist']:
674
          feedback_fn("  - ERROR: communication with node '%s': %s" %
675
                          (node, node_result['nodelist'][node]))
676
    hyp_result = node_result.get('hypervisor', None)
677
    if hyp_result is not None:
678
      feedback_fn("  - ERROR: hypervisor verify failure: '%s'" % hyp_result)
679
    return bad
680

    
681
  def _VerifyInstance(self, instance, node_vol_is, node_instance, feedback_fn):
682
    """Verify an instance.
683

684
    This function checks to see if the required block devices are
685
    available on the instance's node.
686

687
    """
688
    bad = False
689

    
690
    instancelist = self.cfg.GetInstanceList()
691
    if not instance in instancelist:
692
      feedback_fn("  - ERROR: instance %s not in instance list %s" %
693
                      (instance, instancelist))
694
      bad = True
695

    
696
    instanceconfig = self.cfg.GetInstanceInfo(instance)
697
    node_current = instanceconfig.primary_node
698

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

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

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

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

    
722
    return not bad
723

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

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

730
    """
731
    bad = False
732

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

    
741

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

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

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

    
757
  def _VerifyNodeConfigFiles(self, ismaster, node, file_list, feedback_fn):
758
    """Verify the list of node config files"""
759

    
760
    bad = False
761
    for file_name in constants.MASTER_CONFIGFILES:
762
      if ismaster and file_name not in file_list:
763
        feedback_fn("  - ERROR: master config file %s missing from master"
764
                    " node %s" % (file_name, node))
765
        bad = True
766
      elif not ismaster and file_name in file_list:
767
        feedback_fn("  - ERROR: master config file %s should not exist"
768
                    " on non-master node %s" % (file_name, node))
769
        bad = True
770

    
771
    for file_name in constants.NODE_CONFIGFILES:
772
      if file_name not in file_list:
773
        feedback_fn("  - ERROR: config file %s missing from node %s" %
774
                    (file_name, node))
775
        bad = True
776

    
777
    return bad
778

    
779
  def CheckPrereq(self):
780
    """Check prerequisites.
781

782
    This has no prerequisites.
783

784
    """
785
    pass
786

    
787
  def Exec(self, feedback_fn):
788
    """Verify integrity of cluster, performing various test on nodes.
789

790
    """
791
    bad = False
792
    feedback_fn("* Verifying global settings")
793
    self.cfg.VerifyConfig()
794

    
795
    master = self.sstore.GetMasterNode()
796
    vg_name = self.cfg.GetVGName()
797
    nodelist = utils.NiceSort(self.cfg.GetNodeList())
798
    instancelist = utils.NiceSort(self.cfg.GetInstanceList())
799
    node_volume = {}
800
    node_instance = {}
801

    
802
    # FIXME: verify OS list
803
    # do local checksums
804
    file_names = constants.CLUSTER_CONF_FILES
805
    local_checksums = utils.FingerprintFiles(file_names)
806

    
807
    feedback_fn("* Gathering data (%d nodes)" % len(nodelist))
808
    all_configfile = rpc.call_configfile_list(nodelist)
809
    all_volumeinfo = rpc.call_volume_list(nodelist, vg_name)
810
    all_instanceinfo = rpc.call_instance_list(nodelist)
811
    all_vglist = rpc.call_vg_list(nodelist)
812
    node_verify_param = {
813
      'filelist': file_names,
814
      'nodelist': nodelist,
815
      'hypervisor': None,
816
      }
817
    all_nvinfo = rpc.call_node_verify(nodelist, node_verify_param)
818
    all_rversion = rpc.call_version(nodelist)
819

    
820
    for node in nodelist:
821
      feedback_fn("* Verifying node %s" % node)
822
      result = self._VerifyNode(node, file_names, local_checksums,
823
                                all_vglist[node], all_nvinfo[node],
824
                                all_rversion[node], feedback_fn)
825
      bad = bad or result
826
      # node_configfile
827
      nodeconfigfile = all_configfile[node]
828

    
829
      if not nodeconfigfile:
830
        feedback_fn("  - ERROR: connection to %s failed" % (node))
831
        bad = True
832
        continue
833

    
834
      bad = bad or self._VerifyNodeConfigFiles(node==master, node,
835
                                               nodeconfigfile, feedback_fn)
836

    
837
      # node_volume
838
      volumeinfo = all_volumeinfo[node]
839

    
840
      if type(volumeinfo) != dict:
841
        feedback_fn("  - ERROR: connection to %s failed" % (node,))
842
        bad = True
843
        continue
844

    
845
      node_volume[node] = volumeinfo
846

    
847
      # node_instance
848
      nodeinstance = all_instanceinfo[node]
849
      if type(nodeinstance) != list:
850
        feedback_fn("  - ERROR: connection to %s failed" % (node,))
851
        bad = True
852
        continue
853

    
854
      node_instance[node] = nodeinstance
855

    
856
    node_vol_should = {}
857

    
858
    for instance in instancelist:
859
      feedback_fn("* Verifying instance %s" % instance)
860
      result =  self._VerifyInstance(instance, node_volume, node_instance,
861
                                     feedback_fn)
862
      bad = bad or result
863

    
864
      inst_config = self.cfg.GetInstanceInfo(instance)
865

    
866
      inst_config.MapLVsByNode(node_vol_should)
867

    
868
    feedback_fn("* Verifying orphan volumes")
869
    result = self._VerifyOrphanVolumes(node_vol_should, node_volume,
870
                                       feedback_fn)
871
    bad = bad or result
872

    
873
    feedback_fn("* Verifying remaining instances")
874
    result = self._VerifyOrphanInstances(instancelist, node_instance,
875
                                         feedback_fn)
876
    bad = bad or result
877

    
878
    return int(bad)
879

    
880

    
881
def _WaitForSync(cfgw, instance, oneshot=False, unlock=False):
882
  """Sleep and poll for an instance's disk to sync.
883

884
  """
885
  if not instance.disks:
886
    return True
887

    
888
  if not oneshot:
889
    logger.ToStdout("Waiting for instance %s to sync disks." % instance.name)
890

    
891
  node = instance.primary_node
892

    
893
  for dev in instance.disks:
894
    cfgw.SetDiskID(dev, node)
895

    
896
  retries = 0
897
  while True:
898
    max_time = 0
899
    done = True
900
    cumul_degraded = False
901
    rstats = rpc.call_blockdev_getmirrorstatus(node, instance.disks)
902
    if not rstats:
903
      logger.ToStderr("Can't get any data from node %s" % node)
904
      retries += 1
905
      if retries >= 10:
906
        raise errors.RemoteError, ("Can't contact node %s for mirror data,"
907
                                   " aborting." % node)
908
      time.sleep(6)
909
      continue
910
    retries = 0
911
    for i in range(len(rstats)):
912
      mstat = rstats[i]
913
      if mstat is None:
914
        logger.ToStderr("Can't compute data for node %s/%s" %
915
                        (node, instance.disks[i].iv_name))
916
        continue
917
      perc_done, est_time, is_degraded = mstat
918
      cumul_degraded = cumul_degraded or (is_degraded and perc_done is None)
919
      if perc_done is not None:
920
        done = False
921
        if est_time is not None:
922
          rem_time = "%d estimated seconds remaining" % est_time
923
          max_time = est_time
924
        else:
925
          rem_time = "no time estimate"
926
        logger.ToStdout("- device %s: %5.2f%% done, %s" %
927
                        (instance.disks[i].iv_name, perc_done, rem_time))
928
    if done or oneshot:
929
      break
930

    
931
    if unlock:
932
      utils.Unlock('cmd')
933
    try:
934
      time.sleep(min(60, max_time))
935
    finally:
936
      if unlock:
937
        utils.Lock('cmd')
938

    
939
  if done:
940
    logger.ToStdout("Instance %s's disks are in sync." % instance.name)
941
  return not cumul_degraded
942

    
943

    
944
def _CheckDiskConsistency(cfgw, dev, node, on_primary):
945
  """Check that mirrors are not degraded.
946

947
  """
948

    
949
  cfgw.SetDiskID(dev, node)
950

    
951
  result = True
952
  if on_primary or dev.AssembleOnSecondary():
953
    rstats = rpc.call_blockdev_find(node, dev)
954
    if not rstats:
955
      logger.ToStderr("Can't get any data from node %s" % node)
956
      result = False
957
    else:
958
      result = result and (not rstats[5])
959
  if dev.children:
960
    for child in dev.children:
961
      result = result and _CheckDiskConsistency(cfgw, child, node, on_primary)
962

    
963
  return result
964

    
965

    
966
class LUDiagnoseOS(NoHooksLU):
967
  """Logical unit for OS diagnose/query.
968

969
  """
970
  _OP_REQP = []
971

    
972
  def CheckPrereq(self):
973
    """Check prerequisites.
974

975
    This always succeeds, since this is a pure query LU.
976

977
    """
978
    return
979

    
980
  def Exec(self, feedback_fn):
981
    """Compute the list of OSes.
982

983
    """
984
    node_list = self.cfg.GetNodeList()
985
    node_data = rpc.call_os_diagnose(node_list)
986
    if node_data == False:
987
      raise errors.OpExecError, "Can't gather the list of OSes"
988
    return node_data
989

    
990

    
991
class LURemoveNode(LogicalUnit):
992
  """Logical unit for removing a node.
993

994
  """
995
  HPATH = "node-remove"
996
  HTYPE = constants.HTYPE_NODE
997
  _OP_REQP = ["node_name"]
998

    
999
  def BuildHooksEnv(self):
1000
    """Build hooks env.
1001

1002
    This doesn't run on the target node in the pre phase as a failed
1003
    node would not allows itself to run.
1004

1005
    """
1006
    all_nodes = self.cfg.GetNodeList()
1007
    all_nodes.remove(self.op.node_name)
1008
    return {"NODE_NAME": self.op.node_name}, all_nodes, all_nodes
1009

    
1010
  def CheckPrereq(self):
1011
    """Check prerequisites.
1012

1013
    This checks:
1014
     - the node exists in the configuration
1015
     - it does not have primary or secondary instances
1016
     - it's not the master
1017

1018
    Any errors are signalled by raising errors.OpPrereqError.
1019

1020
    """
1021

    
1022
    node = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.node_name))
1023
    if node is None:
1024
      logger.Error("Error: Node '%s' is unknown." % self.op.node_name)
1025
      return 1
1026

    
1027
    instance_list = self.cfg.GetInstanceList()
1028

    
1029
    masternode = self.sstore.GetMasterNode()
1030
    if node.name == masternode:
1031
      raise errors.OpPrereqError, ("Node is the master node,"
1032
                                   " you need to failover first.")
1033

    
1034
    for instance_name in instance_list:
1035
      instance = self.cfg.GetInstanceInfo(instance_name)
1036
      if node.name == instance.primary_node:
1037
        raise errors.OpPrereqError, ("Instance %s still running on the node,"
1038
                                     " please remove first." % instance_name)
1039
      if node.name in instance.secondary_nodes:
1040
        raise errors.OpPrereqError, ("Instance %s has node as a secondary,"
1041
                                     " please remove first." % instance_name)
1042
    self.op.node_name = node.name
1043
    self.node = node
1044

    
1045
  def Exec(self, feedback_fn):
1046
    """Removes the node from the cluster.
1047

1048
    """
1049
    node = self.node
1050
    logger.Info("stopping the node daemon and removing configs from node %s" %
1051
                node.name)
1052

    
1053
    rpc.call_node_leave_cluster(node.name)
1054

    
1055
    ssh.SSHCall(node.name, 'root', "%s stop" % constants.NODE_INITD_SCRIPT)
1056

    
1057
    logger.Info("Removing node %s from config" % node.name)
1058

    
1059
    self.cfg.RemoveNode(node.name)
1060

    
1061

    
1062
class LUQueryNodes(NoHooksLU):
1063
  """Logical unit for querying nodes.
1064

1065
  """
1066
  _OP_REQP = ["output_fields"]
1067

    
1068
  def CheckPrereq(self):
1069
    """Check prerequisites.
1070

1071
    This checks that the fields required are valid output fields.
1072

1073
    """
1074
    self.dynamic_fields = frozenset(["dtotal", "dfree",
1075
                                     "mtotal", "mnode", "mfree"])
1076

    
1077
    _CheckOutputFields(static=["name", "pinst", "sinst", "pip", "sip"],
1078
                       dynamic=self.dynamic_fields,
1079
                       selected=self.op.output_fields)
1080

    
1081

    
1082
  def Exec(self, feedback_fn):
1083
    """Computes the list of nodes and their attributes.
1084

1085
    """
1086
    nodenames = utils.NiceSort(self.cfg.GetNodeList())
1087
    nodelist = [self.cfg.GetNodeInfo(name) for name in nodenames]
1088

    
1089

    
1090
    # begin data gathering
1091

    
1092
    if self.dynamic_fields.intersection(self.op.output_fields):
1093
      live_data = {}
1094
      node_data = rpc.call_node_info(nodenames, self.cfg.GetVGName())
1095
      for name in nodenames:
1096
        nodeinfo = node_data.get(name, None)
1097
        if nodeinfo:
1098
          live_data[name] = {
1099
            "mtotal": utils.TryConvert(int, nodeinfo['memory_total']),
1100
            "mnode": utils.TryConvert(int, nodeinfo['memory_dom0']),
1101
            "mfree": utils.TryConvert(int, nodeinfo['memory_free']),
1102
            "dtotal": utils.TryConvert(int, nodeinfo['vg_size']),
1103
            "dfree": utils.TryConvert(int, nodeinfo['vg_free']),
1104
            }
1105
        else:
1106
          live_data[name] = {}
1107
    else:
1108
      live_data = dict.fromkeys(nodenames, {})
1109

    
1110
    node_to_primary = dict.fromkeys(nodenames, 0)
1111
    node_to_secondary = dict.fromkeys(nodenames, 0)
1112

    
1113
    if "pinst" in self.op.output_fields or "sinst" in self.op.output_fields:
1114
      instancelist = self.cfg.GetInstanceList()
1115

    
1116
      for instance in instancelist:
1117
        instanceinfo = self.cfg.GetInstanceInfo(instance)
1118
        node_to_primary[instanceinfo.primary_node] += 1
1119
        for secnode in instanceinfo.secondary_nodes:
1120
          node_to_secondary[secnode] += 1
1121

    
1122
    # end data gathering
1123

    
1124
    output = []
1125
    for node in nodelist:
1126
      node_output = []
1127
      for field in self.op.output_fields:
1128
        if field == "name":
1129
          val = node.name
1130
        elif field == "pinst":
1131
          val = node_to_primary[node.name]
1132
        elif field == "sinst":
1133
          val = node_to_secondary[node.name]
1134
        elif field == "pip":
1135
          val = node.primary_ip
1136
        elif field == "sip":
1137
          val = node.secondary_ip
1138
        elif field in self.dynamic_fields:
1139
          val = live_data[node.name].get(field, "?")
1140
        else:
1141
          raise errors.ParameterError, field
1142
        val = str(val)
1143
        node_output.append(val)
1144
      output.append(node_output)
1145

    
1146
    return output
1147

    
1148

    
1149
class LUQueryNodeVolumes(NoHooksLU):
1150
  """Logical unit for getting volumes on node(s).
1151

1152
  """
1153
  _OP_REQP = ["nodes", "output_fields"]
1154

    
1155
  def CheckPrereq(self):
1156
    """Check prerequisites.
1157

1158
    This checks that the fields required are valid output fields.
1159

1160
    """
1161
    self.nodes = _GetWantedNodes(self, self.op.nodes)
1162

    
1163
    _CheckOutputFields(static=["node"],
1164
                       dynamic=["phys", "vg", "name", "size", "instance"],
1165
                       selected=self.op.output_fields)
1166

    
1167

    
1168
  def Exec(self, feedback_fn):
1169
    """Computes the list of nodes and their attributes.
1170

1171
    """
1172
    nodenames = utils.NiceSort([node.name for node in self.nodes])
1173
    volumes = rpc.call_node_volumes(nodenames)
1174

    
1175
    ilist = [self.cfg.GetInstanceInfo(iname) for iname
1176
             in self.cfg.GetInstanceList()]
1177

    
1178
    lv_by_node = dict([(inst, inst.MapLVsByNode()) for inst in ilist])
1179

    
1180
    output = []
1181
    for node in nodenames:
1182
      node_vols = volumes[node][:]
1183
      node_vols.sort(key=lambda vol: vol['dev'])
1184

    
1185
      for vol in node_vols:
1186
        node_output = []
1187
        for field in self.op.output_fields:
1188
          if field == "node":
1189
            val = node
1190
          elif field == "phys":
1191
            val = vol['dev']
1192
          elif field == "vg":
1193
            val = vol['vg']
1194
          elif field == "name":
1195
            val = vol['name']
1196
          elif field == "size":
1197
            val = int(float(vol['size']))
1198
          elif field == "instance":
1199
            for inst in ilist:
1200
              if node not in lv_by_node[inst]:
1201
                continue
1202
              if vol['name'] in lv_by_node[inst][node]:
1203
                val = inst.name
1204
                break
1205
            else:
1206
              val = '-'
1207
          else:
1208
            raise errors.ParameterError, field
1209
          node_output.append(str(val))
1210

    
1211
        output.append(node_output)
1212

    
1213
    return output
1214

    
1215

    
1216
def _CheckNodesDirs(node_list, paths):
1217
  """Verify if the given nodes have the same files.
1218

1219
  Args:
1220
    node_list: the list of node names to check
1221
    paths: the list of directories to checksum and compare
1222

1223
  Returns:
1224
    list of (node, different_file, message); if empty, the files are in sync
1225

1226
  """
1227
  file_names = []
1228
  for dir_name in paths:
1229
    flist = [os.path.join(dir_name, name) for name in os.listdir(dir_name)]
1230
    flist = [name for name in flist if os.path.isfile(name)]
1231
    file_names.extend(flist)
1232

    
1233
  local_checksums = utils.FingerprintFiles(file_names)
1234

    
1235
  results = []
1236
  verify_params = {'filelist': file_names}
1237
  all_node_results = rpc.call_node_verify(node_list, verify_params)
1238
  for node_name in node_list:
1239
    node_result = all_node_results.get(node_name, False)
1240
    if not node_result or 'filelist' not in node_result:
1241
      results.append((node_name, "'all files'", "node communication error"))
1242
      continue
1243
    remote_checksums = node_result['filelist']
1244
    for fname in local_checksums:
1245
      if fname not in remote_checksums:
1246
        results.append((node_name, fname, "missing file"))
1247
      elif remote_checksums[fname] != local_checksums[fname]:
1248
        results.append((node_name, fname, "wrong checksum"))
1249
  return results
1250

    
1251

    
1252
class LUAddNode(LogicalUnit):
1253
  """Logical unit for adding node to the cluster.
1254

1255
  """
1256
  HPATH = "node-add"
1257
  HTYPE = constants.HTYPE_NODE
1258
  _OP_REQP = ["node_name"]
1259

    
1260
  def BuildHooksEnv(self):
1261
    """Build hooks env.
1262

1263
    This will run on all nodes before, and on all nodes + the new node after.
1264

1265
    """
1266
    env = {
1267
      "NODE_NAME": self.op.node_name,
1268
      "NODE_PIP": self.op.primary_ip,
1269
      "NODE_SIP": self.op.secondary_ip,
1270
      }
1271
    nodes_0 = self.cfg.GetNodeList()
1272
    nodes_1 = nodes_0 + [self.op.node_name, ]
1273
    return env, nodes_0, nodes_1
1274

    
1275
  def CheckPrereq(self):
1276
    """Check prerequisites.
1277

1278
    This checks:
1279
     - the new node is not already in the config
1280
     - it is resolvable
1281
     - its parameters (single/dual homed) matches the cluster
1282

1283
    Any errors are signalled by raising errors.OpPrereqError.
1284

1285
    """
1286
    node_name = self.op.node_name
1287
    cfg = self.cfg
1288

    
1289
    dns_data = utils.LookupHostname(node_name)
1290
    if not dns_data:
1291
      raise errors.OpPrereqError, ("Node %s is not resolvable" % node_name)
1292

    
1293
    node = dns_data['hostname']
1294
    primary_ip = self.op.primary_ip = dns_data['ip']
1295
    secondary_ip = getattr(self.op, "secondary_ip", None)
1296
    if secondary_ip is None:
1297
      secondary_ip = primary_ip
1298
    if not utils.IsValidIP(secondary_ip):
1299
      raise errors.OpPrereqError, ("Invalid secondary IP given")
1300
    self.op.secondary_ip = secondary_ip
1301
    node_list = cfg.GetNodeList()
1302
    if node in node_list:
1303
      raise errors.OpPrereqError, ("Node %s is already in the configuration"
1304
                                   % node)
1305

    
1306
    for existing_node_name in node_list:
1307
      existing_node = cfg.GetNodeInfo(existing_node_name)
1308
      if (existing_node.primary_ip == primary_ip or
1309
          existing_node.secondary_ip == primary_ip or
1310
          existing_node.primary_ip == secondary_ip or
1311
          existing_node.secondary_ip == secondary_ip):
1312
        raise errors.OpPrereqError, ("New node ip address(es) conflict with"
1313
                                     " existing node %s" % existing_node.name)
1314

    
1315
    # check that the type of the node (single versus dual homed) is the
1316
    # same as for the master
1317
    myself = cfg.GetNodeInfo(self.sstore.GetMasterNode())
1318
    master_singlehomed = myself.secondary_ip == myself.primary_ip
1319
    newbie_singlehomed = secondary_ip == primary_ip
1320
    if master_singlehomed != newbie_singlehomed:
1321
      if master_singlehomed:
1322
        raise errors.OpPrereqError, ("The master has no private ip but the"
1323
                                     " new node has one")
1324
      else:
1325
        raise errors.OpPrereqError ("The master has a private ip but the"
1326
                                    " new node doesn't have one")
1327

    
1328
    # checks reachablity
1329
    command = ["fping", "-q", primary_ip]
1330
    result = utils.RunCmd(command)
1331
    if result.failed:
1332
      raise errors.OpPrereqError, ("Node not reachable by ping")
1333

    
1334
    if not newbie_singlehomed:
1335
      # check reachability from my secondary ip to newbie's secondary ip
1336
      command = ["fping", "-S%s" % myself.secondary_ip, "-q", secondary_ip]
1337
      result = utils.RunCmd(command)
1338
      if result.failed:
1339
        raise errors.OpPrereqError, ("Node secondary ip not reachable by ping")
1340

    
1341
    self.new_node = objects.Node(name=node,
1342
                                 primary_ip=primary_ip,
1343
                                 secondary_ip=secondary_ip)
1344

    
1345
  def Exec(self, feedback_fn):
1346
    """Adds the new node to the cluster.
1347

1348
    """
1349
    new_node = self.new_node
1350
    node = new_node.name
1351

    
1352
    # set up inter-node password and certificate and restarts the node daemon
1353
    gntpass = self.sstore.GetNodeDaemonPassword()
1354
    if not re.match('^[a-zA-Z0-9.]{1,64}$', gntpass):
1355
      raise errors.OpExecError, ("ganeti password corruption detected")
1356
    f = open(constants.SSL_CERT_FILE)
1357
    try:
1358
      gntpem = f.read(8192)
1359
    finally:
1360
      f.close()
1361
    # in the base64 pem encoding, neither '!' nor '.' are valid chars,
1362
    # so we use this to detect an invalid certificate; as long as the
1363
    # cert doesn't contain this, the here-document will be correctly
1364
    # parsed by the shell sequence below
1365
    if re.search('^!EOF\.', gntpem, re.MULTILINE):
1366
      raise errors.OpExecError, ("invalid PEM encoding in the SSL certificate")
1367
    if not gntpem.endswith("\n"):
1368
      raise errors.OpExecError, ("PEM must end with newline")
1369
    logger.Info("copy cluster pass to %s and starting the node daemon" % node)
1370

    
1371
    # remove first the root's known_hosts file
1372
    utils.RemoveFile("/root/.ssh/known_hosts")
1373
    # and then connect with ssh to set password and start ganeti-noded
1374
    # note that all the below variables are sanitized at this point,
1375
    # either by being constants or by the checks above
1376
    ss = self.sstore
1377
    mycommand = ("umask 077 && "
1378
                 "echo '%s' > '%s' && "
1379
                 "cat > '%s' << '!EOF.' && \n"
1380
                 "%s!EOF.\n%s restart" %
1381
                 (gntpass, ss.KeyToFilename(ss.SS_NODED_PASS),
1382
                  constants.SSL_CERT_FILE, gntpem,
1383
                  constants.NODE_INITD_SCRIPT))
1384

    
1385
    result = ssh.SSHCall(node, 'root', mycommand, batch=False, ask_key=True)
1386
    if result.failed:
1387
      raise errors.OpExecError, ("Remote command on node %s, error: %s,"
1388
                                 " output: %s" %
1389
                                 (node, result.fail_reason, result.output))
1390

    
1391
    # check connectivity
1392
    time.sleep(4)
1393

    
1394
    result = rpc.call_version([node])[node]
1395
    if result:
1396
      if constants.PROTOCOL_VERSION == result:
1397
        logger.Info("communication to node %s fine, sw version %s match" %
1398
                    (node, result))
1399
      else:
1400
        raise errors.OpExecError, ("Version mismatch master version %s,"
1401
                                   " node version %s" %
1402
                                   (constants.PROTOCOL_VERSION, result))
1403
    else:
1404
      raise errors.OpExecError, ("Cannot get version from the new node")
1405

    
1406
    # setup ssh on node
1407
    logger.Info("copy ssh key to node %s" % node)
1408
    keyarray = []
1409
    keyfiles = ["/etc/ssh/ssh_host_dsa_key", "/etc/ssh/ssh_host_dsa_key.pub",
1410
                "/etc/ssh/ssh_host_rsa_key", "/etc/ssh/ssh_host_rsa_key.pub",
1411
                "/root/.ssh/id_dsa", "/root/.ssh/id_dsa.pub"]
1412

    
1413
    for i in keyfiles:
1414
      f = open(i, 'r')
1415
      try:
1416
        keyarray.append(f.read())
1417
      finally:
1418
        f.close()
1419

    
1420
    result = rpc.call_node_add(node, keyarray[0], keyarray[1], keyarray[2],
1421
                               keyarray[3], keyarray[4], keyarray[5])
1422

    
1423
    if not result:
1424
      raise errors.OpExecError, ("Cannot transfer ssh keys to the new node")
1425

    
1426
    # Add node to our /etc/hosts, and add key to known_hosts
1427
    _UpdateEtcHosts(new_node.name, new_node.primary_ip)
1428
    _UpdateKnownHosts(new_node.name, new_node.primary_ip,
1429
                      self.cfg.GetHostKey())
1430

    
1431
    if new_node.secondary_ip != new_node.primary_ip:
1432
      result = ssh.SSHCall(node, "root",
1433
                           "fping -S 127.0.0.1 -q %s" % new_node.secondary_ip)
1434
      if result.failed:
1435
        raise errors.OpExecError, ("Node claims it doesn't have the"
1436
                                   " secondary ip you gave (%s).\n"
1437
                                   "Please fix and re-run this command." %
1438
                                   new_node.secondary_ip)
1439

    
1440
    # Distribute updated /etc/hosts and known_hosts to all nodes,
1441
    # including the node just added
1442
    myself = self.cfg.GetNodeInfo(self.sstore.GetMasterNode())
1443
    dist_nodes = self.cfg.GetNodeList() + [node]
1444
    if myself.name in dist_nodes:
1445
      dist_nodes.remove(myself.name)
1446

    
1447
    logger.Debug("Copying hosts and known_hosts to all nodes")
1448
    for fname in ("/etc/hosts", "/etc/ssh/ssh_known_hosts"):
1449
      result = rpc.call_upload_file(dist_nodes, fname)
1450
      for to_node in dist_nodes:
1451
        if not result[to_node]:
1452
          logger.Error("copy of file %s to node %s failed" %
1453
                       (fname, to_node))
1454

    
1455
    to_copy = [constants.MASTER_CRON_FILE]
1456
    to_copy.extend(ss.GetFileList())
1457
    for fname in to_copy:
1458
      if not ssh.CopyFileToNode(node, fname):
1459
        logger.Error("could not copy file %s to node %s" % (fname, node))
1460

    
1461
    logger.Info("adding node %s to cluster.conf" % node)
1462
    self.cfg.AddNode(new_node)
1463

    
1464

    
1465
class LUMasterFailover(LogicalUnit):
1466
  """Failover the master node to the current node.
1467

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

1470
  """
1471
  HPATH = "master-failover"
1472
  HTYPE = constants.HTYPE_CLUSTER
1473
  REQ_MASTER = False
1474
  _OP_REQP = []
1475

    
1476
  def BuildHooksEnv(self):
1477
    """Build hooks env.
1478

1479
    This will run on the new master only in the pre phase, and on all
1480
    the nodes in the post phase.
1481

1482
    """
1483
    env = {
1484
      "NEW_MASTER": self.new_master,
1485
      "OLD_MASTER": self.old_master,
1486
      }
1487
    return env, [self.new_master], self.cfg.GetNodeList()
1488

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

1492
    This checks that we are not already the master.
1493

1494
    """
1495
    self.new_master = socket.gethostname()
1496

    
1497
    self.old_master = self.sstore.GetMasterNode()
1498

    
1499
    if self.old_master == self.new_master:
1500
      raise errors.OpPrereqError, ("This commands must be run on the node"
1501
                                   " where you want the new master to be.\n"
1502
                                   "%s is already the master" %
1503
                                   self.old_master)
1504

    
1505
  def Exec(self, feedback_fn):
1506
    """Failover the master node.
1507

1508
    This command, when run on a non-master node, will cause the current
1509
    master to cease being master, and the non-master to become new
1510
    master.
1511

1512
    """
1513

    
1514
    #TODO: do not rely on gethostname returning the FQDN
1515
    logger.Info("setting master to %s, old master: %s" %
1516
                (self.new_master, self.old_master))
1517

    
1518
    if not rpc.call_node_stop_master(self.old_master):
1519
      logger.Error("could disable the master role on the old master"
1520
                   " %s, please disable manually" % self.old_master)
1521

    
1522
    ss = self.sstore
1523
    ss.SetKey(ss.SS_MASTER_NODE, self.new_master)
1524
    if not rpc.call_upload_file(self.cfg.GetNodeList(),
1525
                                ss.KeyToFilename(ss.SS_MASTER_NODE)):
1526
      logger.Error("could not distribute the new simple store master file"
1527
                   " to the other nodes, please check.")
1528

    
1529
    if not rpc.call_node_start_master(self.new_master):
1530
      logger.Error("could not start the master role on the new master"
1531
                   " %s, please check" % self.new_master)
1532
      feedback_fn("Error in activating the master IP on the new master,\n"
1533
                  "please fix manually.")
1534

    
1535

    
1536

    
1537
class LUQueryClusterInfo(NoHooksLU):
1538
  """Query cluster configuration.
1539

1540
  """
1541
  _OP_REQP = []
1542

    
1543
  def CheckPrereq(self):
1544
    """No prerequsites needed for this LU.
1545

1546
    """
1547
    pass
1548

    
1549
  def Exec(self, feedback_fn):
1550
    """Return cluster config.
1551

1552
    """
1553
    instances = [self.cfg.GetInstanceInfo(name)
1554
                 for name in self.cfg.GetInstanceList()]
1555
    result = {
1556
      "name": self.cfg.GetClusterName(),
1557
      "software_version": constants.RELEASE_VERSION,
1558
      "protocol_version": constants.PROTOCOL_VERSION,
1559
      "config_version": constants.CONFIG_VERSION,
1560
      "os_api_version": constants.OS_API_VERSION,
1561
      "export_version": constants.EXPORT_VERSION,
1562
      "master": self.sstore.GetMasterNode(),
1563
      "architecture": (platform.architecture()[0], platform.machine()),
1564
      "instances": [(instance.name, instance.primary_node)
1565
                    for instance in instances],
1566
      "nodes": self.cfg.GetNodeList(),
1567
      }
1568

    
1569
    return result
1570

    
1571

    
1572
class LUClusterCopyFile(NoHooksLU):
1573
  """Copy file to cluster.
1574

1575
  """
1576
  _OP_REQP = ["nodes", "filename"]
1577

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

1581
    It should check that the named file exists and that the given list
1582
    of nodes is valid.
1583

1584
    """
1585
    if not os.path.exists(self.op.filename):
1586
      raise errors.OpPrereqError("No such filename '%s'" % self.op.filename)
1587

    
1588
    self.nodes = _GetWantedNodes(self, self.op.nodes)
1589

    
1590
  def Exec(self, feedback_fn):
1591
    """Copy a file from master to some nodes.
1592

1593
    Args:
1594
      opts - class with options as members
1595
      args - list containing a single element, the file name
1596
    Opts used:
1597
      nodes - list containing the name of target nodes; if empty, all nodes
1598

1599
    """
1600
    filename = self.op.filename
1601

    
1602
    myname = socket.gethostname()
1603

    
1604
    for node in self.nodes:
1605
      if node == myname:
1606
        continue
1607
      if not ssh.CopyFileToNode(node, filename):
1608
        logger.Error("Copy of file %s to node %s failed" % (filename, node))
1609

    
1610

    
1611
class LUDumpClusterConfig(NoHooksLU):
1612
  """Return a text-representation of the cluster-config.
1613

1614
  """
1615
  _OP_REQP = []
1616

    
1617
  def CheckPrereq(self):
1618
    """No prerequisites.
1619

1620
    """
1621
    pass
1622

    
1623
  def Exec(self, feedback_fn):
1624
    """Dump a representation of the cluster config to the standard output.
1625

1626
    """
1627
    return self.cfg.DumpConfig()
1628

    
1629

    
1630
class LURunClusterCommand(NoHooksLU):
1631
  """Run a command on some nodes.
1632

1633
  """
1634
  _OP_REQP = ["command", "nodes"]
1635

    
1636
  def CheckPrereq(self):
1637
    """Check prerequisites.
1638

1639
    It checks that the given list of nodes is valid.
1640

1641
    """
1642
    self.nodes = _GetWantedNodes(self, self.op.nodes)
1643

    
1644
  def Exec(self, feedback_fn):
1645
    """Run a command on some nodes.
1646

1647
    """
1648
    data = []
1649
    for node in self.nodes:
1650
      result = utils.RunCmd(["ssh", node.name, self.op.command])
1651
      data.append((node.name, result.cmd, result.output, result.exit_code))
1652

    
1653
    return data
1654

    
1655

    
1656
class LUActivateInstanceDisks(NoHooksLU):
1657
  """Bring up an instance's disks.
1658

1659
  """
1660
  _OP_REQP = ["instance_name"]
1661

    
1662
  def CheckPrereq(self):
1663
    """Check prerequisites.
1664

1665
    This checks that the instance is in the cluster.
1666

1667
    """
1668
    instance = self.cfg.GetInstanceInfo(
1669
      self.cfg.ExpandInstanceName(self.op.instance_name))
1670
    if instance is None:
1671
      raise errors.OpPrereqError, ("Instance '%s' not known" %
1672
                                   self.op.instance_name)
1673
    self.instance = instance
1674

    
1675

    
1676
  def Exec(self, feedback_fn):
1677
    """Activate the disks.
1678

1679
    """
1680
    disks_ok, disks_info = _AssembleInstanceDisks(self.instance, self.cfg)
1681
    if not disks_ok:
1682
      raise errors.OpExecError, ("Cannot activate block devices")
1683

    
1684
    return disks_info
1685

    
1686

    
1687
def _AssembleInstanceDisks(instance, cfg, ignore_secondaries=False):
1688
  """Prepare the block devices for an instance.
1689

1690
  This sets up the block devices on all nodes.
1691

1692
  Args:
1693
    instance: a ganeti.objects.Instance object
1694
    ignore_secondaries: if true, errors on secondary nodes won't result
1695
                        in an error return from the function
1696

1697
  Returns:
1698
    false if the operation failed
1699
    list of (host, instance_visible_name, node_visible_name) if the operation
1700
         suceeded with the mapping from node devices to instance devices
1701
  """
1702
  device_info = []
1703
  disks_ok = True
1704
  for inst_disk in instance.disks:
1705
    master_result = None
1706
    for node, node_disk in inst_disk.ComputeNodeTree(instance.primary_node):
1707
      cfg.SetDiskID(node_disk, node)
1708
      is_primary = node == instance.primary_node
1709
      result = rpc.call_blockdev_assemble(node, node_disk, is_primary)
1710
      if not result:
1711
        logger.Error("could not prepare block device %s on node %s (is_pri"
1712
                     "mary=%s)" % (inst_disk.iv_name, node, is_primary))
1713
        if is_primary or not ignore_secondaries:
1714
          disks_ok = False
1715
      if is_primary:
1716
        master_result = result
1717
    device_info.append((instance.primary_node, inst_disk.iv_name,
1718
                        master_result))
1719

    
1720
  return disks_ok, device_info
1721

    
1722

    
1723
class LUDeactivateInstanceDisks(NoHooksLU):
1724
  """Shutdown an instance's disks.
1725

1726
  """
1727
  _OP_REQP = ["instance_name"]
1728

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

1732
    This checks that the instance is in the cluster.
1733

1734
    """
1735
    instance = self.cfg.GetInstanceInfo(
1736
      self.cfg.ExpandInstanceName(self.op.instance_name))
1737
    if instance is None:
1738
      raise errors.OpPrereqError, ("Instance '%s' not known" %
1739
                                   self.op.instance_name)
1740
    self.instance = instance
1741

    
1742
  def Exec(self, feedback_fn):
1743
    """Deactivate the disks
1744

1745
    """
1746
    instance = self.instance
1747
    ins_l = rpc.call_instance_list([instance.primary_node])
1748
    ins_l = ins_l[instance.primary_node]
1749
    if not type(ins_l) is list:
1750
      raise errors.OpExecError, ("Can't contact node '%s'" %
1751
                                 instance.primary_node)
1752

    
1753
    if self.instance.name in ins_l:
1754
      raise errors.OpExecError, ("Instance is running, can't shutdown"
1755
                                 " block devices.")
1756

    
1757
    _ShutdownInstanceDisks(instance, self.cfg)
1758

    
1759

    
1760
def _ShutdownInstanceDisks(instance, cfg, ignore_primary=False):
1761
  """Shutdown block devices of an instance.
1762

1763
  This does the shutdown on all nodes of the instance.
1764

1765
  If the ignore_primary is false, errors on the primary node are
1766
  ignored.
1767

1768
  """
1769
  result = True
1770
  for disk in instance.disks:
1771
    for node, top_disk in disk.ComputeNodeTree(instance.primary_node):
1772
      cfg.SetDiskID(top_disk, node)
1773
      if not rpc.call_blockdev_shutdown(node, top_disk):
1774
        logger.Error("could not shutdown block device %s on node %s" %
1775
                     (disk.iv_name, node))
1776
        if not ignore_primary or node != instance.primary_node:
1777
          result = False
1778
  return result
1779

    
1780

    
1781
class LUStartupInstance(LogicalUnit):
1782
  """Starts an instance.
1783

1784
  """
1785
  HPATH = "instance-start"
1786
  HTYPE = constants.HTYPE_INSTANCE
1787
  _OP_REQP = ["instance_name", "force"]
1788

    
1789
  def BuildHooksEnv(self):
1790
    """Build hooks env.
1791

1792
    This runs on master, primary and secondary nodes of the instance.
1793

1794
    """
1795
    env = {
1796
      "INSTANCE_NAME": self.op.instance_name,
1797
      "INSTANCE_PRIMARY": self.instance.primary_node,
1798
      "INSTANCE_SECONDARIES": " ".join(self.instance.secondary_nodes),
1799
      "FORCE": self.op.force,
1800
      }
1801
    nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
1802
          list(self.instance.secondary_nodes))
1803
    return env, nl, nl
1804

    
1805
  def CheckPrereq(self):
1806
    """Check prerequisites.
1807

1808
    This checks that the instance is in the cluster.
1809

1810
    """
1811
    instance = self.cfg.GetInstanceInfo(
1812
      self.cfg.ExpandInstanceName(self.op.instance_name))
1813
    if instance is None:
1814
      raise errors.OpPrereqError, ("Instance '%s' not known" %
1815
                                   self.op.instance_name)
1816

    
1817
    # check bridges existance
1818
    brlist = [nic.bridge for nic in instance.nics]
1819
    if not rpc.call_bridges_exist(instance.primary_node, brlist):
1820
      raise errors.OpPrereqError, ("one or more target bridges %s does not"
1821
                                   " exist on destination node '%s'" %
1822
                                   (brlist, instance.primary_node))
1823

    
1824
    self.instance = instance
1825
    self.op.instance_name = instance.name
1826

    
1827
  def Exec(self, feedback_fn):
1828
    """Start the instance.
1829

1830
    """
1831
    instance = self.instance
1832
    force = self.op.force
1833
    extra_args = getattr(self.op, "extra_args", "")
1834

    
1835
    node_current = instance.primary_node
1836

    
1837
    nodeinfo = rpc.call_node_info([node_current], self.cfg.GetVGName())
1838
    if not nodeinfo:
1839
      raise errors.OpExecError, ("Could not contact node %s for infos" %
1840
                                 (node_current))
1841

    
1842
    freememory = nodeinfo[node_current]['memory_free']
1843
    memory = instance.memory
1844
    if memory > freememory:
1845
      raise errors.OpExecError, ("Not enough memory to start instance"
1846
                                 " %s on node %s"
1847
                                 " needed %s MiB, available %s MiB" %
1848
                                 (instance.name, node_current, memory,
1849
                                  freememory))
1850

    
1851
    disks_ok, dummy = _AssembleInstanceDisks(instance, self.cfg,
1852
                                             ignore_secondaries=force)
1853
    if not disks_ok:
1854
      _ShutdownInstanceDisks(instance, self.cfg)
1855
      if not force:
1856
        logger.Error("If the message above refers to a secondary node,"
1857
                     " you can retry the operation using '--force'.")
1858
      raise errors.OpExecError, ("Disk consistency error")
1859

    
1860
    if not rpc.call_instance_start(node_current, instance, extra_args):
1861
      _ShutdownInstanceDisks(instance, self.cfg)
1862
      raise errors.OpExecError, ("Could not start instance")
1863

    
1864
    self.cfg.MarkInstanceUp(instance.name)
1865

    
1866

    
1867
class LUShutdownInstance(LogicalUnit):
1868
  """Shutdown an instance.
1869

1870
  """
1871
  HPATH = "instance-stop"
1872
  HTYPE = constants.HTYPE_INSTANCE
1873
  _OP_REQP = ["instance_name"]
1874

    
1875
  def BuildHooksEnv(self):
1876
    """Build hooks env.
1877

1878
    This runs on master, primary and secondary nodes of the instance.
1879

1880
    """
1881
    env = {
1882
      "INSTANCE_NAME": self.op.instance_name,
1883
      "INSTANCE_PRIMARY": self.instance.primary_node,
1884
      "INSTANCE_SECONDARIES": " ".join(self.instance.secondary_nodes),
1885
      }
1886
    nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
1887
          list(self.instance.secondary_nodes))
1888
    return env, nl, nl
1889

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

1893
    This checks that the instance is in the cluster.
1894

1895
    """
1896
    instance = self.cfg.GetInstanceInfo(
1897
      self.cfg.ExpandInstanceName(self.op.instance_name))
1898
    if instance is None:
1899
      raise errors.OpPrereqError, ("Instance '%s' not known" %
1900
                                   self.op.instance_name)
1901
    self.instance = instance
1902

    
1903
  def Exec(self, feedback_fn):
1904
    """Shutdown the instance.
1905

1906
    """
1907
    instance = self.instance
1908
    node_current = instance.primary_node
1909
    if not rpc.call_instance_shutdown(node_current, instance):
1910
      logger.Error("could not shutdown instance")
1911

    
1912
    self.cfg.MarkInstanceDown(instance.name)
1913
    _ShutdownInstanceDisks(instance, self.cfg)
1914

    
1915

    
1916
class LURemoveInstance(LogicalUnit):
1917
  """Remove an instance.
1918

1919
  """
1920
  HPATH = "instance-remove"
1921
  HTYPE = constants.HTYPE_INSTANCE
1922
  _OP_REQP = ["instance_name"]
1923

    
1924
  def BuildHooksEnv(self):
1925
    """Build hooks env.
1926

1927
    This runs on master, primary and secondary nodes of the instance.
1928

1929
    """
1930
    env = {
1931
      "INSTANCE_NAME": self.op.instance_name,
1932
      "INSTANCE_PRIMARY": self.instance.primary_node,
1933
      "INSTANCE_SECONDARIES": " ".join(self.instance.secondary_nodes),
1934
      }
1935
    nl = ([self.sstore.GetMasterNode(), self.instance.primary_node] +
1936
          list(self.instance.secondary_nodes))
1937
    return env, nl, nl
1938

    
1939
  def CheckPrereq(self):
1940
    """Check prerequisites.
1941

1942
    This checks that the instance is in the cluster.
1943

1944
    """
1945
    instance = self.cfg.GetInstanceInfo(
1946
      self.cfg.ExpandInstanceName(self.op.instance_name))
1947
    if instance is None:
1948
      raise errors.OpPrereqError, ("Instance '%s' not known" %
1949
                                   self.op.instance_name)
1950
    self.instance = instance
1951

    
1952
  def Exec(self, feedback_fn):
1953
    """Remove the instance.
1954

1955
    """
1956
    instance = self.instance
1957
    logger.Info("shutting down instance %s on node %s" %
1958
                (instance.name, instance.primary_node))
1959

    
1960
    if not rpc.call_instance_shutdown(instance.primary_node, instance):
1961
      raise errors.OpExecError, ("Could not shutdown instance %s on node %s" %
1962
                                 (instance.name, instance.primary_node))
1963

    
1964
    logger.Info("removing block devices for instance %s" % instance.name)
1965

    
1966
    _RemoveDisks(instance, self.cfg)
1967

    
1968
    logger.Info("removing instance %s out of cluster config" % instance.name)
1969

    
1970
    self.cfg.RemoveInstance(instance.name)
1971

    
1972

    
1973
class LUQueryInstances(NoHooksLU):
1974
  """Logical unit for querying instances.
1975

1976
  """
1977
  _OP_REQP = ["output_fields"]
1978

    
1979
  def CheckPrereq(self):
1980
    """Check prerequisites.
1981

1982
    This checks that the fields required are valid output fields.
1983

1984
    """
1985
    self.dynamic_fields = frozenset(["oper_state", "oper_ram"])
1986
    _CheckOutputFields(static=["name", "os", "pnode", "snodes",
1987
                               "admin_state", "admin_ram",
1988
                               "disk_template", "ip", "mac", "bridge"],
1989
                       dynamic=self.dynamic_fields,
1990
                       selected=self.op.output_fields)
1991

    
1992
  def Exec(self, feedback_fn):
1993
    """Computes the list of nodes and their attributes.
1994

1995
    """
1996

    
1997
    instance_names = utils.NiceSort(self.cfg.GetInstanceList())
1998
    instance_list = [self.cfg.GetInstanceInfo(iname) for iname
1999
                     in instance_names]
2000

    
2001
    # begin data gathering
2002

    
2003
    nodes = frozenset([inst.primary_node for inst in instance_list])
2004

    
2005
    bad_nodes = []
2006
    if self.dynamic_fields.intersection(self.op.output_fields):
2007
      live_data = {}
2008
      node_data = rpc.call_all_instances_info(nodes)
2009
      for name in nodes:
2010
        result = node_data[name]
2011
        if result:
2012
          live_data.update(result)
2013
        elif result == False:
2014
          bad_nodes.append(name)
2015
        # else no instance is alive
2016
    else:
2017
      live_data = dict([(name, {}) for name in instance_names])
2018

    
2019
    # end data gathering
2020

    
2021
    output = []
2022
    for instance in instance_list:
2023
      iout = []
2024
      for field in self.op.output_fields:
2025
        if field == "name":
2026
          val = instance.name
2027
        elif field == "os":
2028
          val = instance.os
2029
        elif field == "pnode":
2030
          val = instance.primary_node
2031
        elif field == "snodes":
2032
          val = ",".join(instance.secondary_nodes) or "-"
2033
        elif field == "admin_state":
2034
          if instance.status == "down":
2035
            val = "no"
2036
          else:
2037
            val = "yes"
2038
        elif field == "oper_state":
2039
          if instance.primary_node in bad_nodes:
2040
            val = "(node down)"
2041
          else:
2042
            if live_data.get(instance.name):
2043
              val = "running"
2044
            else:
2045
              val = "stopped"
2046
        elif field == "admin_ram":
2047
          val = instance.memory
2048
        elif field == "oper_ram":
2049
          if instance.primary_node in bad_nodes:
2050
            val = "(node down)"
2051
          elif instance.name in live_data:
2052
            val = live_data[instance.name].get("memory", "?")
2053
          else:
2054
            val = "-"
2055
        elif field == "disk_template":
2056
          val = instance.disk_template
2057
        elif field == "ip":
2058
          val = instance.nics[0].ip
2059
        elif field == "bridge":
2060
          val = instance.nics[0].bridge
2061
        elif field == "mac":
2062
          val = instance.nics[0].mac
2063
        else:
2064
          raise errors.ParameterError, field
2065
        val = str(val)
2066
        iout.append(val)
2067
      output.append(iout)
2068

    
2069
    return output
2070

    
2071

    
2072
class LUFailoverInstance(LogicalUnit):
2073
  """Failover an instance.
2074

2075
  """
2076
  HPATH = "instance-failover"
2077
  HTYPE = constants.HTYPE_INSTANCE
2078
  _OP_REQP = ["instance_name", "ignore_consistency"]
2079

    
2080
  def BuildHooksEnv(self):
2081
    """Build hooks env.
2082

2083
    This runs on master, primary and secondary nodes of the instance.
2084

2085
    """
2086
    env = {
2087
      "INSTANCE_NAME": self.op.instance_name,
2088
      "INSTANCE_PRIMARY": self.instance.primary_node,
2089
      "INSTANCE_SECONDARIES": " ".join(self.instance.secondary_nodes),
2090
      "IGNORE_CONSISTENCY": self.op.ignore_consistency,
2091
      }
2092
    nl = [self.sstore.GetMasterNode()] + list(self.instance.secondary_nodes)
2093
    return env, nl, nl
2094

    
2095
  def CheckPrereq(self):
2096
    """Check prerequisites.
2097

2098
    This checks that the instance is in the cluster.
2099

2100
    """
2101
    instance = self.cfg.GetInstanceInfo(
2102
      self.cfg.ExpandInstanceName(self.op.instance_name))
2103
    if instance is None:
2104
      raise errors.OpPrereqError, ("Instance '%s' not known" %
2105
                                   self.op.instance_name)
2106

    
2107
    # check memory requirements on the secondary node
2108
    target_node = instance.secondary_nodes[0]
2109
    nodeinfo = rpc.call_node_info([target_node], self.cfg.GetVGName())
2110
    info = nodeinfo.get(target_node, None)
2111
    if not info:
2112
      raise errors.OpPrereqError, ("Cannot get current information"
2113
                                   " from node '%s'" % nodeinfo)
2114
    if instance.memory > info['memory_free']:
2115
      raise errors.OpPrereqError, ("Not enough memory on target node %s."
2116
                                   " %d MB available, %d MB required" %
2117
                                   (target_node, info['memory_free'],
2118
                                    instance.memory))
2119

    
2120
    # check bridge existance
2121
    brlist = [nic.bridge for nic in instance.nics]
2122
    if not rpc.call_bridges_exist(instance.primary_node, brlist):
2123
      raise errors.OpPrereqError, ("one or more target bridges %s does not"
2124
                                   " exist on destination node '%s'" %
2125
                                   (brlist, instance.primary_node))
2126

    
2127
    self.instance = instance
2128

    
2129
  def Exec(self, feedback_fn):
2130
    """Failover an instance.
2131

2132
    The failover is done by shutting it down on its present node and
2133
    starting it on the secondary.
2134

2135
    """
2136
    instance = self.instance
2137

    
2138
    source_node = instance.primary_node
2139
    target_node = instance.secondary_nodes[0]
2140

    
2141
    feedback_fn("* checking disk consistency between source and target")
2142
    for dev in instance.disks:
2143
      # for remote_raid1, these are md over drbd
2144
      if not _CheckDiskConsistency(self.cfg, dev, target_node, False):
2145
        if not self.op.ignore_consistency:
2146
          raise errors.OpExecError, ("Disk %s is degraded on target node,"
2147
                                     " aborting failover." % dev.iv_name)
2148

    
2149
    feedback_fn("* checking target node resource availability")
2150
    nodeinfo = rpc.call_node_info([target_node], self.cfg.GetVGName())
2151

    
2152
    if not nodeinfo:
2153
      raise errors.OpExecError, ("Could not contact target node %s." %
2154
                                 target_node)
2155

    
2156
    free_memory = int(nodeinfo[target_node]['memory_free'])
2157
    memory = instance.memory
2158
    if memory > free_memory:
2159
      raise errors.OpExecError, ("Not enough memory to create instance %s on"
2160
                                 " node %s. needed %s MiB, available %s MiB" %
2161
                                 (instance.name, target_node, memory,
2162
                                  free_memory))
2163

    
2164
    feedback_fn("* shutting down instance on source node")
2165
    logger.Info("Shutting down instance %s on node %s" %
2166
                (instance.name, source_node))
2167

    
2168
    if not rpc.call_instance_shutdown(source_node, instance):
2169
      logger.Error("Could not shutdown instance %s on node %s. Proceeding"
2170
                   " anyway. Please make sure node %s is down"  %
2171
                   (instance.name, source_node, source_node))
2172

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

    
2177
    instance.primary_node = target_node
2178
    # distribute new instance config to the other nodes
2179
    self.cfg.AddInstance(instance)
2180

    
2181
    feedback_fn("* activating the instance's disks on target node")
2182
    logger.Info("Starting instance %s on node %s" %
2183
                (instance.name, target_node))
2184

    
2185
    disks_ok, dummy = _AssembleInstanceDisks(instance, self.cfg,
2186
                                             ignore_secondaries=True)
2187
    if not disks_ok:
2188
      _ShutdownInstanceDisks(instance, self.cfg)
2189
      raise errors.OpExecError, ("Can't activate the instance's disks")
2190

    
2191
    feedback_fn("* starting the instance on the target node")
2192
    if not rpc.call_instance_start(target_node, instance, None):
2193
      _ShutdownInstanceDisks(instance, self.cfg)
2194
      raise errors.OpExecError("Could not start instance %s on node %s." %
2195
                               (instance.name, target_node))
2196

    
2197

    
2198
def _CreateBlockDevOnPrimary(cfg, node, device):
2199
  """Create a tree of block devices on the primary node.
2200

2201
  This always creates all devices.
2202

2203
  """
2204

    
2205
  if device.children:
2206
    for child in device.children:
2207
      if not _CreateBlockDevOnPrimary(cfg, node, child):
2208
        return False
2209

    
2210
  cfg.SetDiskID(device, node)
2211
  new_id = rpc.call_blockdev_create(node, device, device.size, True)
2212
  if not new_id:
2213
    return False
2214
  if device.physical_id is None:
2215
    device.physical_id = new_id
2216
  return True
2217

    
2218

    
2219
def _CreateBlockDevOnSecondary(cfg, node, device, force):
2220
  """Create a tree of block devices on a secondary node.
2221

2222
  If this device type has to be created on secondaries, create it and
2223
  all its children.
2224

2225
  If not, just recurse to children keeping the same 'force' value.
2226

2227
  """
2228
  if device.CreateOnSecondary():
2229
    force = True
2230
  if device.children:
2231
    for child in device.children:
2232
      if not _CreateBlockDevOnSecondary(cfg, node, child, force):
2233
        return False
2234

    
2235
  if not force:
2236
    return True
2237
  cfg.SetDiskID(device, node)
2238
  new_id = rpc.call_blockdev_create(node, device, device.size, False)
2239
  if not new_id:
2240
    return False
2241
  if device.physical_id is None:
2242
    device.physical_id = new_id
2243
  return True
2244

    
2245

    
2246
def _GenerateMDDRBDBranch(cfg, vgname, primary, secondary, size, base):
2247
  """Generate a drbd device complete with its children.
2248

2249
  """
2250
  port = cfg.AllocatePort()
2251
  base = "%s_%s" % (base, port)
2252
  dev_data = objects.Disk(dev_type="lvm", size=size,
2253
                          logical_id=(vgname, "%s.data" % base))
2254
  dev_meta = objects.Disk(dev_type="lvm", size=128,
2255
                          logical_id=(vgname, "%s.meta" % base))
2256
  drbd_dev = objects.Disk(dev_type="drbd", size=size,
2257
                          logical_id = (primary, secondary, port),
2258
                          children = [dev_data, dev_meta])
2259
  return drbd_dev
2260

    
2261

    
2262
def _GenerateDiskTemplate(cfg, vgname, template_name,
2263
                          instance_name, primary_node,
2264
                          secondary_nodes, disk_sz, swap_sz):
2265
  """Generate the entire disk layout for a given template type.
2266

2267
  """
2268
  #TODO: compute space requirements
2269

    
2270
  if template_name == "diskless":
2271
    disks = []
2272
  elif template_name == "plain":
2273
    if len(secondary_nodes) != 0:
2274
      raise errors.ProgrammerError("Wrong template configuration")
2275
    sda_dev = objects.Disk(dev_type="lvm", size=disk_sz,
2276
                           logical_id=(vgname, "%s.os" % instance_name),
2277
                           iv_name = "sda")
2278
    sdb_dev = objects.Disk(dev_type="lvm", size=swap_sz,
2279
                           logical_id=(vgname, "%s.swap" % instance_name),
2280
                           iv_name = "sdb")
2281
    disks = [sda_dev, sdb_dev]
2282
  elif template_name == "local_raid1":
2283
    if len(secondary_nodes) != 0:
2284
      raise errors.ProgrammerError("Wrong template configuration")
2285
    sda_dev_m1 = objects.Disk(dev_type="lvm", size=disk_sz,
2286
                              logical_id=(vgname, "%s.os_m1" % instance_name))
2287
    sda_dev_m2 = objects.Disk(dev_type="lvm", size=disk_sz,
2288
                              logical_id=(vgname, "%s.os_m2" % instance_name))
2289
    md_sda_dev = objects.Disk(dev_type="md_raid1", iv_name = "sda",
2290
                              size=disk_sz,
2291
                              children = [sda_dev_m1, sda_dev_m2])
2292
    sdb_dev_m1 = objects.Disk(dev_type="lvm", size=swap_sz,
2293
                              logical_id=(vgname, "%s.swap_m1" %
2294
                                          instance_name))
2295
    sdb_dev_m2 = objects.Disk(dev_type="lvm", size=swap_sz,
2296
                              logical_id=(vgname, "%s.swap_m2" %
2297
                                          instance_name))
2298
    md_sdb_dev = objects.Disk(dev_type="md_raid1", iv_name = "sdb",
2299
                              size=swap_sz,
2300
                              children = [sdb_dev_m1, sdb_dev_m2])
2301
    disks = [md_sda_dev, md_sdb_dev]
2302
  elif template_name == "remote_raid1":
2303
    if len(secondary_nodes) != 1:
2304
      raise errors.ProgrammerError("Wrong template configuration")
2305
    remote_node = secondary_nodes[0]
2306
    drbd_sda_dev = _GenerateMDDRBDBranch(cfg, vgname,
2307
                                         primary_node, remote_node, disk_sz,
2308
                                         "%s-sda" % instance_name)
2309
    md_sda_dev = objects.Disk(dev_type="md_raid1", iv_name="sda",
2310
                              children = [drbd_sda_dev], size=disk_sz)
2311
    drbd_sdb_dev = _GenerateMDDRBDBranch(cfg, vgname,
2312
                                         primary_node, remote_node, swap_sz,
2313
                                         "%s-sdb" % instance_name)
2314
    md_sdb_dev = objects.Disk(dev_type="md_raid1", iv_name="sdb",
2315
                              children = [drbd_sdb_dev], size=swap_sz)
2316
    disks = [md_sda_dev, md_sdb_dev]
2317
  else:
2318
    raise errors.ProgrammerError("Invalid disk template '%s'" % template_name)
2319
  return disks
2320

    
2321

    
2322
def _CreateDisks(cfg, instance):
2323
  """Create all disks for an instance.
2324

2325
  This abstracts away some work from AddInstance.
2326

2327
  Args:
2328
    instance: the instance object
2329

2330
  Returns:
2331
    True or False showing the success of the creation process
2332

2333
  """
2334
  for device in instance.disks:
2335
    logger.Info("creating volume %s for instance %s" %
2336
              (device.iv_name, instance.name))
2337
    #HARDCODE
2338
    for secondary_node in instance.secondary_nodes:
2339
      if not _CreateBlockDevOnSecondary(cfg, secondary_node, device, False):
2340
        logger.Error("failed to create volume %s (%s) on secondary node %s!" %
2341
                     (device.iv_name, device, secondary_node))
2342
        return False
2343
    #HARDCODE
2344
    if not _CreateBlockDevOnPrimary(cfg, instance.primary_node, device):
2345
      logger.Error("failed to create volume %s on primary!" %
2346
                   device.iv_name)
2347
      return False
2348
  return True
2349

    
2350

    
2351
def _RemoveDisks(instance, cfg):
2352
  """Remove all disks for an instance.
2353

2354
  This abstracts away some work from `AddInstance()` and
2355
  `RemoveInstance()`. Note that in case some of the devices couldn't
2356
  be remove, the removal will continue with the other ones (compare
2357
  with `_CreateDisks()`).
2358

2359
  Args:
2360
    instance: the instance object
2361

2362
  Returns:
2363
    True or False showing the success of the removal proces
2364

2365
  """
2366
  logger.Info("removing block devices for instance %s" % instance.name)
2367

    
2368
  result = True
2369
  for device in instance.disks:
2370
    for node, disk in device.ComputeNodeTree(instance.primary_node):
2371
      cfg.SetDiskID(disk, node)
2372
      if not rpc.call_blockdev_remove(node, disk):
2373
        logger.Error("could not remove block device %s on node %s,"
2374
                     " continuing anyway" %
2375
                     (device.iv_name, node))
2376
        result = False
2377
  return result
2378

    
2379

    
2380
class LUCreateInstance(LogicalUnit):
2381
  """Create an instance.
2382

2383
  """
2384
  HPATH = "instance-add"
2385
  HTYPE = constants.HTYPE_INSTANCE
2386
  _OP_REQP = ["instance_name", "mem_size", "disk_size", "pnode",
2387
              "disk_template", "swap_size", "mode", "start", "vcpus",
2388
              "wait_for_sync"]
2389

    
2390
  def BuildHooksEnv(self):
2391
    """Build hooks env.
2392

2393
    This runs on master, primary and secondary nodes of the instance.
2394

2395
    """
2396
    env = {
2397
      "INSTANCE_NAME": self.op.instance_name,
2398
      "INSTANCE_PRIMARY": self.op.pnode,
2399
      "INSTANCE_SECONDARIES": " ".join(self.secondaries),
2400
      "DISK_TEMPLATE": self.op.disk_template,
2401
      "MEM_SIZE": self.op.mem_size,
2402
      "DISK_SIZE": self.op.disk_size,
2403
      "SWAP_SIZE": self.op.swap_size,
2404
      "VCPUS": self.op.vcpus,
2405
      "BRIDGE": self.op.bridge,
2406
      "INSTANCE_ADD_MODE": self.op.mode,
2407
      }
2408
    if self.op.mode == constants.INSTANCE_IMPORT:
2409
      env["SRC_NODE"] = self.op.src_node
2410
      env["SRC_PATH"] = self.op.src_path
2411
      env["SRC_IMAGE"] = self.src_image
2412
    if self.inst_ip:
2413
      env["INSTANCE_IP"] = self.inst_ip
2414

    
2415
    nl = ([self.sstore.GetMasterNode(), self.op.pnode] +
2416
          self.secondaries)
2417
    return env, nl, nl
2418

    
2419

    
2420
  def CheckPrereq(self):
2421
    """Check prerequisites.
2422

2423
    """
2424
    if self.op.mode not in (constants.INSTANCE_CREATE,
2425
                            constants.INSTANCE_IMPORT):
2426
      raise errors.OpPrereqError, ("Invalid instance creation mode '%s'" %
2427
                                   self.op.mode)
2428

    
2429
    if self.op.mode == constants.INSTANCE_IMPORT:
2430
      src_node = getattr(self.op, "src_node", None)
2431
      src_path = getattr(self.op, "src_path", None)
2432
      if src_node is None or src_path is None:
2433
        raise errors.OpPrereqError, ("Importing an instance requires source"
2434
                                     " node and path options")
2435
      src_node_full = self.cfg.ExpandNodeName(src_node)
2436
      if src_node_full is None:
2437
        raise errors.OpPrereqError, ("Unknown source node '%s'" % src_node)
2438
      self.op.src_node = src_node = src_node_full
2439

    
2440
      if not os.path.isabs(src_path):
2441
        raise errors.OpPrereqError, ("The source path must be absolute")
2442

    
2443
      export_info = rpc.call_export_info(src_node, src_path)
2444

    
2445
      if not export_info:
2446
        raise errors.OpPrereqError, ("No export found in dir %s" % src_path)
2447

    
2448
      if not export_info.has_section(constants.INISECT_EXP):
2449
        raise errors.ProgrammerError, ("Corrupted export config")
2450

    
2451
      ei_version = export_info.get(constants.INISECT_EXP, 'version')
2452
      if (int(ei_version) != constants.EXPORT_VERSION):
2453
        raise errors.OpPrereqError, ("Wrong export version %s (wanted %d)" %
2454
                                     (ei_version, constants.EXPORT_VERSION))
2455

    
2456
      if int(export_info.get(constants.INISECT_INS, 'disk_count')) > 1:
2457
        raise errors.OpPrereqError, ("Can't import instance with more than"
2458
                                     " one data disk")
2459

    
2460
      # FIXME: are the old os-es, disk sizes, etc. useful?
2461
      self.op.os_type = export_info.get(constants.INISECT_EXP, 'os')
2462
      diskimage = os.path.join(src_path, export_info.get(constants.INISECT_INS,
2463
                                                         'disk0_dump'))
2464
      self.src_image = diskimage
2465
    else: # INSTANCE_CREATE
2466
      if getattr(self.op, "os_type", None) is None:
2467
        raise errors.OpPrereqError, ("No guest OS specified")
2468

    
2469
    # check primary node
2470
    pnode = self.cfg.GetNodeInfo(self.cfg.ExpandNodeName(self.op.pnode))
2471
    if pnode is None:
2472
      raise errors.OpPrereqError, ("Primary node '%s' is uknown" %
2473
                                   self.op.pnode)
2474
    self.op.pnode = pnode.name
2475
    self.pnode = pnode
2476
    self.secondaries = []
2477
    # disk template and mirror node verification
2478
    if self.op.disk_template not in constants.DISK_TEMPLATES:
2479
      raise errors.OpPrereqError, ("Invalid disk template name")
2480

    
2481
    if self.op.disk_template == constants.DT_REMOTE_RAID1:
2482
      if getattr(self.op, "snode", None) is None:
2483
        raise errors.OpPrereqError, ("The 'remote_raid1' disk template needs"
2484
                                     " a mirror node")
2485

    
2486
      snode_name = self.cfg.ExpandNodeName(self.op.snode)
2487
      if snode_name is None:
2488
        raise errors.OpPrereqError, ("Unknown secondary node '%s'" %
2489
                                     self.op.snode)
2490
      elif snode_name == pnode.name:
2491
        raise errors.OpPrereqError, ("The secondary node cannot be"
2492
                                     " the primary node.")
2493
      self.secondaries.append(snode_name)
2494

    
2495
    # Check lv size requirements
2496
    nodenames = [pnode.name] + self.secondaries
2497
    nodeinfo = rpc.call_node_info(nodenames, self.cfg.GetVGName())
2498

    
2499
    # Required free disk space as a function of disk and swap space
2500
    req_size_dict = {
2501
      constants.DT_DISKLESS: 0,
2502
      constants.DT_PLAIN: self.op.disk_size + self.op.swap_size,
2503
      constants.DT_LOCAL_RAID1: (self.op.disk_size + self.op.swap_size) * 2,
2504
      # 256 MB are added for drbd metadata, 128MB for each drbd device
2505
      constants.DT_REMOTE_RAID1: self.op.disk_size + self.op.swap_size + 256,
2506
    }
2507

    
2508
    if self.op.disk_template not in req_size_dict:
2509
      raise errors.ProgrammerError, ("Disk template '%s' size requirement"
2510
                                     " is unknown" %  self.op.disk_template)
2511

    
2512
    req_size = req_size_dict[self.op.disk_template]
2513

    
2514
    for node in nodenames:
2515
      info = nodeinfo.get(node, None)
2516
      if not info:
2517
        raise errors.OpPrereqError, ("Cannot get current information"
2518
                                     " from node '%s'" % nodeinfo)
2519
      if req_size > info['vg_free']:
2520
        raise errors.OpPrereqError, ("Not enough disk space on target node %s."
2521
                                     " %d MB available, %d MB required" %
2522
                                     (node, info['vg_free'], req_size))
2523

    
2524
    # os verification
2525
    os_obj = rpc.call_os_get([pnode.name], self.op.os_type)[pnode.name]
2526
    if not isinstance(os_obj, objects.OS):
2527
      raise errors.OpPrereqError, ("OS '%s' not in supported os list for"
2528
                                   " primary node"  % self.op.os_type)
2529

    
2530
    # instance verification
2531
    hostname1 = utils.LookupHostname(self.op.instance_name)
2532
    if not hostname1:
2533
      raise errors.OpPrereqError, ("Instance name '%s' not found in dns" %
2534
                                   self.op.instance_name)
2535

    
2536
    self.op.instance_name = instance_name = hostname1['hostname']
2537
    instance_list = self.cfg.GetInstanceList()
2538
    if instance_name in instance_list:
2539
      raise errors.OpPrereqError, ("Instance '%s' is already in the cluster" %
2540
                                   instance_name)
2541

    
2542
    ip = getattr(self.op, "ip", None)
2543
    if ip is None or ip.lower() == "none":
2544
      inst_ip = None
2545
    elif ip.lower() == "auto":
2546
      inst_ip = hostname1['ip']
2547
    else:
2548
      if not utils.IsValidIP(ip):
2549
        raise errors.OpPrereqError, ("given IP address '%s' doesn't look"
2550
                                     " like a valid IP" % ip)
2551
      inst_ip = ip
2552
    self.inst_ip = inst_ip
2553

    
2554
    command = ["fping", "-q", hostname1['ip']]
2555
    result = utils.RunCmd(command)
2556
    if not result.failed:
2557
      raise errors.OpPrereqError, ("IP %s of instance %s already in use" %
2558
                                   (hostname1['ip'], instance_name))
2559

    
2560
    # bridge verification
2561
    bridge = getattr(self.op, "bridge", None)
2562
    if bridge is None:
2563
      self.op.bridge = self.cfg.GetDefBridge()
2564
    else:
2565
      self.op.bridge = bridge
2566

    
2567
    if not rpc.call_bridges_exist(self.pnode.name, [self.op.bridge]):
2568
      raise errors.OpPrereqError, ("target bridge '%s' does not exist on"
2569
                                   " destination node '%s'" %
2570
                                   (self.op.bridge, pnode.name))
2571

    
2572
    if self.op.start:
2573
      self.instance_status = 'up'
2574
    else:
2575
      self.instance_status = 'down'
2576

    
2577
  def Exec(self, feedback_fn):
2578
    """Create and add the instance to the cluster.
2579

2580
    """
2581
    instance = self.op.instance_name
2582
    pnode_name = self.pnode.name
2583

    
2584
    nic = objects.NIC(bridge=self.op.bridge, mac=self.cfg.GenerateMAC())
2585
    if self.inst_ip is not None:
2586
      nic.ip = self.inst_ip
2587

    
2588
    disks = _GenerateDiskTemplate(self.cfg, self.cfg.GetVGName(),
2589
                                  self.op.disk_template,
2590
                                  instance, pnode_name,
2591
                                  self.secondaries, self.op.disk_size,
2592
                                  self.op.swap_size)
2593

    
2594
    iobj = objects.Instance(name=instance, os=self.op.os_type,
2595
                            primary_node=pnode_name,
2596
                            memory=self.op.mem_size,
2597
                            vcpus=self.op.vcpus,
2598
                            nics=[nic], disks=disks,
2599
                            disk_template=self.op.disk_template,
2600
                            status=self.instance_status,
2601
                            )
2602

    
2603
    feedback_fn("* creating instance disks...")
2604
    if not _CreateDisks(self.cfg, iobj):
2605
      _RemoveDisks(iobj, self.cfg)
2606
      raise errors.OpExecError, ("Device creation failed, reverting...")
2607

    
2608
    feedback_fn("adding instance %s to cluster config" % instance)
2609

    
2610
    self.cfg.AddInstance(iobj)
2611

    
2612
    if self.op.wait_for_sync:
2613
      disk_abort = not _WaitForSync(self.cfg, iobj)
2614
    elif iobj.disk_template == "remote_raid1":
2615
      # make sure the disks are not degraded (still sync-ing is ok)
2616
      time.sleep(15)
2617
      feedback_fn("* checking mirrors status")
2618
      disk_abort = not _WaitForSync(self.cfg, iobj, oneshot=True)
2619
    else:
2620
      disk_abort = False
2621

    
2622
    if disk_abort:
2623
      _RemoveDisks(iobj, self.cfg)
2624
      self.cfg.RemoveInstance(iobj.name)
2625
      raise errors.OpExecError, ("There are some degraded disks for"
2626
                                      " this instance")
2627

    
2628
    feedback_fn("creating os for instance %s on node %s" %
2629
                (instance, pnode_name))
2630

    
2631
    if iobj.disk_template != constants.DT_DISKLESS:
2632
      if self.op.mode == constants.INSTANCE_CREATE:
2633
        feedback_fn("* running the instance OS create scripts...")
2634
        if not rpc.call_instance_os_add(pnode_name, iobj, "sda", "sdb"):
2635
          raise errors.OpExecError, ("could not add os for instance %s"
2636
                                          " on node %s" %
2637
                                          (instance, pnode_name))
2638

    
2639
      elif self.op.mode == constants.INSTANCE_IMPORT:
2640
        feedback_fn("* running the instance OS import scripts...")
2641
        src_node = self.op.src_node
2642
        src_image = self.src_image
2643
        if not rpc.call_instance_os_import(pnode_name, iobj, "sda", "sdb",
2644
                                                src_node, src_image):
2645
          raise errors.OpExecError, ("Could not import os for instance"
2646
                                          " %s on node %s" %
2647
                                          (instance, pnode_name))
2648
      else:
2649
        # also checked in the prereq part
2650
        raise errors.ProgrammerError, ("Unknown OS initialization mode '%s'"
2651
                                       % self.op.mode)
2652

    
2653
    if self.op.start:
2654
      logger.Info("starting instance %s on node %s" % (instance, pnode_name))
2655
      feedback_fn("* starting instance...")
2656
      if not rpc.call_instance_start(pnode_name, iobj, None):
2657
        raise errors.OpExecError, ("Could not start instance")
2658

    
2659

    
2660
class LUConnectConsole(NoHooksLU):
2661
  """Connect to an instance's console.
2662

2663
  This is somewhat special in that it returns the command line that
2664
  you need to run on the master node in order to connect to the
2665
  console.
2666

2667
  """
2668
  _OP_REQP = ["instance_name"]
2669

    
2670
  def CheckPrereq(self):
2671
    """Check prerequisites.
2672

2673
    This checks that the instance is in the cluster.
2674

2675
    """
2676
    instance = self.cfg.GetInstanceInfo(
2677
      self.cfg.ExpandInstanceName(self.op.instance_name))
2678
    if instance is None:
2679
      raise errors.OpPrereqError, ("Instance '%s' not known" %
2680
                                   self.op.instance_name)
2681
    self.instance = instance
2682

    
2683
  def Exec(self, feedback_fn):
2684
    """Connect to the console of an instance
2685

2686
    """
2687
    instance = self.instance
2688
    node = instance.primary_node
2689

    
2690
    node_insts = rpc.call_instance_list([node])[node]
2691
    if node_insts is False:
2692
      raise errors.OpExecError, ("Can't connect to node %s." % node)
2693

    
2694
    if instance.name not in node_insts:
2695
      raise errors.OpExecError, ("Instance %s is not running." % instance.name)
2696

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

    
2699
    hyper = hypervisor.GetHypervisor()
2700
    console_cmd = hyper.GetShellCommandForConsole(instance.name)
2701
    return node, console_cmd
2702

    
2703

    
2704
class LUAddMDDRBDComponent(LogicalUnit):
2705
  """Adda new mirror member to an instance's disk.
2706

2707
  """
2708
  HPATH = "mirror-add"
2709
  HTYPE = constants.HTYPE_INSTANCE
2710
  _OP_REQP = ["instance_name", "remote_node", "disk_name"]
2711

    
2712
  def BuildHooksEnv(self):
2713
    """Build hooks env.
2714

2715
    This runs on the master, the primary and all the secondaries.
2716

2717
    """
2718
    env = {
2719
      "INSTANCE_NAME": self.op.instance_name,
2720
      "NEW_SECONDARY": self.op.remote_node,
2721
      "DISK_NAME": self.op.disk_name,
2722
      }
2723
    nl = [self.sstore.GetMasterNode(), self.instance.primary_node,
2724
          self.op.remote_node,] + list(self.instance.secondary_nodes)
2725
    return env, nl, nl
2726

    
2727
  def CheckPrereq(self):
2728
    """Check prerequisites.
2729

2730
    This checks that the instance is in the cluster.
2731

2732
    """
2733
    instance = self.cfg.GetInstanceInfo(
2734
      self.cfg.ExpandInstanceName(self.op.instance_name))
2735
    if instance is None:
2736
      raise errors.OpPrereqError, ("Instance '%s' not known" %
2737
                                   self.op.instance_name)
2738
    self.instance = instance
2739

    
2740
    remote_node = self.cfg.ExpandNodeName(self.op.remote_node)
2741
    if remote_node is None:
2742
      raise errors.OpPrereqError, ("Node '%s' not known" % self.op.remote_node)
2743
    self.remote_node = remote_node
2744

    
2745
    if remote_node == instance.primary_node:
2746
      raise errors.OpPrereqError, ("The specified node is the primary node of"
2747
                                   " the instance.")
2748

    
2749
    if instance.disk_template != constants.DT_REMOTE_RAID1:
2750
      raise errors.OpPrereqError, ("Instance's disk layout is not"
2751
                                   " remote_raid1.")
2752
    for disk in instance.disks:
2753
      if disk.iv_name == self.op.disk_name:
2754
        break
2755
    else:
2756
      raise errors.OpPrereqError, ("Can't find this device ('%s') in the"
2757
                                   " instance." % self.op.disk_name)
2758
    if len(disk.children) > 1:
2759
      raise errors.OpPrereqError, ("The device already has two slave"
2760
                                   " devices.\n"
2761
                                   "This would create a 3-disk raid1"
2762
                                   " which we don't allow.")
2763
    self.disk = disk
2764

    
2765
  def Exec(self, feedback_fn):
2766
    """Add the mirror component
2767

2768
    """
2769
    disk = self.disk
2770
    instance = self.instance
2771

    
2772
    remote_node = self.remote_node
2773
    new_drbd = _GenerateMDDRBDBranch(self.cfg, self.cfg.GetVGName(),
2774
                                     instance.primary_node, remote_node,
2775
                                     disk.size, "%s-%s" %
2776
                                     (instance.name, self.op.disk_name))
2777

    
2778
    logger.Info("adding new mirror component on secondary")
2779
    #HARDCODE
2780
    if not _CreateBlockDevOnSecondary(self.cfg, remote_node, new_drbd, False):
2781
      raise errors.OpExecError, ("Failed to create new component on secondary"
2782
                                 " node %s" % remote_node)
2783

    
2784
    logger.Info("adding new mirror component on primary")
2785
    #HARDCODE
2786
    if not _CreateBlockDevOnPrimary(self.cfg, instance.primary_node, new_drbd):
2787
      # remove secondary dev
2788
      self.cfg.SetDiskID(new_drbd, remote_node)
2789
      rpc.call_blockdev_remove(remote_node, new_drbd)
2790
      raise errors.OpExecError, ("Failed to create volume on primary")
2791

    
2792
    # the device exists now
2793
    # call the primary node to add the mirror to md
2794
    logger.Info("adding new mirror component to md")
2795
    if not rpc.call_blockdev_addchild(instance.primary_node,
2796
                                           disk, new_drbd):
2797
      logger.Error("Can't add mirror compoment to md!")
2798
      self.cfg.SetDiskID(new_drbd, remote_node)
2799
      if not rpc.call_blockdev_remove(remote_node, new_drbd):
2800
        logger.Error("Can't rollback on secondary")
2801
      self.cfg.SetDiskID(new_drbd, instance.primary_node)
2802
      if not rpc.call_blockdev_remove(instance.primary_node, new_drbd):
2803
        logger.Error("Can't rollback on primary")
2804
      raise errors.OpExecError, "Can't add mirror component to md array"
2805

    
2806
    disk.children.append(new_drbd)
2807

    
2808
    self.cfg.AddInstance(instance)
2809

    
2810
    _WaitForSync(self.cfg, instance)
2811

    
2812
    return 0
2813

    
2814

    
2815
class LURemoveMDDRBDComponent(LogicalUnit):
2816
  """Remove a component from a remote_raid1 disk.
2817

2818
  """
2819
  HPATH = "mirror-remove"
2820
  HTYPE = constants.HTYPE_INSTANCE
2821
  _OP_REQP = ["instance_name", "disk_name", "disk_id"]
2822

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

2826
    This runs on the master, the primary and all the secondaries.
2827

2828
    """
2829
    env = {
2830
      "INSTANCE_NAME": self.op.instance_name,
2831
      "DISK_NAME": self.op.disk_name,
2832
      "DISK_ID": self.op.disk_id,
2833
      "OLD_SECONDARY": self.old_secondary,
2834
      }
2835
    nl = [self.sstore.GetMasterNode(),
2836
          self.instance.primary_node] + list(self.instance.secondary_nodes)
2837
    return env, nl, nl
2838

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

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

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

    
2852
    if instance.disk_template != constants.DT_REMOTE_RAID1:
2853
      raise errors.OpPrereqError, ("Instance's disk layout is not"
2854
                                   " remote_raid1.")
2855
    for disk in instance.disks:
2856
      if disk.iv_name == self.op.disk_name:
2857
        break
2858
    else:
2859
      raise errors.OpPrereqError, ("Can't find this device ('%s') in the"
2860
                                   " instance." % self.op.disk_name)
2861
    for child in disk.children:
2862
      if child.dev_type == "drbd" and child.logical_id[2] == self.op.disk_id:
2863
        break
2864
    else:
2865
      raise errors.OpPrereqError, ("Can't find the device with this port.")
2866

    
2867
    if len(disk.children) < 2:
2868
      raise errors.OpPrereqError, ("Cannot remove the last component from"
2869
                                   " a mirror.")
2870
    self.disk = disk
2871
    self.child = child
2872
    if self.child.logical_id[0] == instance.primary_node:
2873
      oid = 1
2874
    else:
2875
      oid = 0
2876
    self.old_secondary = self.child.logical_id[oid]
2877

    
2878
  def Exec(self, feedback_fn):
2879
    """Remove the mirror component
2880

2881
    """
2882
    instance = self.instance
2883
    disk = self.disk
2884
    child = self.child
2885
    logger.Info("remove mirror component")
2886
    self.cfg.SetDiskID(disk, instance.primary_node)
2887
    if not rpc.call_blockdev_removechild(instance.primary_node,
2888
                                              disk, child):
2889
      raise errors.OpExecError, ("Can't remove child from mirror.")
2890

    
2891
    for node in child.logical_id[:2]:
2892
      self.cfg.SetDiskID(child, node)
2893
      if not rpc.call_blockdev_remove(node, child):
2894
        logger.Error("Warning: failed to remove device from node %s,"
2895
                     " continuing operation." % node)
2896

    
2897
    disk.children.remove(child)
2898
    self.cfg.AddInstance(instance)
2899

    
2900

    
2901
class LUReplaceDisks(LogicalUnit):
2902
  """Replace the disks of an instance.
2903

2904
  """
2905
  HPATH = "mirrors-replace"
2906
  HTYPE = constants.HTYPE_INSTANCE
2907
  _OP_REQP = ["instance_name"]
2908

    
2909
  def BuildHooksEnv(self):
2910
    """Build hooks env.
2911

2912
    This runs on the master, the primary and all the secondaries.
2913

2914
    """
2915
    env = {
2916
      "INSTANCE_NAME": self.op.instance_name,
2917
      "NEW_SECONDARY": self.op.remote_node,
2918
      "OLD_SECONDARY": self.instance.secondary_nodes[0],
2919
      }
2920
    nl = [self.sstore.GetMasterNode(),
2921
          self.instance.primary_node] + list(self.instance.secondary_nodes)
2922
    return env, nl, nl
2923

    
2924
  def CheckPrereq(self):
2925
    """Check prerequisites.
2926

2927
    This checks that the instance is in the cluster.
2928

2929
    """
2930
    instance = self.cfg.GetInstanceInfo(
2931
      self.cfg.ExpandInstanceName(self.op.instance_name))
2932
    if instance is None:
2933
      raise errors.OpPrereqError, ("Instance '%s' not known" %
2934
                                   self.op.instance_name)
2935
    self.instance = instance
2936

    
2937
    if instance.disk_template != constants.DT_REMOTE_RAID1:
2938
      raise errors.OpPrereqError, ("Instance's disk layout is not"
2939
                                   " remote_raid1.")
2940

    
2941
    if len(instance.secondary_nodes) != 1:
2942
      raise errors.OpPrereqError, ("The instance has a strange layout,"
2943
                                   " expected one secondary but found %d" %
2944
                                   len(instance.secondary_nodes))
2945

    
2946
    remote_node = getattr(self.op, "remote_node", None)
2947
    if remote_node is None:
2948
      remote_node = instance.secondary_nodes[0]
2949
    else:
2950
      remote_node = self.cfg.ExpandNodeName(remote_node)
2951
      if remote_node is None:
2952
        raise errors.OpPrereqError, ("Node '%s' not known" %
2953
                                     self.op.remote_node)
2954
    if remote_node == instance.primary_node:
2955
      raise errors.OpPrereqError, ("The specified node is the primary node of"
2956
                                   " the instance.")
2957
    self.op.remote_node = remote_node
2958

    
2959
  def Exec(self, feedback_fn):
2960
    """Replace the disks of an instance.
2961

2962
    """
2963
    instance = self.instance
2964
    iv_names = {}
2965
    # start of work
2966
    remote_node = self.op.remote_node
2967
    cfg = self.cfg
2968
    vgname = cfg.GetVGName()
2969
    for dev in instance.disks:
2970
      size = dev.size
2971
      new_drbd = _GenerateMDDRBDBranch(cfg, vgname, instance.primary_node,
2972
                                       remote_node, size,
2973
                                       "%s-%s" % (instance.name, dev.iv_name))
2974
      iv_names[dev.iv_name] = (dev, dev.children[0], new_drbd)
2975
      logger.Info("adding new mirror component on secondary for %s" %
2976
                  dev.iv_name)
2977
      #HARDCODE
2978
      if not _CreateBlockDevOnSecondary(cfg, remote_node, new_drbd, False):
2979
        raise errors.OpExecError, ("Failed to create new component on"
2980
                                   " secondary node %s\n"
2981
                                   "Full abort, cleanup manually!" %
2982
                                   remote_node)
2983

    
2984
      logger.Info("adding new mirror component on primary")
2985
      #HARDCODE
2986
      if not _CreateBlockDevOnPrimary(cfg, instance.primary_node, new_drbd):
2987
        # remove secondary dev
2988
        cfg.SetDiskID(new_drbd, remote_node)
2989
        rpc.call_blockdev_remove(remote_node, new_drbd)
2990
        raise errors.OpExecError("Failed to create volume on primary!\n"
2991
                                 "Full abort, cleanup manually!!")
2992

    
2993
      # the device exists now
2994
      # call the primary node to add the mirror to md
2995
      logger.Info("adding new mirror component to md")
2996
      if not rpc.call_blockdev_addchild(instance.primary_node, dev,
2997
                                        new_drbd):
2998
        logger.Error("Can't add mirror compoment to md!")
2999
        cfg.SetDiskID(new_drbd, remote_node)
3000
        if not rpc.call_blockdev_remove(remote_node, new_drbd):
3001
          logger.Error("Can't rollback on secondary")
3002
        cfg.SetDiskID(new_drbd, instance.primary_node)
3003
        if not rpc.call_blockdev_remove(instance.primary_node, new_drbd):
3004
          logger.Error("Can't rollback on primary")
3005
        raise errors.OpExecError, ("Full abort, cleanup manually!!")
3006

    
3007
      dev.children.append(new_drbd)
3008
      cfg.AddInstance(instance)
3009

    
3010
    # this can fail as the old devices are degraded and _WaitForSync
3011
    # does a combined result over all disks, so we don't check its
3012
    # return value
3013
    _WaitForSync(cfg, instance, unlock=True)
3014

    
3015
    # so check manually all the devices
3016
    for name in iv_names:
3017
      dev, child, new_drbd = iv_names[name]
3018
      cfg.SetDiskID(dev, instance.primary_node)
3019
      is_degr = rpc.call_blockdev_find(instance.primary_node, dev)[5]
3020
      if is_degr:
3021
        raise errors.OpExecError, ("MD device %s is degraded!" % name)
3022
      cfg.SetDiskID(new_drbd, instance.primary_node)
3023
      is_degr = rpc.call_blockdev_find(instance.primary_node, new_drbd)[5]
3024
      if is_degr:
3025
        raise errors.OpExecError, ("New drbd device %s is degraded!" % name)
3026

    
3027
    for name in iv_names:
3028
      dev, child, new_drbd = iv_names[name]
3029
      logger.Info("remove mirror %s component" % name)
3030
      cfg.SetDiskID(dev, instance.primary_node)
3031
      if not rpc.call_blockdev_removechild(instance.primary_node,
3032
                                                dev, child):
3033
        logger.Error("Can't remove child from mirror, aborting"
3034
                     " *this device cleanup*.\nYou need to cleanup manually!!")
3035
        continue
3036

    
3037
      for node in child.logical_id[:2]:
3038
        logger.Info("remove child device on %s" % node)
3039
        cfg.SetDiskID(child, node)
3040
        if not rpc.call_blockdev_remove(node, child):
3041
          logger.Error("Warning: failed to remove device from node %s,"
3042
                       " continuing operation." % node)
3043

    
3044
      dev.children.remove(child)
3045

    
3046
      cfg.AddInstance(instance)
3047

    
3048

    
3049
class LUQueryInstanceData(NoHooksLU):
3050
  """Query runtime instance data.
3051

3052
  """
3053
  _OP_REQP = ["instances"]
3054

    
3055
  def CheckPrereq(self):
3056
    """Check prerequisites.
3057

3058
    This only checks the optional instance list against the existing names.
3059

3060
    """
3061
    if not isinstance(self.op.instances, list):
3062
      raise errors.OpPrereqError, "Invalid argument type 'instances'"
3063
    if self.op.instances:
3064
      self.wanted_instances = []
3065
      names = self.op.instances
3066
      for name in names:
3067
        instance = self.cfg.GetInstanceInfo(self.cfg.ExpandInstanceName(name))
3068
        if instance is None:
3069
          raise errors.OpPrereqError, ("No such instance name '%s'" % name)
3070
      self.wanted_instances.append(instance)
3071
    else:
3072
      self.wanted_instances = [self.cfg.GetInstanceInfo(name) for name
3073
                               in self.cfg.GetInstanceList()]
3074
    return
3075

    
3076

    
3077
  def _ComputeDiskStatus(self, instance, snode, dev):
3078
    """Compute block device status.
3079

3080
    """
3081
    self.cfg.SetDiskID(dev, instance.primary_node)
3082
    dev_pstatus = rpc.call_blockdev_find(instance.primary_node, dev)
3083
    if dev.dev_type == "drbd":
3084
      # we change the snode then (otherwise we use the one passed in)
3085
      if dev.logical_id[0] == instance.primary_node:
3086
        snode = dev.logical_id[1]
3087
      else:
3088
        snode = dev.logical_id[0]
3089

    
3090
    if snode:
3091
      self.cfg.SetDiskID(dev, snode)
3092
      dev_sstatus = rpc.call_blockdev_find(snode, dev)
3093
    else:
3094
      dev_sstatus = None
3095

    
3096
    if dev.children:
3097
      dev_children = [self._ComputeDiskStatus(instance, snode, child)
3098
                      for child in dev.children]
3099
    else:
3100
      dev_children = []
3101

    
3102
    data = {
3103
      "iv_name": dev.iv_name,
3104
      "dev_type": dev.dev_type,
3105
      "logical_id": dev.logical_id,
3106
      "physical_id": dev.physical_id,
3107
      "pstatus": dev_pstatus,
3108
      "sstatus": dev_sstatus,
3109
      "children": dev_children,
3110
      }
3111

    
3112
    return data
3113

    
3114
  def Exec(self, feedback_fn):
3115
    """Gather and return data"""
3116

    
3117
    result = {}
3118
    for instance in self.wanted_instances:
3119
      remote_info = rpc.call_instance_info(instance.primary_node,
3120
                                                instance.name)
3121
      if remote_info and "state" in remote_info:
3122
        remote_state = "up"
3123
      else:
3124
        remote_state = "down"
3125
      if instance.status == "down":
3126
        config_state = "down"
3127
      else:
3128
        config_state = "up"
3129

    
3130
      disks = [self._ComputeDiskStatus(instance, None, device)
3131
               for device in instance.disks]
3132

    
3133
      idict = {
3134
        "name": instance.name,
3135
        "config_state": config_state,
3136
        "run_state": remote_state,
3137
        "pnode": instance.primary_node,
3138
        "snodes": instance.secondary_nodes,
3139
        "os": instance.os,
3140
        "memory": instance.memory,
3141
        "nics": [(nic.mac, nic.ip, nic.bridge) for nic in instance.nics],
3142
        "disks": disks,
3143
        }
3144

    
3145
      result[instance.name] = idict
3146

    
3147
    return result
3148

    
3149

    
3150
class LUQueryNodeData(NoHooksLU):
3151
  """Logical unit for querying node data.
3152

3153
  """
3154
  _OP_REQP = ["nodes"]
3155

    
3156
  def CheckPrereq(self):
3157
    """Check prerequisites.
3158

3159
    This only checks the optional node list against the existing names.
3160

3161
    """
3162
    self.wanted_nodes = _GetWantedNodes(self, self.op.nodes)
3163

    
3164
  def Exec(self, feedback_fn):
3165
    """Compute and return the list of nodes.
3166

3167
    """
3168

    
3169
    ilist = [self.cfg.GetInstanceInfo(iname) for iname
3170
             in self.cfg.GetInstanceList()]
3171
    result = []
3172
    for node in self.wanted_nodes:
3173
      result.append((node.name, node.primary_ip, node.secondary_ip,
3174
                     [inst.name for inst in ilist
3175
                      if inst.primary_node == node.name],
3176
                     [inst.name for inst in ilist
3177
                      if node.name in inst.secondary_nodes],
3178
                     ))
3179
    return result
3180

    
3181

    
3182
class LUSetInstanceParms(LogicalUnit):
3183
  """Modifies an instances's parameters.
3184

3185
  """
3186
  HPATH = "instance-modify"
3187
  HTYPE = constants.HTYPE_INSTANCE
3188
  _OP_REQP = ["instance_name"]
3189

    
3190
  def BuildHooksEnv(self):
3191
    """Build hooks env.
3192

3193
    This runs on the master, primary and secondaries.
3194

3195
    """
3196
    env = {
3197
      "INSTANCE_NAME": self.op.instance_name,
3198
      }
3199
    if self.mem:
3200
      env["MEM_SIZE"] = self.mem
3201
    if self.vcpus:
3202
      env["VCPUS"] = self.vcpus
3203
    if self.do_ip:
3204
      env["INSTANCE_IP"] = self.ip
3205
    if self.bridge:
3206
      env["BRIDGE"] = self.bridge
3207

    
3208
    nl = [self.sstore.GetMasterNode(),
3209
          self.instance.primary_node] + list(self.instance.secondary_nodes)
3210

    
3211
    return env, nl, nl
3212

    
3213
  def CheckPrereq(self):
3214
    """Check prerequisites.
3215

3216
    This only checks the instance list against the existing names.
3217

3218
    """
3219
    self.mem = getattr(self.op, "mem", None)
3220
    self.vcpus = getattr(self.op, "vcpus", None)
3221
    self.ip = getattr(self.op, "ip", None)
3222
    self.bridge = getattr(self.op, "bridge", None)
3223
    if [self.mem, self.vcpus, self.ip, self.bridge].count(None) == 4:
3224
      raise errors.OpPrereqError, ("No changes submitted")
3225
    if self.mem is not None:
3226
      try:
3227
        self.mem = int(self.mem)
3228
      except ValueError, err:
3229
        raise errors.OpPrereqError, ("Invalid memory size: %s" % str(err))
3230
    if self.vcpus is not None:
3231
      try:
3232
        self.vcpus = int(self.vcpus)
3233
      except ValueError, err:
3234
        raise errors.OpPrereqError, ("Invalid vcpus number: %s" % str(err))
3235
    if self.ip is not None:
3236
      self.do_ip = True
3237
      if self.ip.lower() == "none":
3238
        self.ip = None
3239
      else:
3240
        if not utils.IsValidIP(self.ip):
3241
          raise errors.OpPrereqError, ("Invalid IP address '%s'." % self.ip)
3242
    else:
3243
      self.do_ip = False
3244

    
3245
    instance = self.cfg.GetInstanceInfo(
3246
      self.cfg.ExpandInstanceName(self.op.instance_name))
3247
    if instance is None:
3248
      raise errors.OpPrereqError, ("No such instance name '%s'" %
3249
                                   self.op.instance_name)
3250
    self.op.instance_name = instance.name
3251
    self.instance = instance
3252
    return
3253

    
3254
  def Exec(self, feedback_fn):
3255
    """Modifies an instance.
3256

3257
    All parameters take effect only at the next restart of the instance.
3258
    """
3259
    result = []
3260
    instance = self.instance
3261
    if self.mem:
3262
      instance.memory = self.mem
3263
      result.append(("mem", self.mem))
3264
    if self.vcpus:
3265
      instance.vcpus = self.vcpus
3266
      result.append(("vcpus",  self.vcpus))
3267
    if self.do_ip:
3268
      instance.nics[0].ip = self.ip
3269
      result.append(("ip", self.ip))
3270
    if self.bridge:
3271
      instance.nics[0].bridge = self.bridge
3272
      result.append(("bridge", self.bridge))
3273

    
3274
    self.cfg.AddInstance(instance)
3275

    
3276
    return result
3277

    
3278

    
3279
class LUQueryExports(NoHooksLU):
3280
  """Query the exports list
3281

3282
  """
3283
  _OP_REQP = []
3284

    
3285
  def CheckPrereq(self):
3286
    """Check that the nodelist contains only existing nodes.
3287

3288
    """
3289
    self.nodes = _GetWantedNodes(self, getattr(self.op, "nodes", None))
3290

    
3291
  def Exec(self, feedback_fn):
3292
    """Compute the list of all the exported system images.
3293

3294
    Returns:
3295
      a dictionary with the structure node->(export-list)
3296
      where export-list is a list of the instances exported on
3297
      that node.
3298

3299
    """
3300
    return rpc.call_export_list([node.name for node in self.nodes])
3301

    
3302

    
3303
class LUExportInstance(LogicalUnit):
3304
  """Export an instance to an image in the cluster.
3305

3306
  """
3307
  HPATH = "instance-export"
3308
  HTYPE = constants.HTYPE_INSTANCE
3309
  _OP_REQP = ["instance_name", "target_node", "shutdown"]
3310

    
3311
  def BuildHooksEnv(self):
3312
    """Build hooks env.
3313

3314
    This will run on the master, primary node and target node.
3315

3316
    """
3317
    env = {
3318
      "INSTANCE_NAME": self.op.instance_name,
3319
      "EXPORT_NODE": self.op.target_node,
3320
      "EXPORT_DO_SHUTDOWN": self.op.shutdown,
3321
      }
3322
    nl = [self.sstore.GetMasterNode(), self.instance.primary_node,
3323
          self.op.target_node]
3324
    return env, nl, nl
3325

    
3326
  def CheckPrereq(self):
3327
    """Check prerequisites.
3328

3329
    This checks that the instance name is a valid one.
3330

3331
    """
3332
    instance_name = self.cfg.ExpandInstanceName(self.op.instance_name)
3333
    self.instance = self.cfg.GetInstanceInfo(instance_name)
3334
    if self.instance is None:
3335
      raise errors.OpPrereqError, ("Instance '%s' not found" %
3336
                                   self.op.instance_name)
3337

    
3338
    # node verification
3339
    dst_node_short = self.cfg.ExpandNodeName(self.op.target_node)
3340
    self.dst_node = self.cfg.GetNodeInfo(dst_node_short)
3341

    
3342
    if self.dst_node is None:
3343
      raise errors.OpPrereqError, ("Destination node '%s' is uknown." %
3344
                                   self.op.target_node)
3345
    self.op.target_node = self.dst_node.name
3346

    
3347
  def Exec(self, feedback_fn):
3348
    """Export an instance to an image in the cluster.
3349

3350
    """
3351
    instance = self.instance
3352
    dst_node = self.dst_node
3353
    src_node = instance.primary_node
3354
    # shutdown the instance, unless requested not to do so
3355
    if self.op.shutdown:
3356
      op = opcodes.OpShutdownInstance(instance_name=instance.name)
3357
      self.processor.ChainOpCode(op, feedback_fn)
3358

    
3359
    vgname = self.cfg.GetVGName()
3360

    
3361
    snap_disks = []
3362

    
3363
    try:
3364
      for disk in instance.disks:
3365
        if disk.iv_name == "sda":
3366
          # new_dev_name will be a snapshot of an lvm leaf of the one we passed
3367
          new_dev_name = rpc.call_blockdev_snapshot(src_node, disk)
3368

    
3369
          if not new_dev_name:
3370
            logger.Error("could not snapshot block device %s on node %s" %
3371
                         (disk.logical_id[1], src_node))
3372
          else:
3373
            new_dev = objects.Disk(dev_type="lvm", size=disk.size,
3374
                                      logical_id=(vgname, new_dev_name),
3375
                                      physical_id=(vgname, new_dev_name),
3376
                                      iv_name=disk.iv_name)
3377
            snap_disks.append(new_dev)
3378

    
3379
    finally:
3380
      if self.op.shutdown:
3381
        op = opcodes.OpStartupInstance(instance_name=instance.name,
3382
                                       force=False)
3383
        self.processor.ChainOpCode(op, feedback_fn)
3384

    
3385
    # TODO: check for size
3386

    
3387
    for dev in snap_disks:
3388
      if not rpc.call_snapshot_export(src_node, dev, dst_node.name,
3389
                                           instance):
3390
        logger.Error("could not export block device %s from node"
3391
                     " %s to node %s" %
3392
                     (dev.logical_id[1], src_node, dst_node.name))
3393
      if not rpc.call_blockdev_remove(src_node, dev):
3394
        logger.Error("could not remove snapshot block device %s from"
3395
                     " node %s" % (dev.logical_id[1], src_node))
3396

    
3397
    if not rpc.call_finalize_export(dst_node.name, instance, snap_disks):
3398
      logger.Error("could not finalize export for instance %s on node %s" %
3399
                   (instance.name, dst_node.name))
3400

    
3401
    nodelist = self.cfg.GetNodeList()
3402
    nodelist.remove(dst_node.name)
3403

    
3404
    # on one-node clusters nodelist will be empty after the removal
3405
    # if we proceed the backup would be removed because OpQueryExports
3406
    # substitutes an empty list with the full cluster node list.
3407
    if nodelist:
3408
      op = opcodes.OpQueryExports(nodes=nodelist)
3409
      exportlist = self.processor.ChainOpCode(op, feedback_fn)
3410
      for node in exportlist:
3411
        if instance.name in exportlist[node]:
3412
          if not rpc.call_export_remove(node, instance.name):
3413
            logger.Error("could not remove older export for instance %s"
3414
                         " on node %s" % (instance.name, node))