Statistics
| Branch: | Tag: | Revision:

root / lib / backend.py @ ff89a747

History | View | Annotate | Download (94.2 kB)

1
#
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
"""Functions used by the node daemon
23

24
@var _ALLOWED_UPLOAD_FILES: denotes which files are accepted in
25
     the L{UploadFile} function
26
@var _ALLOWED_CLEAN_DIRS: denotes which directories are accepted
27
     in the L{_CleanDirectory} function
28

29
"""
30

    
31
# pylint: disable-msg=E1103
32

    
33
# E1103: %s %r has no %r member (but some types could not be
34
# inferred), because the _TryOSFromDisk returns either (True, os_obj)
35
# or (False, "string") which confuses pylint
36

    
37

    
38
import os
39
import os.path
40
import shutil
41
import time
42
import stat
43
import errno
44
import re
45
import random
46
import logging
47
import tempfile
48
import zlib
49
import base64
50
import signal
51

    
52
from ganeti import errors
53
from ganeti import utils
54
from ganeti import ssh
55
from ganeti import hypervisor
56
from ganeti import constants
57
from ganeti import bdev
58
from ganeti import objects
59
from ganeti import ssconf
60
from ganeti import serializer
61

    
62

    
63
_BOOT_ID_PATH = "/proc/sys/kernel/random/boot_id"
64
_ALLOWED_CLEAN_DIRS = frozenset([
65
  constants.DATA_DIR,
66
  constants.JOB_QUEUE_ARCHIVE_DIR,
67
  constants.QUEUE_DIR,
68
  constants.CRYPTO_KEYS_DIR,
69
  ])
70
_MAX_SSL_CERT_VALIDITY = 7 * 24 * 60 * 60
71
_X509_KEY_FILE = "key"
72
_X509_CERT_FILE = "cert"
73
_IES_STATUS_FILE = "status"
74
_IES_PID_FILE = "pid"
75
_IES_CA_FILE = "ca"
76

    
77

    
78
class RPCFail(Exception):
79
  """Class denoting RPC failure.
80

81
  Its argument is the error message.
82

83
  """
84

    
85

    
86
def _Fail(msg, *args, **kwargs):
87
  """Log an error and the raise an RPCFail exception.
88

89
  This exception is then handled specially in the ganeti daemon and
90
  turned into a 'failed' return type. As such, this function is a
91
  useful shortcut for logging the error and returning it to the master
92
  daemon.
93

94
  @type msg: string
95
  @param msg: the text of the exception
96
  @raise RPCFail
97

98
  """
99
  if args:
100
    msg = msg % args
101
  if "log" not in kwargs or kwargs["log"]: # if we should log this error
102
    if "exc" in kwargs and kwargs["exc"]:
103
      logging.exception(msg)
104
    else:
105
      logging.error(msg)
106
  raise RPCFail(msg)
107

    
108

    
109
def _GetConfig():
110
  """Simple wrapper to return a SimpleStore.
111

112
  @rtype: L{ssconf.SimpleStore}
113
  @return: a SimpleStore instance
114

115
  """
116
  return ssconf.SimpleStore()
117

    
118

    
119
def _GetSshRunner(cluster_name):
120
  """Simple wrapper to return an SshRunner.
121

122
  @type cluster_name: str
123
  @param cluster_name: the cluster name, which is needed
124
      by the SshRunner constructor
125
  @rtype: L{ssh.SshRunner}
126
  @return: an SshRunner instance
127

128
  """
129
  return ssh.SshRunner(cluster_name)
130

    
131

    
132
def _Decompress(data):
133
  """Unpacks data compressed by the RPC client.
134

135
  @type data: list or tuple
136
  @param data: Data sent by RPC client
137
  @rtype: str
138
  @return: Decompressed data
139

140
  """
141
  assert isinstance(data, (list, tuple))
142
  assert len(data) == 2
143
  (encoding, content) = data
144
  if encoding == constants.RPC_ENCODING_NONE:
145
    return content
146
  elif encoding == constants.RPC_ENCODING_ZLIB_BASE64:
147
    return zlib.decompress(base64.b64decode(content))
148
  else:
149
    raise AssertionError("Unknown data encoding")
150

    
151

    
152
def _CleanDirectory(path, exclude=None):
153
  """Removes all regular files in a directory.
154

155
  @type path: str
156
  @param path: the directory to clean
157
  @type exclude: list
158
  @param exclude: list of files to be excluded, defaults
159
      to the empty list
160

161
  """
162
  if path not in _ALLOWED_CLEAN_DIRS:
163
    _Fail("Path passed to _CleanDirectory not in allowed clean targets: '%s'",
164
          path)
165

    
166
  if not os.path.isdir(path):
167
    return
168
  if exclude is None:
169
    exclude = []
170
  else:
171
    # Normalize excluded paths
172
    exclude = [os.path.normpath(i) for i in exclude]
173

    
174
  for rel_name in utils.ListVisibleFiles(path):
175
    full_name = utils.PathJoin(path, rel_name)
176
    if full_name in exclude:
177
      continue
178
    if os.path.isfile(full_name) and not os.path.islink(full_name):
179
      utils.RemoveFile(full_name)
180

    
181

    
182
def _BuildUploadFileList():
183
  """Build the list of allowed upload files.
184

185
  This is abstracted so that it's built only once at module import time.
186

187
  """
188
  allowed_files = set([
189
    constants.CLUSTER_CONF_FILE,
190
    constants.ETC_HOSTS,
191
    constants.SSH_KNOWN_HOSTS_FILE,
192
    constants.VNC_PASSWORD_FILE,
193
    constants.RAPI_CERT_FILE,
194
    constants.RAPI_USERS_FILE,
195
    constants.CONFD_HMAC_KEY,
196
    constants.CLUSTER_DOMAIN_SECRET_FILE,
197
    ])
198

    
199
  for hv_name in constants.HYPER_TYPES:
200
    hv_class = hypervisor.GetHypervisorClass(hv_name)
201
    allowed_files.update(hv_class.GetAncillaryFiles())
202

    
203
  return frozenset(allowed_files)
204

    
205

    
206
_ALLOWED_UPLOAD_FILES = _BuildUploadFileList()
207

    
208

    
209
def JobQueuePurge():
210
  """Removes job queue files and archived jobs.
211

212
  @rtype: tuple
213
  @return: True, None
214

215
  """
216
  _CleanDirectory(constants.QUEUE_DIR, exclude=[constants.JOB_QUEUE_LOCK_FILE])
217
  _CleanDirectory(constants.JOB_QUEUE_ARCHIVE_DIR)
218

    
219

    
220
def GetMasterInfo():
221
  """Returns master information.
222

223
  This is an utility function to compute master information, either
224
  for consumption here or from the node daemon.
225

226
  @rtype: tuple
227
  @return: master_netdev, master_ip, master_name
228
  @raise RPCFail: in case of errors
229

230
  """
231
  try:
232
    cfg = _GetConfig()
233
    master_netdev = cfg.GetMasterNetdev()
234
    master_ip = cfg.GetMasterIP()
235
    master_node = cfg.GetMasterNode()
236
  except errors.ConfigurationError, err:
237
    _Fail("Cluster configuration incomplete: %s", err, exc=True)
238
  return (master_netdev, master_ip, master_node)
239

    
240

    
241
def StartMaster(start_daemons, no_voting):
242
  """Activate local node as master node.
243

244
  The function will always try activate the IP address of the master
245
  (unless someone else has it). It will also start the master daemons,
246
  based on the start_daemons parameter.
247

248
  @type start_daemons: boolean
249
  @param start_daemons: whether to also start the master
250
      daemons (ganeti-masterd and ganeti-rapi)
251
  @type no_voting: boolean
252
  @param no_voting: whether to start ganeti-masterd without a node vote
253
      (if start_daemons is True), but still non-interactively
254
  @rtype: None
255

256
  """
257
  # GetMasterInfo will raise an exception if not able to return data
258
  master_netdev, master_ip, _ = GetMasterInfo()
259

    
260
  err_msgs = []
261
  if utils.TcpPing(master_ip, constants.DEFAULT_NODED_PORT):
262
    if utils.OwnIpAddress(master_ip):
263
      # we already have the ip:
264
      logging.debug("Master IP already configured, doing nothing")
265
    else:
266
      msg = "Someone else has the master ip, not activating"
267
      logging.error(msg)
268
      err_msgs.append(msg)
269
  else:
270
    result = utils.RunCmd(["ip", "address", "add", "%s/32" % master_ip,
271
                           "dev", master_netdev, "label",
272
                           "%s:0" % master_netdev])
273
    if result.failed:
274
      msg = "Can't activate master IP: %s" % result.output
275
      logging.error(msg)
276
      err_msgs.append(msg)
277

    
278
    result = utils.RunCmd(["arping", "-q", "-U", "-c 3", "-I", master_netdev,
279
                           "-s", master_ip, master_ip])
280
    # we'll ignore the exit code of arping
281

    
282
  # and now start the master and rapi daemons
283
  if start_daemons:
284
    if no_voting:
285
      masterd_args = "--no-voting --yes-do-it"
286
    else:
287
      masterd_args = ""
288

    
289
    env = {
290
      "EXTRA_MASTERD_ARGS": masterd_args,
291
      }
292

    
293
    result = utils.RunCmd([constants.DAEMON_UTIL, "start-master"], env=env)
294
    if result.failed:
295
      msg = "Can't start Ganeti master: %s" % result.output
296
      logging.error(msg)
297
      err_msgs.append(msg)
298

    
299
  if err_msgs:
300
    _Fail("; ".join(err_msgs))
301

    
302

    
303
def StopMaster(stop_daemons):
304
  """Deactivate this node as master.
305

306
  The function will always try to deactivate the IP address of the
307
  master. It will also stop the master daemons depending on the
308
  stop_daemons parameter.
309

310
  @type stop_daemons: boolean
311
  @param stop_daemons: whether to also stop the master daemons
312
      (ganeti-masterd and ganeti-rapi)
313
  @rtype: None
314

315
  """
316
  # TODO: log and report back to the caller the error failures; we
317
  # need to decide in which case we fail the RPC for this
318

    
319
  # GetMasterInfo will raise an exception if not able to return data
320
  master_netdev, master_ip, _ = GetMasterInfo()
321

    
322
  result = utils.RunCmd(["ip", "address", "del", "%s/32" % master_ip,
323
                         "dev", master_netdev])
324
  if result.failed:
325
    logging.error("Can't remove the master IP, error: %s", result.output)
326
    # but otherwise ignore the failure
327

    
328
  if stop_daemons:
329
    result = utils.RunCmd([constants.DAEMON_UTIL, "stop-master"])
330
    if result.failed:
331
      logging.error("Could not stop Ganeti master, command %s had exitcode %s"
332
                    " and error %s",
333
                    result.cmd, result.exit_code, result.output)
334

    
335

    
336
def AddNode(dsa, dsapub, rsa, rsapub, sshkey, sshpub):
337
  """Joins this node to the cluster.
338

339
  This does the following:
340
      - updates the hostkeys of the machine (rsa and dsa)
341
      - adds the ssh private key to the user
342
      - adds the ssh public key to the users' authorized_keys file
343

344
  @type dsa: str
345
  @param dsa: the DSA private key to write
346
  @type dsapub: str
347
  @param dsapub: the DSA public key to write
348
  @type rsa: str
349
  @param rsa: the RSA private key to write
350
  @type rsapub: str
351
  @param rsapub: the RSA public key to write
352
  @type sshkey: str
353
  @param sshkey: the SSH private key to write
354
  @type sshpub: str
355
  @param sshpub: the SSH public key to write
356
  @rtype: boolean
357
  @return: the success of the operation
358

359
  """
360
  sshd_keys =  [(constants.SSH_HOST_RSA_PRIV, rsa, 0600),
361
                (constants.SSH_HOST_RSA_PUB, rsapub, 0644),
362
                (constants.SSH_HOST_DSA_PRIV, dsa, 0600),
363
                (constants.SSH_HOST_DSA_PUB, dsapub, 0644)]
364
  for name, content, mode in sshd_keys:
365
    utils.WriteFile(name, data=content, mode=mode)
366

    
367
  try:
368
    priv_key, pub_key, auth_keys = ssh.GetUserFiles(constants.GANETI_RUNAS,
369
                                                    mkdir=True)
370
  except errors.OpExecError, err:
371
    _Fail("Error while processing user ssh files: %s", err, exc=True)
372

    
373
  for name, content in [(priv_key, sshkey), (pub_key, sshpub)]:
374
    utils.WriteFile(name, data=content, mode=0600)
375

    
376
  utils.AddAuthorizedKey(auth_keys, sshpub)
377

    
378
  result = utils.RunCmd([constants.DAEMON_UTIL, "reload-ssh-keys"])
379
  if result.failed:
380
    _Fail("Unable to reload SSH keys (command %r, exit code %s, output %r)",
381
          result.cmd, result.exit_code, result.output)
382

    
383

    
384
def LeaveCluster(modify_ssh_setup):
385
  """Cleans up and remove the current node.
386

387
  This function cleans up and prepares the current node to be removed
388
  from the cluster.
389

390
  If processing is successful, then it raises an
391
  L{errors.QuitGanetiException} which is used as a special case to
392
  shutdown the node daemon.
393

394
  @param modify_ssh_setup: boolean
395

396
  """
397
  _CleanDirectory(constants.DATA_DIR)
398
  _CleanDirectory(constants.CRYPTO_KEYS_DIR)
399
  JobQueuePurge()
400

    
401
  if modify_ssh_setup:
402
    try:
403
      priv_key, pub_key, auth_keys = ssh.GetUserFiles(constants.GANETI_RUNAS)
404

    
405
      utils.RemoveAuthorizedKey(auth_keys, utils.ReadFile(pub_key))
406

    
407
      utils.RemoveFile(priv_key)
408
      utils.RemoveFile(pub_key)
409
    except errors.OpExecError:
410
      logging.exception("Error while processing ssh files")
411

    
412
  try:
413
    utils.RemoveFile(constants.CONFD_HMAC_KEY)
414
    utils.RemoveFile(constants.RAPI_CERT_FILE)
415
    utils.RemoveFile(constants.NODED_CERT_FILE)
416
  except: # pylint: disable-msg=W0702
417
    logging.exception("Error while removing cluster secrets")
418

    
419
  result = utils.RunCmd([constants.DAEMON_UTIL, "stop", constants.CONFD])
420
  if result.failed:
421
    logging.error("Command %s failed with exitcode %s and error %s",
422
                  result.cmd, result.exit_code, result.output)
423

    
424
  # Raise a custom exception (handled in ganeti-noded)
425
  raise errors.QuitGanetiException(True, 'Shutdown scheduled')
426

    
427

    
428
def GetNodeInfo(vgname, hypervisor_type):
429
  """Gives back a hash with different information about the node.
430

431
  @type vgname: C{string}
432
  @param vgname: the name of the volume group to ask for disk space information
433
  @type hypervisor_type: C{str}
434
  @param hypervisor_type: the name of the hypervisor to ask for
435
      memory information
436
  @rtype: C{dict}
437
  @return: dictionary with the following keys:
438
      - vg_size is the size of the configured volume group in MiB
439
      - vg_free is the free size of the volume group in MiB
440
      - memory_dom0 is the memory allocated for domain0 in MiB
441
      - memory_free is the currently available (free) ram in MiB
442
      - memory_total is the total number of ram in MiB
443

444
  """
445
  outputarray = {}
446
  vginfo = _GetVGInfo(vgname)
447
  outputarray['vg_size'] = vginfo['vg_size']
448
  outputarray['vg_free'] = vginfo['vg_free']
449

    
450
  hyper = hypervisor.GetHypervisor(hypervisor_type)
451
  hyp_info = hyper.GetNodeInfo()
452
  if hyp_info is not None:
453
    outputarray.update(hyp_info)
454

    
455
  outputarray["bootid"] = utils.ReadFile(_BOOT_ID_PATH, size=128).rstrip("\n")
456

    
457
  return outputarray
458

    
459

    
460
def VerifyNode(what, cluster_name):
461
  """Verify the status of the local node.
462

463
  Based on the input L{what} parameter, various checks are done on the
464
  local node.
465

466
  If the I{filelist} key is present, this list of
467
  files is checksummed and the file/checksum pairs are returned.
468

469
  If the I{nodelist} key is present, we check that we have
470
  connectivity via ssh with the target nodes (and check the hostname
471
  report).
472

473
  If the I{node-net-test} key is present, we check that we have
474
  connectivity to the given nodes via both primary IP and, if
475
  applicable, secondary IPs.
476

477
  @type what: C{dict}
478
  @param what: a dictionary of things to check:
479
      - filelist: list of files for which to compute checksums
480
      - nodelist: list of nodes we should check ssh communication with
481
      - node-net-test: list of nodes we should check node daemon port
482
        connectivity with
483
      - hypervisor: list with hypervisors to run the verify for
484
  @rtype: dict
485
  @return: a dictionary with the same keys as the input dict, and
486
      values representing the result of the checks
487

488
  """
489
  result = {}
490
  my_name = utils.HostInfo().name
491
  port = utils.GetDaemonPort(constants.NODED)
492

    
493
  if constants.NV_HYPERVISOR in what:
494
    result[constants.NV_HYPERVISOR] = tmp = {}
495
    for hv_name in what[constants.NV_HYPERVISOR]:
496
      try:
497
        val = hypervisor.GetHypervisor(hv_name).Verify()
498
      except errors.HypervisorError, err:
499
        val = "Error while checking hypervisor: %s" % str(err)
500
      tmp[hv_name] = val
501

    
502
  if constants.NV_FILELIST in what:
503
    result[constants.NV_FILELIST] = utils.FingerprintFiles(
504
      what[constants.NV_FILELIST])
505

    
506
  if constants.NV_NODELIST in what:
507
    result[constants.NV_NODELIST] = tmp = {}
508
    random.shuffle(what[constants.NV_NODELIST])
509
    for node in what[constants.NV_NODELIST]:
510
      success, message = _GetSshRunner(cluster_name).VerifyNodeHostname(node)
511
      if not success:
512
        tmp[node] = message
513

    
514
  if constants.NV_NODENETTEST in what:
515
    result[constants.NV_NODENETTEST] = tmp = {}
516
    my_pip = my_sip = None
517
    for name, pip, sip in what[constants.NV_NODENETTEST]:
518
      if name == my_name:
519
        my_pip = pip
520
        my_sip = sip
521
        break
522
    if not my_pip:
523
      tmp[my_name] = ("Can't find my own primary/secondary IP"
524
                      " in the node list")
525
    else:
526
      for name, pip, sip in what[constants.NV_NODENETTEST]:
527
        fail = []
528
        if not utils.TcpPing(pip, port, source=my_pip):
529
          fail.append("primary")
530
        if sip != pip:
531
          if not utils.TcpPing(sip, port, source=my_sip):
532
            fail.append("secondary")
533
        if fail:
534
          tmp[name] = ("failure using the %s interface(s)" %
535
                       " and ".join(fail))
536

    
537
  if constants.NV_MASTERIP in what:
538
    # FIXME: add checks on incoming data structures (here and in the
539
    # rest of the function)
540
    master_name, master_ip = what[constants.NV_MASTERIP]
541
    if master_name == my_name:
542
      source = constants.LOCALHOST_IP_ADDRESS
543
    else:
544
      source = None
545
    result[constants.NV_MASTERIP] = utils.TcpPing(master_ip, port,
546
                                                  source=source)
547

    
548
  if constants.NV_LVLIST in what:
549
    try:
550
      val = GetVolumeList(what[constants.NV_LVLIST])
551
    except RPCFail, err:
552
      val = str(err)
553
    result[constants.NV_LVLIST] = val
554

    
555
  if constants.NV_INSTANCELIST in what:
556
    # GetInstanceList can fail
557
    try:
558
      val = GetInstanceList(what[constants.NV_INSTANCELIST])
559
    except RPCFail, err:
560
      val = str(err)
561
    result[constants.NV_INSTANCELIST] = val
562

    
563
  if constants.NV_VGLIST in what:
564
    result[constants.NV_VGLIST] = utils.ListVolumeGroups()
565

    
566
  if constants.NV_PVLIST in what:
567
    result[constants.NV_PVLIST] = \
568
      bdev.LogicalVolume.GetPVInfo(what[constants.NV_PVLIST],
569
                                   filter_allocatable=False)
570

    
571
  if constants.NV_VERSION in what:
572
    result[constants.NV_VERSION] = (constants.PROTOCOL_VERSION,
573
                                    constants.RELEASE_VERSION)
574

    
575
  if constants.NV_HVINFO in what:
576
    hyper = hypervisor.GetHypervisor(what[constants.NV_HVINFO])
577
    result[constants.NV_HVINFO] = hyper.GetNodeInfo()
578

    
579
  if constants.NV_DRBDLIST in what:
580
    try:
581
      used_minors = bdev.DRBD8.GetUsedDevs().keys()
582
    except errors.BlockDeviceError, err:
583
      logging.warning("Can't get used minors list", exc_info=True)
584
      used_minors = str(err)
585
    result[constants.NV_DRBDLIST] = used_minors
586

    
587
  if constants.NV_NODESETUP in what:
588
    result[constants.NV_NODESETUP] = tmpr = []
589
    if not os.path.isdir("/sys/block") or not os.path.isdir("/sys/class/net"):
590
      tmpr.append("The sysfs filesytem doesn't seem to be mounted"
591
                  " under /sys, missing required directories /sys/block"
592
                  " and /sys/class/net")
593
    if (not os.path.isdir("/proc/sys") or
594
        not os.path.isfile("/proc/sysrq-trigger")):
595
      tmpr.append("The procfs filesystem doesn't seem to be mounted"
596
                  " under /proc, missing required directory /proc/sys and"
597
                  " the file /proc/sysrq-trigger")
598

    
599
  if constants.NV_TIME in what:
600
    result[constants.NV_TIME] = utils.SplitTime(time.time())
601

    
602
  return result
603

    
604

    
605
def GetVolumeList(vg_name):
606
  """Compute list of logical volumes and their size.
607

608
  @type vg_name: str
609
  @param vg_name: the volume group whose LVs we should list
610
  @rtype: dict
611
  @return:
612
      dictionary of all partions (key) with value being a tuple of
613
      their size (in MiB), inactive and online status::
614

615
        {'test1': ('20.06', True, True)}
616

617
      in case of errors, a string is returned with the error
618
      details.
619

620
  """
621
  lvs = {}
622
  sep = '|'
623
  result = utils.RunCmd(["lvs", "--noheadings", "--units=m", "--nosuffix",
624
                         "--separator=%s" % sep,
625
                         "-olv_name,lv_size,lv_attr", vg_name])
626
  if result.failed:
627
    _Fail("Failed to list logical volumes, lvs output: %s", result.output)
628

    
629
  valid_line_re = re.compile("^ *([^|]+)\|([0-9.]+)\|([^|]{6})\|?$")
630
  for line in result.stdout.splitlines():
631
    line = line.strip()
632
    match = valid_line_re.match(line)
633
    if not match:
634
      logging.error("Invalid line returned from lvs output: '%s'", line)
635
      continue
636
    name, size, attr = match.groups()
637
    inactive = attr[4] == '-'
638
    online = attr[5] == 'o'
639
    virtual = attr[0] == 'v'
640
    if virtual:
641
      # we don't want to report such volumes as existing, since they
642
      # don't really hold data
643
      continue
644
    lvs[name] = (size, inactive, online)
645

    
646
  return lvs
647

    
648

    
649
def ListVolumeGroups():
650
  """List the volume groups and their size.
651

652
  @rtype: dict
653
  @return: dictionary with keys volume name and values the
654
      size of the volume
655

656
  """
657
  return utils.ListVolumeGroups()
658

    
659

    
660
def NodeVolumes():
661
  """List all volumes on this node.
662

663
  @rtype: list
664
  @return:
665
    A list of dictionaries, each having four keys:
666
      - name: the logical volume name,
667
      - size: the size of the logical volume
668
      - dev: the physical device on which the LV lives
669
      - vg: the volume group to which it belongs
670

671
    In case of errors, we return an empty list and log the
672
    error.
673

674
    Note that since a logical volume can live on multiple physical
675
    volumes, the resulting list might include a logical volume
676
    multiple times.
677

678
  """
679
  result = utils.RunCmd(["lvs", "--noheadings", "--units=m", "--nosuffix",
680
                         "--separator=|",
681
                         "--options=lv_name,lv_size,devices,vg_name"])
682
  if result.failed:
683
    _Fail("Failed to list logical volumes, lvs output: %s",
684
          result.output)
685

    
686
  def parse_dev(dev):
687
    return dev.split('(')[0]
688

    
689
  def handle_dev(dev):
690
    return [parse_dev(x) for x in dev.split(",")]
691

    
692
  def map_line(line):
693
    line = [v.strip() for v in line]
694
    return [{'name': line[0], 'size': line[1],
695
             'dev': dev, 'vg': line[3]} for dev in handle_dev(line[2])]
696

    
697
  all_devs = []
698
  for line in result.stdout.splitlines():
699
    if line.count('|') >= 3:
700
      all_devs.extend(map_line(line.split('|')))
701
    else:
702
      logging.warning("Strange line in the output from lvs: '%s'", line)
703
  return all_devs
704

    
705

    
706
def BridgesExist(bridges_list):
707
  """Check if a list of bridges exist on the current node.
708

709
  @rtype: boolean
710
  @return: C{True} if all of them exist, C{False} otherwise
711

712
  """
713
  missing = []
714
  for bridge in bridges_list:
715
    if not utils.BridgeExists(bridge):
716
      missing.append(bridge)
717

    
718
  if missing:
719
    _Fail("Missing bridges %s", utils.CommaJoin(missing))
720

    
721

    
722
def GetInstanceList(hypervisor_list):
723
  """Provides a list of instances.
724

725
  @type hypervisor_list: list
726
  @param hypervisor_list: the list of hypervisors to query information
727

728
  @rtype: list
729
  @return: a list of all running instances on the current node
730
    - instance1.example.com
731
    - instance2.example.com
732

733
  """
734
  results = []
735
  for hname in hypervisor_list:
736
    try:
737
      names = hypervisor.GetHypervisor(hname).ListInstances()
738
      results.extend(names)
739
    except errors.HypervisorError, err:
740
      _Fail("Error enumerating instances (hypervisor %s): %s",
741
            hname, err, exc=True)
742

    
743
  return results
744

    
745

    
746
def GetInstanceInfo(instance, hname):
747
  """Gives back the information about an instance as a dictionary.
748

749
  @type instance: string
750
  @param instance: the instance name
751
  @type hname: string
752
  @param hname: the hypervisor type of the instance
753

754
  @rtype: dict
755
  @return: dictionary with the following keys:
756
      - memory: memory size of instance (int)
757
      - state: xen state of instance (string)
758
      - time: cpu time of instance (float)
759

760
  """
761
  output = {}
762

    
763
  iinfo = hypervisor.GetHypervisor(hname).GetInstanceInfo(instance)
764
  if iinfo is not None:
765
    output['memory'] = iinfo[2]
766
    output['state'] = iinfo[4]
767
    output['time'] = iinfo[5]
768

    
769
  return output
770

    
771

    
772
def GetInstanceMigratable(instance):
773
  """Gives whether an instance can be migrated.
774

775
  @type instance: L{objects.Instance}
776
  @param instance: object representing the instance to be checked.
777

778
  @rtype: tuple
779
  @return: tuple of (result, description) where:
780
      - result: whether the instance can be migrated or not
781
      - description: a description of the issue, if relevant
782

783
  """
784
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
785
  iname = instance.name
786
  if iname not in hyper.ListInstances():
787
    _Fail("Instance %s is not running", iname)
788

    
789
  for idx in range(len(instance.disks)):
790
    link_name = _GetBlockDevSymlinkPath(iname, idx)
791
    if not os.path.islink(link_name):
792
      _Fail("Instance %s was not restarted since ganeti 1.2.5", iname)
793

    
794

    
795
def GetAllInstancesInfo(hypervisor_list):
796
  """Gather data about all instances.
797

798
  This is the equivalent of L{GetInstanceInfo}, except that it
799
  computes data for all instances at once, thus being faster if one
800
  needs data about more than one instance.
801

802
  @type hypervisor_list: list
803
  @param hypervisor_list: list of hypervisors to query for instance data
804

805
  @rtype: dict
806
  @return: dictionary of instance: data, with data having the following keys:
807
      - memory: memory size of instance (int)
808
      - state: xen state of instance (string)
809
      - time: cpu time of instance (float)
810
      - vcpus: the number of vcpus
811

812
  """
813
  output = {}
814

    
815
  for hname in hypervisor_list:
816
    iinfo = hypervisor.GetHypervisor(hname).GetAllInstancesInfo()
817
    if iinfo:
818
      for name, _, memory, vcpus, state, times in iinfo:
819
        value = {
820
          'memory': memory,
821
          'vcpus': vcpus,
822
          'state': state,
823
          'time': times,
824
          }
825
        if name in output:
826
          # we only check static parameters, like memory and vcpus,
827
          # and not state and time which can change between the
828
          # invocations of the different hypervisors
829
          for key in 'memory', 'vcpus':
830
            if value[key] != output[name][key]:
831
              _Fail("Instance %s is running twice"
832
                    " with different parameters", name)
833
        output[name] = value
834

    
835
  return output
836

    
837

    
838
def _InstanceLogName(kind, os_name, instance):
839
  """Compute the OS log filename for a given instance and operation.
840

841
  The instance name and os name are passed in as strings since not all
842
  operations have these as part of an instance object.
843

844
  @type kind: string
845
  @param kind: the operation type (e.g. add, import, etc.)
846
  @type os_name: string
847
  @param os_name: the os name
848
  @type instance: string
849
  @param instance: the name of the instance being imported/added/etc.
850

851
  """
852
  # TODO: Use tempfile.mkstemp to create unique filename
853
  base = ("%s-%s-%s-%s.log" %
854
          (kind, os_name, instance, utils.TimestampForFilename()))
855
  return utils.PathJoin(constants.LOG_OS_DIR, base)
856

    
857

    
858
def InstanceOsAdd(instance, reinstall, debug):
859
  """Add an OS to an instance.
860

861
  @type instance: L{objects.Instance}
862
  @param instance: Instance whose OS is to be installed
863
  @type reinstall: boolean
864
  @param reinstall: whether this is an instance reinstall
865
  @type debug: integer
866
  @param debug: debug level, passed to the OS scripts
867
  @rtype: None
868

869
  """
870
  inst_os = OSFromDisk(instance.os)
871

    
872
  create_env = OSEnvironment(instance, inst_os, debug)
873
  if reinstall:
874
    create_env['INSTANCE_REINSTALL'] = "1"
875

    
876
  logfile = _InstanceLogName("add", instance.os, instance.name)
877

    
878
  result = utils.RunCmd([inst_os.create_script], env=create_env,
879
                        cwd=inst_os.path, output=logfile,)
880
  if result.failed:
881
    logging.error("os create command '%s' returned error: %s, logfile: %s,"
882
                  " output: %s", result.cmd, result.fail_reason, logfile,
883
                  result.output)
884
    lines = [utils.SafeEncode(val)
885
             for val in utils.TailFile(logfile, lines=20)]
886
    _Fail("OS create script failed (%s), last lines in the"
887
          " log file:\n%s", result.fail_reason, "\n".join(lines), log=False)
888

    
889

    
890
def RunRenameInstance(instance, old_name, debug):
891
  """Run the OS rename script for an instance.
892

893
  @type instance: L{objects.Instance}
894
  @param instance: Instance whose OS is to be installed
895
  @type old_name: string
896
  @param old_name: previous instance name
897
  @type debug: integer
898
  @param debug: debug level, passed to the OS scripts
899
  @rtype: boolean
900
  @return: the success of the operation
901

902
  """
903
  inst_os = OSFromDisk(instance.os)
904

    
905
  rename_env = OSEnvironment(instance, inst_os, debug)
906
  rename_env['OLD_INSTANCE_NAME'] = old_name
907

    
908
  logfile = _InstanceLogName("rename", instance.os,
909
                             "%s-%s" % (old_name, instance.name))
910

    
911
  result = utils.RunCmd([inst_os.rename_script], env=rename_env,
912
                        cwd=inst_os.path, output=logfile)
913

    
914
  if result.failed:
915
    logging.error("os create command '%s' returned error: %s output: %s",
916
                  result.cmd, result.fail_reason, result.output)
917
    lines = [utils.SafeEncode(val)
918
             for val in utils.TailFile(logfile, lines=20)]
919
    _Fail("OS rename script failed (%s), last lines in the"
920
          " log file:\n%s", result.fail_reason, "\n".join(lines), log=False)
921

    
922

    
923
def _GetVGInfo(vg_name):
924
  """Get information about the volume group.
925

926
  @type vg_name: str
927
  @param vg_name: the volume group which we query
928
  @rtype: dict
929
  @return:
930
    A dictionary with the following keys:
931
      - C{vg_size} is the total size of the volume group in MiB
932
      - C{vg_free} is the free size of the volume group in MiB
933
      - C{pv_count} are the number of physical disks in that VG
934

935
    If an error occurs during gathering of data, we return the same dict
936
    with keys all set to None.
937

938
  """
939
  retdic = dict.fromkeys(["vg_size", "vg_free", "pv_count"])
940

    
941
  retval = utils.RunCmd(["vgs", "-ovg_size,vg_free,pv_count", "--noheadings",
942
                         "--nosuffix", "--units=m", "--separator=:", vg_name])
943

    
944
  if retval.failed:
945
    logging.error("volume group %s not present", vg_name)
946
    return retdic
947
  valarr = retval.stdout.strip().rstrip(':').split(':')
948
  if len(valarr) == 3:
949
    try:
950
      retdic = {
951
        "vg_size": int(round(float(valarr[0]), 0)),
952
        "vg_free": int(round(float(valarr[1]), 0)),
953
        "pv_count": int(valarr[2]),
954
        }
955
    except (TypeError, ValueError), err:
956
      logging.exception("Fail to parse vgs output: %s", err)
957
  else:
958
    logging.error("vgs output has the wrong number of fields (expected"
959
                  " three): %s", str(valarr))
960
  return retdic
961

    
962

    
963
def _GetBlockDevSymlinkPath(instance_name, idx):
964
  return utils.PathJoin(constants.DISK_LINKS_DIR,
965
                        "%s:%d" % (instance_name, idx))
966

    
967

    
968
def _SymlinkBlockDev(instance_name, device_path, idx):
969
  """Set up symlinks to a instance's block device.
970

971
  This is an auxiliary function run when an instance is start (on the primary
972
  node) or when an instance is migrated (on the target node).
973

974

975
  @param instance_name: the name of the target instance
976
  @param device_path: path of the physical block device, on the node
977
  @param idx: the disk index
978
  @return: absolute path to the disk's symlink
979

980
  """
981
  link_name = _GetBlockDevSymlinkPath(instance_name, idx)
982
  try:
983
    os.symlink(device_path, link_name)
984
  except OSError, err:
985
    if err.errno == errno.EEXIST:
986
      if (not os.path.islink(link_name) or
987
          os.readlink(link_name) != device_path):
988
        os.remove(link_name)
989
        os.symlink(device_path, link_name)
990
    else:
991
      raise
992

    
993
  return link_name
994

    
995

    
996
def _RemoveBlockDevLinks(instance_name, disks):
997
  """Remove the block device symlinks belonging to the given instance.
998

999
  """
1000
  for idx, _ in enumerate(disks):
1001
    link_name = _GetBlockDevSymlinkPath(instance_name, idx)
1002
    if os.path.islink(link_name):
1003
      try:
1004
        os.remove(link_name)
1005
      except OSError:
1006
        logging.exception("Can't remove symlink '%s'", link_name)
1007

    
1008

    
1009
def _GatherAndLinkBlockDevs(instance):
1010
  """Set up an instance's block device(s).
1011

1012
  This is run on the primary node at instance startup. The block
1013
  devices must be already assembled.
1014

1015
  @type instance: L{objects.Instance}
1016
  @param instance: the instance whose disks we shoul assemble
1017
  @rtype: list
1018
  @return: list of (disk_object, device_path)
1019

1020
  """
1021
  block_devices = []
1022
  for idx, disk in enumerate(instance.disks):
1023
    device = _RecursiveFindBD(disk)
1024
    if device is None:
1025
      raise errors.BlockDeviceError("Block device '%s' is not set up." %
1026
                                    str(disk))
1027
    device.Open()
1028
    try:
1029
      link_name = _SymlinkBlockDev(instance.name, device.dev_path, idx)
1030
    except OSError, e:
1031
      raise errors.BlockDeviceError("Cannot create block device symlink: %s" %
1032
                                    e.strerror)
1033

    
1034
    block_devices.append((disk, link_name))
1035

    
1036
  return block_devices
1037

    
1038

    
1039
def StartInstance(instance):
1040
  """Start an instance.
1041

1042
  @type instance: L{objects.Instance}
1043
  @param instance: the instance object
1044
  @rtype: None
1045

1046
  """
1047
  running_instances = GetInstanceList([instance.hypervisor])
1048

    
1049
  if instance.name in running_instances:
1050
    logging.info("Instance %s already running, not starting", instance.name)
1051
    return
1052

    
1053
  try:
1054
    block_devices = _GatherAndLinkBlockDevs(instance)
1055
    hyper = hypervisor.GetHypervisor(instance.hypervisor)
1056
    hyper.StartInstance(instance, block_devices)
1057
  except errors.BlockDeviceError, err:
1058
    _Fail("Block device error: %s", err, exc=True)
1059
  except errors.HypervisorError, err:
1060
    _RemoveBlockDevLinks(instance.name, instance.disks)
1061
    _Fail("Hypervisor error: %s", err, exc=True)
1062

    
1063

    
1064
def InstanceShutdown(instance, timeout):
1065
  """Shut an instance down.
1066

1067
  @note: this functions uses polling with a hardcoded timeout.
1068

1069
  @type instance: L{objects.Instance}
1070
  @param instance: the instance object
1071
  @type timeout: integer
1072
  @param timeout: maximum timeout for soft shutdown
1073
  @rtype: None
1074

1075
  """
1076
  hv_name = instance.hypervisor
1077
  hyper = hypervisor.GetHypervisor(hv_name)
1078
  iname = instance.name
1079

    
1080
  if instance.name not in hyper.ListInstances():
1081
    logging.info("Instance %s not running, doing nothing", iname)
1082
    return
1083

    
1084
  class _TryShutdown:
1085
    def __init__(self):
1086
      self.tried_once = False
1087

    
1088
    def __call__(self):
1089
      if iname not in hyper.ListInstances():
1090
        return
1091

    
1092
      try:
1093
        hyper.StopInstance(instance, retry=self.tried_once)
1094
      except errors.HypervisorError, err:
1095
        if iname not in hyper.ListInstances():
1096
          # if the instance is no longer existing, consider this a
1097
          # success and go to cleanup
1098
          return
1099

    
1100
        _Fail("Failed to stop instance %s: %s", iname, err)
1101

    
1102
      self.tried_once = True
1103

    
1104
      raise utils.RetryAgain()
1105

    
1106
  try:
1107
    utils.Retry(_TryShutdown(), 5, timeout)
1108
  except utils.RetryTimeout:
1109
    # the shutdown did not succeed
1110
    logging.error("Shutdown of '%s' unsuccessful, forcing", iname)
1111

    
1112
    try:
1113
      hyper.StopInstance(instance, force=True)
1114
    except errors.HypervisorError, err:
1115
      if iname in hyper.ListInstances():
1116
        # only raise an error if the instance still exists, otherwise
1117
        # the error could simply be "instance ... unknown"!
1118
        _Fail("Failed to force stop instance %s: %s", iname, err)
1119

    
1120
    time.sleep(1)
1121

    
1122
    if iname in hyper.ListInstances():
1123
      _Fail("Could not shutdown instance %s even by destroy", iname)
1124

    
1125
  try:
1126
    hyper.CleanupInstance(instance.name)
1127
  except errors.HypervisorError, err:
1128
    logging.warning("Failed to execute post-shutdown cleanup step: %s", err)
1129

    
1130
  _RemoveBlockDevLinks(iname, instance.disks)
1131

    
1132

    
1133
def InstanceReboot(instance, reboot_type, shutdown_timeout):
1134
  """Reboot an instance.
1135

1136
  @type instance: L{objects.Instance}
1137
  @param instance: the instance object to reboot
1138
  @type reboot_type: str
1139
  @param reboot_type: the type of reboot, one the following
1140
    constants:
1141
      - L{constants.INSTANCE_REBOOT_SOFT}: only reboot the
1142
        instance OS, do not recreate the VM
1143
      - L{constants.INSTANCE_REBOOT_HARD}: tear down and
1144
        restart the VM (at the hypervisor level)
1145
      - the other reboot type (L{constants.INSTANCE_REBOOT_FULL}) is
1146
        not accepted here, since that mode is handled differently, in
1147
        cmdlib, and translates into full stop and start of the
1148
        instance (instead of a call_instance_reboot RPC)
1149
  @type shutdown_timeout: integer
1150
  @param shutdown_timeout: maximum timeout for soft shutdown
1151
  @rtype: None
1152

1153
  """
1154
  running_instances = GetInstanceList([instance.hypervisor])
1155

    
1156
  if instance.name not in running_instances:
1157
    _Fail("Cannot reboot instance %s that is not running", instance.name)
1158

    
1159
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1160
  if reboot_type == constants.INSTANCE_REBOOT_SOFT:
1161
    try:
1162
      hyper.RebootInstance(instance)
1163
    except errors.HypervisorError, err:
1164
      _Fail("Failed to soft reboot instance %s: %s", instance.name, err)
1165
  elif reboot_type == constants.INSTANCE_REBOOT_HARD:
1166
    try:
1167
      InstanceShutdown(instance, shutdown_timeout)
1168
      return StartInstance(instance)
1169
    except errors.HypervisorError, err:
1170
      _Fail("Failed to hard reboot instance %s: %s", instance.name, err)
1171
  else:
1172
    _Fail("Invalid reboot_type received: %s", reboot_type)
1173

    
1174

    
1175
def MigrationInfo(instance):
1176
  """Gather information about an instance to be migrated.
1177

1178
  @type instance: L{objects.Instance}
1179
  @param instance: the instance definition
1180

1181
  """
1182
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1183
  try:
1184
    info = hyper.MigrationInfo(instance)
1185
  except errors.HypervisorError, err:
1186
    _Fail("Failed to fetch migration information: %s", err, exc=True)
1187
  return info
1188

    
1189

    
1190
def AcceptInstance(instance, info, target):
1191
  """Prepare the node to accept an instance.
1192

1193
  @type instance: L{objects.Instance}
1194
  @param instance: the instance definition
1195
  @type info: string/data (opaque)
1196
  @param info: migration information, from the source node
1197
  @type target: string
1198
  @param target: target host (usually ip), on this node
1199

1200
  """
1201
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1202
  try:
1203
    hyper.AcceptInstance(instance, info, target)
1204
  except errors.HypervisorError, err:
1205
    _Fail("Failed to accept instance: %s", err, exc=True)
1206

    
1207

    
1208
def FinalizeMigration(instance, info, success):
1209
  """Finalize any preparation to accept an instance.
1210

1211
  @type instance: L{objects.Instance}
1212
  @param instance: the instance definition
1213
  @type info: string/data (opaque)
1214
  @param info: migration information, from the source node
1215
  @type success: boolean
1216
  @param success: whether the migration was a success or a failure
1217

1218
  """
1219
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1220
  try:
1221
    hyper.FinalizeMigration(instance, info, success)
1222
  except errors.HypervisorError, err:
1223
    _Fail("Failed to finalize migration: %s", err, exc=True)
1224

    
1225

    
1226
def MigrateInstance(instance, target, live):
1227
  """Migrates an instance to another node.
1228

1229
  @type instance: L{objects.Instance}
1230
  @param instance: the instance definition
1231
  @type target: string
1232
  @param target: the target node name
1233
  @type live: boolean
1234
  @param live: whether the migration should be done live or not (the
1235
      interpretation of this parameter is left to the hypervisor)
1236
  @rtype: tuple
1237
  @return: a tuple of (success, msg) where:
1238
      - succes is a boolean denoting the success/failure of the operation
1239
      - msg is a string with details in case of failure
1240

1241
  """
1242
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1243

    
1244
  try:
1245
    hyper.MigrateInstance(instance, target, live)
1246
  except errors.HypervisorError, err:
1247
    _Fail("Failed to migrate instance: %s", err, exc=True)
1248

    
1249

    
1250
def BlockdevCreate(disk, size, owner, on_primary, info):
1251
  """Creates a block device for an instance.
1252

1253
  @type disk: L{objects.Disk}
1254
  @param disk: the object describing the disk we should create
1255
  @type size: int
1256
  @param size: the size of the physical underlying device, in MiB
1257
  @type owner: str
1258
  @param owner: the name of the instance for which disk is created,
1259
      used for device cache data
1260
  @type on_primary: boolean
1261
  @param on_primary:  indicates if it is the primary node or not
1262
  @type info: string
1263
  @param info: string that will be sent to the physical device
1264
      creation, used for example to set (LVM) tags on LVs
1265

1266
  @return: the new unique_id of the device (this can sometime be
1267
      computed only after creation), or None. On secondary nodes,
1268
      it's not required to return anything.
1269

1270
  """
1271
  # TODO: remove the obsolete 'size' argument
1272
  # pylint: disable-msg=W0613
1273
  clist = []
1274
  if disk.children:
1275
    for child in disk.children:
1276
      try:
1277
        crdev = _RecursiveAssembleBD(child, owner, on_primary)
1278
      except errors.BlockDeviceError, err:
1279
        _Fail("Can't assemble device %s: %s", child, err)
1280
      if on_primary or disk.AssembleOnSecondary():
1281
        # we need the children open in case the device itself has to
1282
        # be assembled
1283
        try:
1284
          # pylint: disable-msg=E1103
1285
          crdev.Open()
1286
        except errors.BlockDeviceError, err:
1287
          _Fail("Can't make child '%s' read-write: %s", child, err)
1288
      clist.append(crdev)
1289

    
1290
  try:
1291
    device = bdev.Create(disk.dev_type, disk.physical_id, clist, disk.size)
1292
  except errors.BlockDeviceError, err:
1293
    _Fail("Can't create block device: %s", err)
1294

    
1295
  if on_primary or disk.AssembleOnSecondary():
1296
    try:
1297
      device.Assemble()
1298
    except errors.BlockDeviceError, err:
1299
      _Fail("Can't assemble device after creation, unusual event: %s", err)
1300
    device.SetSyncSpeed(constants.SYNC_SPEED)
1301
    if on_primary or disk.OpenOnSecondary():
1302
      try:
1303
        device.Open(force=True)
1304
      except errors.BlockDeviceError, err:
1305
        _Fail("Can't make device r/w after creation, unusual event: %s", err)
1306
    DevCacheManager.UpdateCache(device.dev_path, owner,
1307
                                on_primary, disk.iv_name)
1308

    
1309
  device.SetInfo(info)
1310

    
1311
  return device.unique_id
1312

    
1313

    
1314
def BlockdevRemove(disk):
1315
  """Remove a block device.
1316

1317
  @note: This is intended to be called recursively.
1318

1319
  @type disk: L{objects.Disk}
1320
  @param disk: the disk object we should remove
1321
  @rtype: boolean
1322
  @return: the success of the operation
1323

1324
  """
1325
  msgs = []
1326
  try:
1327
    rdev = _RecursiveFindBD(disk)
1328
  except errors.BlockDeviceError, err:
1329
    # probably can't attach
1330
    logging.info("Can't attach to device %s in remove", disk)
1331
    rdev = None
1332
  if rdev is not None:
1333
    r_path = rdev.dev_path
1334
    try:
1335
      rdev.Remove()
1336
    except errors.BlockDeviceError, err:
1337
      msgs.append(str(err))
1338
    if not msgs:
1339
      DevCacheManager.RemoveCache(r_path)
1340

    
1341
  if disk.children:
1342
    for child in disk.children:
1343
      try:
1344
        BlockdevRemove(child)
1345
      except RPCFail, err:
1346
        msgs.append(str(err))
1347

    
1348
  if msgs:
1349
    _Fail("; ".join(msgs))
1350

    
1351

    
1352
def _RecursiveAssembleBD(disk, owner, as_primary):
1353
  """Activate a block device for an instance.
1354

1355
  This is run on the primary and secondary nodes for an instance.
1356

1357
  @note: this function is called recursively.
1358

1359
  @type disk: L{objects.Disk}
1360
  @param disk: the disk we try to assemble
1361
  @type owner: str
1362
  @param owner: the name of the instance which owns the disk
1363
  @type as_primary: boolean
1364
  @param as_primary: if we should make the block device
1365
      read/write
1366

1367
  @return: the assembled device or None (in case no device
1368
      was assembled)
1369
  @raise errors.BlockDeviceError: in case there is an error
1370
      during the activation of the children or the device
1371
      itself
1372

1373
  """
1374
  children = []
1375
  if disk.children:
1376
    mcn = disk.ChildrenNeeded()
1377
    if mcn == -1:
1378
      mcn = 0 # max number of Nones allowed
1379
    else:
1380
      mcn = len(disk.children) - mcn # max number of Nones
1381
    for chld_disk in disk.children:
1382
      try:
1383
        cdev = _RecursiveAssembleBD(chld_disk, owner, as_primary)
1384
      except errors.BlockDeviceError, err:
1385
        if children.count(None) >= mcn:
1386
          raise
1387
        cdev = None
1388
        logging.error("Error in child activation (but continuing): %s",
1389
                      str(err))
1390
      children.append(cdev)
1391

    
1392
  if as_primary or disk.AssembleOnSecondary():
1393
    r_dev = bdev.Assemble(disk.dev_type, disk.physical_id, children, disk.size)
1394
    r_dev.SetSyncSpeed(constants.SYNC_SPEED)
1395
    result = r_dev
1396
    if as_primary or disk.OpenOnSecondary():
1397
      r_dev.Open()
1398
    DevCacheManager.UpdateCache(r_dev.dev_path, owner,
1399
                                as_primary, disk.iv_name)
1400

    
1401
  else:
1402
    result = True
1403
  return result
1404

    
1405

    
1406
def BlockdevAssemble(disk, owner, as_primary):
1407
  """Activate a block device for an instance.
1408

1409
  This is a wrapper over _RecursiveAssembleBD.
1410

1411
  @rtype: str or boolean
1412
  @return: a C{/dev/...} path for primary nodes, and
1413
      C{True} for secondary nodes
1414

1415
  """
1416
  try:
1417
    result = _RecursiveAssembleBD(disk, owner, as_primary)
1418
    if isinstance(result, bdev.BlockDev):
1419
      # pylint: disable-msg=E1103
1420
      result = result.dev_path
1421
  except errors.BlockDeviceError, err:
1422
    _Fail("Error while assembling disk: %s", err, exc=True)
1423

    
1424
  return result
1425

    
1426

    
1427
def BlockdevShutdown(disk):
1428
  """Shut down a block device.
1429

1430
  First, if the device is assembled (Attach() is successful), then
1431
  the device is shutdown. Then the children of the device are
1432
  shutdown.
1433

1434
  This function is called recursively. Note that we don't cache the
1435
  children or such, as oppossed to assemble, shutdown of different
1436
  devices doesn't require that the upper device was active.
1437

1438
  @type disk: L{objects.Disk}
1439
  @param disk: the description of the disk we should
1440
      shutdown
1441
  @rtype: None
1442

1443
  """
1444
  msgs = []
1445
  r_dev = _RecursiveFindBD(disk)
1446
  if r_dev is not None:
1447
    r_path = r_dev.dev_path
1448
    try:
1449
      r_dev.Shutdown()
1450
      DevCacheManager.RemoveCache(r_path)
1451
    except errors.BlockDeviceError, err:
1452
      msgs.append(str(err))
1453

    
1454
  if disk.children:
1455
    for child in disk.children:
1456
      try:
1457
        BlockdevShutdown(child)
1458
      except RPCFail, err:
1459
        msgs.append(str(err))
1460

    
1461
  if msgs:
1462
    _Fail("; ".join(msgs))
1463

    
1464

    
1465
def BlockdevAddchildren(parent_cdev, new_cdevs):
1466
  """Extend a mirrored block device.
1467

1468
  @type parent_cdev: L{objects.Disk}
1469
  @param parent_cdev: the disk to which we should add children
1470
  @type new_cdevs: list of L{objects.Disk}
1471
  @param new_cdevs: the list of children which we should add
1472
  @rtype: None
1473

1474
  """
1475
  parent_bdev = _RecursiveFindBD(parent_cdev)
1476
  if parent_bdev is None:
1477
    _Fail("Can't find parent device '%s' in add children", parent_cdev)
1478
  new_bdevs = [_RecursiveFindBD(disk) for disk in new_cdevs]
1479
  if new_bdevs.count(None) > 0:
1480
    _Fail("Can't find new device(s) to add: %s:%s", new_bdevs, new_cdevs)
1481
  parent_bdev.AddChildren(new_bdevs)
1482

    
1483

    
1484
def BlockdevRemovechildren(parent_cdev, new_cdevs):
1485
  """Shrink a mirrored block device.
1486

1487
  @type parent_cdev: L{objects.Disk}
1488
  @param parent_cdev: the disk from which we should remove children
1489
  @type new_cdevs: list of L{objects.Disk}
1490
  @param new_cdevs: the list of children which we should remove
1491
  @rtype: None
1492

1493
  """
1494
  parent_bdev = _RecursiveFindBD(parent_cdev)
1495
  if parent_bdev is None:
1496
    _Fail("Can't find parent device '%s' in remove children", parent_cdev)
1497
  devs = []
1498
  for disk in new_cdevs:
1499
    rpath = disk.StaticDevPath()
1500
    if rpath is None:
1501
      bd = _RecursiveFindBD(disk)
1502
      if bd is None:
1503
        _Fail("Can't find device %s while removing children", disk)
1504
      else:
1505
        devs.append(bd.dev_path)
1506
    else:
1507
      if not utils.IsNormAbsPath(rpath):
1508
        _Fail("Strange path returned from StaticDevPath: '%s'", rpath)
1509
      devs.append(rpath)
1510
  parent_bdev.RemoveChildren(devs)
1511

    
1512

    
1513
def BlockdevGetmirrorstatus(disks):
1514
  """Get the mirroring status of a list of devices.
1515

1516
  @type disks: list of L{objects.Disk}
1517
  @param disks: the list of disks which we should query
1518
  @rtype: disk
1519
  @return:
1520
      a list of (mirror_done, estimated_time) tuples, which
1521
      are the result of L{bdev.BlockDev.CombinedSyncStatus}
1522
  @raise errors.BlockDeviceError: if any of the disks cannot be
1523
      found
1524

1525
  """
1526
  stats = []
1527
  for dsk in disks:
1528
    rbd = _RecursiveFindBD(dsk)
1529
    if rbd is None:
1530
      _Fail("Can't find device %s", dsk)
1531

    
1532
    stats.append(rbd.CombinedSyncStatus())
1533

    
1534
  return stats
1535

    
1536

    
1537
def _RecursiveFindBD(disk):
1538
  """Check if a device is activated.
1539

1540
  If so, return information about the real device.
1541

1542
  @type disk: L{objects.Disk}
1543
  @param disk: the disk object we need to find
1544

1545
  @return: None if the device can't be found,
1546
      otherwise the device instance
1547

1548
  """
1549
  children = []
1550
  if disk.children:
1551
    for chdisk in disk.children:
1552
      children.append(_RecursiveFindBD(chdisk))
1553

    
1554
  return bdev.FindDevice(disk.dev_type, disk.physical_id, children, disk.size)
1555

    
1556

    
1557
def _OpenRealBD(disk):
1558
  """Opens the underlying block device of a disk.
1559

1560
  @type disk: L{objects.Disk}
1561
  @param disk: the disk object we want to open
1562

1563
  """
1564
  real_disk = _RecursiveFindBD(disk)
1565
  if real_disk is None:
1566
    _Fail("Block device '%s' is not set up", disk)
1567

    
1568
  real_disk.Open()
1569

    
1570
  return real_disk
1571

    
1572

    
1573
def BlockdevFind(disk):
1574
  """Check if a device is activated.
1575

1576
  If it is, return information about the real device.
1577

1578
  @type disk: L{objects.Disk}
1579
  @param disk: the disk to find
1580
  @rtype: None or objects.BlockDevStatus
1581
  @return: None if the disk cannot be found, otherwise a the current
1582
           information
1583

1584
  """
1585
  try:
1586
    rbd = _RecursiveFindBD(disk)
1587
  except errors.BlockDeviceError, err:
1588
    _Fail("Failed to find device: %s", err, exc=True)
1589

    
1590
  if rbd is None:
1591
    return None
1592

    
1593
  return rbd.GetSyncStatus()
1594

    
1595

    
1596
def BlockdevGetsize(disks):
1597
  """Computes the size of the given disks.
1598

1599
  If a disk is not found, returns None instead.
1600

1601
  @type disks: list of L{objects.Disk}
1602
  @param disks: the list of disk to compute the size for
1603
  @rtype: list
1604
  @return: list with elements None if the disk cannot be found,
1605
      otherwise the size
1606

1607
  """
1608
  result = []
1609
  for cf in disks:
1610
    try:
1611
      rbd = _RecursiveFindBD(cf)
1612
    except errors.BlockDeviceError:
1613
      result.append(None)
1614
      continue
1615
    if rbd is None:
1616
      result.append(None)
1617
    else:
1618
      result.append(rbd.GetActualSize())
1619
  return result
1620

    
1621

    
1622
def BlockdevExport(disk, dest_node, dest_path, cluster_name):
1623
  """Export a block device to a remote node.
1624

1625
  @type disk: L{objects.Disk}
1626
  @param disk: the description of the disk to export
1627
  @type dest_node: str
1628
  @param dest_node: the destination node to export to
1629
  @type dest_path: str
1630
  @param dest_path: the destination path on the target node
1631
  @type cluster_name: str
1632
  @param cluster_name: the cluster name, needed for SSH hostalias
1633
  @rtype: None
1634

1635
  """
1636
  real_disk = _OpenRealBD(disk)
1637

    
1638
  # the block size on the read dd is 1MiB to match our units
1639
  expcmd = utils.BuildShellCmd("set -e; set -o pipefail; "
1640
                               "dd if=%s bs=1048576 count=%s",
1641
                               real_disk.dev_path, str(disk.size))
1642

    
1643
  # we set here a smaller block size as, due to ssh buffering, more
1644
  # than 64-128k will mostly ignored; we use nocreat to fail if the
1645
  # device is not already there or we pass a wrong path; we use
1646
  # notrunc to no attempt truncate on an LV device; we use oflag=dsync
1647
  # to not buffer too much memory; this means that at best, we flush
1648
  # every 64k, which will not be very fast
1649
  destcmd = utils.BuildShellCmd("dd of=%s conv=nocreat,notrunc bs=65536"
1650
                                " oflag=dsync", dest_path)
1651

    
1652
  remotecmd = _GetSshRunner(cluster_name).BuildCmd(dest_node,
1653
                                                   constants.GANETI_RUNAS,
1654
                                                   destcmd)
1655

    
1656
  # all commands have been checked, so we're safe to combine them
1657
  command = '|'.join([expcmd, utils.ShellQuoteArgs(remotecmd)])
1658

    
1659
  result = utils.RunCmd(["bash", "-c", command])
1660

    
1661
  if result.failed:
1662
    _Fail("Disk copy command '%s' returned error: %s"
1663
          " output: %s", command, result.fail_reason, result.output)
1664

    
1665

    
1666
def UploadFile(file_name, data, mode, uid, gid, atime, mtime):
1667
  """Write a file to the filesystem.
1668

1669
  This allows the master to overwrite(!) a file. It will only perform
1670
  the operation if the file belongs to a list of configuration files.
1671

1672
  @type file_name: str
1673
  @param file_name: the target file name
1674
  @type data: str
1675
  @param data: the new contents of the file
1676
  @type mode: int
1677
  @param mode: the mode to give the file (can be None)
1678
  @type uid: int
1679
  @param uid: the owner of the file (can be -1 for default)
1680
  @type gid: int
1681
  @param gid: the group of the file (can be -1 for default)
1682
  @type atime: float
1683
  @param atime: the atime to set on the file (can be None)
1684
  @type mtime: float
1685
  @param mtime: the mtime to set on the file (can be None)
1686
  @rtype: None
1687

1688
  """
1689
  if not os.path.isabs(file_name):
1690
    _Fail("Filename passed to UploadFile is not absolute: '%s'", file_name)
1691

    
1692
  if file_name not in _ALLOWED_UPLOAD_FILES:
1693
    _Fail("Filename passed to UploadFile not in allowed upload targets: '%s'",
1694
          file_name)
1695

    
1696
  raw_data = _Decompress(data)
1697

    
1698
  utils.WriteFile(file_name, data=raw_data, mode=mode, uid=uid, gid=gid,
1699
                  atime=atime, mtime=mtime)
1700

    
1701

    
1702
def WriteSsconfFiles(values):
1703
  """Update all ssconf files.
1704

1705
  Wrapper around the SimpleStore.WriteFiles.
1706

1707
  """
1708
  ssconf.SimpleStore().WriteFiles(values)
1709

    
1710

    
1711
def _ErrnoOrStr(err):
1712
  """Format an EnvironmentError exception.
1713

1714
  If the L{err} argument has an errno attribute, it will be looked up
1715
  and converted into a textual C{E...} description. Otherwise the
1716
  string representation of the error will be returned.
1717

1718
  @type err: L{EnvironmentError}
1719
  @param err: the exception to format
1720

1721
  """
1722
  if hasattr(err, 'errno'):
1723
    detail = errno.errorcode[err.errno]
1724
  else:
1725
    detail = str(err)
1726
  return detail
1727

    
1728

    
1729
def _OSOndiskAPIVersion(os_dir):
1730
  """Compute and return the API version of a given OS.
1731

1732
  This function will try to read the API version of the OS residing in
1733
  the 'os_dir' directory.
1734

1735
  @type os_dir: str
1736
  @param os_dir: the directory in which we should look for the OS
1737
  @rtype: tuple
1738
  @return: tuple (status, data) with status denoting the validity and
1739
      data holding either the vaid versions or an error message
1740

1741
  """
1742
  api_file = utils.PathJoin(os_dir, constants.OS_API_FILE)
1743

    
1744
  try:
1745
    st = os.stat(api_file)
1746
  except EnvironmentError, err:
1747
    return False, ("Required file '%s' not found under path %s: %s" %
1748
                   (constants.OS_API_FILE, os_dir, _ErrnoOrStr(err)))
1749

    
1750
  if not stat.S_ISREG(stat.S_IFMT(st.st_mode)):
1751
    return False, ("File '%s' in %s is not a regular file" %
1752
                   (constants.OS_API_FILE, os_dir))
1753

    
1754
  try:
1755
    api_versions = utils.ReadFile(api_file).splitlines()
1756
  except EnvironmentError, err:
1757
    return False, ("Error while reading the API version file at %s: %s" %
1758
                   (api_file, _ErrnoOrStr(err)))
1759

    
1760
  try:
1761
    api_versions = [int(version.strip()) for version in api_versions]
1762
  except (TypeError, ValueError), err:
1763
    return False, ("API version(s) can't be converted to integer: %s" %
1764
                   str(err))
1765

    
1766
  return True, api_versions
1767

    
1768

    
1769
def DiagnoseOS(top_dirs=None):
1770
  """Compute the validity for all OSes.
1771

1772
  @type top_dirs: list
1773
  @param top_dirs: the list of directories in which to
1774
      search (if not given defaults to
1775
      L{constants.OS_SEARCH_PATH})
1776
  @rtype: list of L{objects.OS}
1777
  @return: a list of tuples (name, path, status, diagnose, variants)
1778
      for all (potential) OSes under all search paths, where:
1779
          - name is the (potential) OS name
1780
          - path is the full path to the OS
1781
          - status True/False is the validity of the OS
1782
          - diagnose is the error message for an invalid OS, otherwise empty
1783
          - variants is a list of supported OS variants, if any
1784

1785
  """
1786
  if top_dirs is None:
1787
    top_dirs = constants.OS_SEARCH_PATH
1788

    
1789
  result = []
1790
  for dir_name in top_dirs:
1791
    if os.path.isdir(dir_name):
1792
      try:
1793
        f_names = utils.ListVisibleFiles(dir_name)
1794
      except EnvironmentError, err:
1795
        logging.exception("Can't list the OS directory %s: %s", dir_name, err)
1796
        break
1797
      for name in f_names:
1798
        os_path = utils.PathJoin(dir_name, name)
1799
        status, os_inst = _TryOSFromDisk(name, base_dir=dir_name)
1800
        if status:
1801
          diagnose = ""
1802
          variants = os_inst.supported_variants
1803
        else:
1804
          diagnose = os_inst
1805
          variants = []
1806
        result.append((name, os_path, status, diagnose, variants))
1807

    
1808
  return result
1809

    
1810

    
1811
def _TryOSFromDisk(name, base_dir=None):
1812
  """Create an OS instance from disk.
1813

1814
  This function will return an OS instance if the given name is a
1815
  valid OS name.
1816

1817
  @type base_dir: string
1818
  @keyword base_dir: Base directory containing OS installations.
1819
                     Defaults to a search in all the OS_SEARCH_PATH dirs.
1820
  @rtype: tuple
1821
  @return: success and either the OS instance if we find a valid one,
1822
      or error message
1823

1824
  """
1825
  if base_dir is None:
1826
    os_dir = utils.FindFile(name, constants.OS_SEARCH_PATH, os.path.isdir)
1827
  else:
1828
    os_dir = utils.FindFile(name, [base_dir], os.path.isdir)
1829

    
1830
  if os_dir is None:
1831
    return False, "Directory for OS %s not found in search path" % name
1832

    
1833
  status, api_versions = _OSOndiskAPIVersion(os_dir)
1834
  if not status:
1835
    # push the error up
1836
    return status, api_versions
1837

    
1838
  if not constants.OS_API_VERSIONS.intersection(api_versions):
1839
    return False, ("API version mismatch for path '%s': found %s, want %s." %
1840
                   (os_dir, api_versions, constants.OS_API_VERSIONS))
1841

    
1842
  # OS Files dictionary, we will populate it with the absolute path names
1843
  os_files = dict.fromkeys(constants.OS_SCRIPTS)
1844

    
1845
  if max(api_versions) >= constants.OS_API_V15:
1846
    os_files[constants.OS_VARIANTS_FILE] = ''
1847

    
1848
  for filename in os_files:
1849
    os_files[filename] = utils.PathJoin(os_dir, filename)
1850

    
1851
    try:
1852
      st = os.stat(os_files[filename])
1853
    except EnvironmentError, err:
1854
      return False, ("File '%s' under path '%s' is missing (%s)" %
1855
                     (filename, os_dir, _ErrnoOrStr(err)))
1856

    
1857
    if not stat.S_ISREG(stat.S_IFMT(st.st_mode)):
1858
      return False, ("File '%s' under path '%s' is not a regular file" %
1859
                     (filename, os_dir))
1860

    
1861
    if filename in constants.OS_SCRIPTS:
1862
      if stat.S_IMODE(st.st_mode) & stat.S_IXUSR != stat.S_IXUSR:
1863
        return False, ("File '%s' under path '%s' is not executable" %
1864
                       (filename, os_dir))
1865

    
1866
  variants = None
1867
  if constants.OS_VARIANTS_FILE in os_files:
1868
    variants_file = os_files[constants.OS_VARIANTS_FILE]
1869
    try:
1870
      variants = utils.ReadFile(variants_file).splitlines()
1871
    except EnvironmentError, err:
1872
      return False, ("Error while reading the OS variants file at %s: %s" %
1873
                     (variants_file, _ErrnoOrStr(err)))
1874
    if not variants:
1875
      return False, ("No supported os variant found")
1876

    
1877
  os_obj = objects.OS(name=name, path=os_dir,
1878
                      create_script=os_files[constants.OS_SCRIPT_CREATE],
1879
                      export_script=os_files[constants.OS_SCRIPT_EXPORT],
1880
                      import_script=os_files[constants.OS_SCRIPT_IMPORT],
1881
                      rename_script=os_files[constants.OS_SCRIPT_RENAME],
1882
                      supported_variants=variants,
1883
                      api_versions=api_versions)
1884
  return True, os_obj
1885

    
1886

    
1887
def OSFromDisk(name, base_dir=None):
1888
  """Create an OS instance from disk.
1889

1890
  This function will return an OS instance if the given name is a
1891
  valid OS name. Otherwise, it will raise an appropriate
1892
  L{RPCFail} exception, detailing why this is not a valid OS.
1893

1894
  This is just a wrapper over L{_TryOSFromDisk}, which doesn't raise
1895
  an exception but returns true/false status data.
1896

1897
  @type base_dir: string
1898
  @keyword base_dir: Base directory containing OS installations.
1899
                     Defaults to a search in all the OS_SEARCH_PATH dirs.
1900
  @rtype: L{objects.OS}
1901
  @return: the OS instance if we find a valid one
1902
  @raise RPCFail: if we don't find a valid OS
1903

1904
  """
1905
  name_only = name.split("+", 1)[0]
1906
  status, payload = _TryOSFromDisk(name_only, base_dir)
1907

    
1908
  if not status:
1909
    _Fail(payload)
1910

    
1911
  return payload
1912

    
1913

    
1914
def OSEnvironment(instance, inst_os, debug=0):
1915
  """Calculate the environment for an os script.
1916

1917
  @type instance: L{objects.Instance}
1918
  @param instance: target instance for the os script run
1919
  @type inst_os: L{objects.OS}
1920
  @param inst_os: operating system for which the environment is being built
1921
  @type debug: integer
1922
  @param debug: debug level (0 or 1, for OS Api 10)
1923
  @rtype: dict
1924
  @return: dict of environment variables
1925
  @raise errors.BlockDeviceError: if the block device
1926
      cannot be found
1927

1928
  """
1929
  result = {}
1930
  api_version = \
1931
    max(constants.OS_API_VERSIONS.intersection(inst_os.api_versions))
1932
  result['OS_API_VERSION'] = '%d' % api_version
1933
  result['INSTANCE_NAME'] = instance.name
1934
  result['INSTANCE_OS'] = instance.os
1935
  result['HYPERVISOR'] = instance.hypervisor
1936
  result['DISK_COUNT'] = '%d' % len(instance.disks)
1937
  result['NIC_COUNT'] = '%d' % len(instance.nics)
1938
  result['DEBUG_LEVEL'] = '%d' % debug
1939
  if api_version >= constants.OS_API_V15:
1940
    try:
1941
      variant = instance.os.split('+', 1)[1]
1942
    except IndexError:
1943
      variant = inst_os.supported_variants[0]
1944
    result['OS_VARIANT'] = variant
1945
  for idx, disk in enumerate(instance.disks):
1946
    real_disk = _OpenRealBD(disk)
1947
    result['DISK_%d_PATH' % idx] = real_disk.dev_path
1948
    result['DISK_%d_ACCESS' % idx] = disk.mode
1949
    if constants.HV_DISK_TYPE in instance.hvparams:
1950
      result['DISK_%d_FRONTEND_TYPE' % idx] = \
1951
        instance.hvparams[constants.HV_DISK_TYPE]
1952
    if disk.dev_type in constants.LDS_BLOCK:
1953
      result['DISK_%d_BACKEND_TYPE' % idx] = 'block'
1954
    elif disk.dev_type == constants.LD_FILE:
1955
      result['DISK_%d_BACKEND_TYPE' % idx] = \
1956
        'file:%s' % disk.physical_id[0]
1957
  for idx, nic in enumerate(instance.nics):
1958
    result['NIC_%d_MAC' % idx] = nic.mac
1959
    if nic.ip:
1960
      result['NIC_%d_IP' % idx] = nic.ip
1961
    result['NIC_%d_MODE' % idx] = nic.nicparams[constants.NIC_MODE]
1962
    if nic.nicparams[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED:
1963
      result['NIC_%d_BRIDGE' % idx] = nic.nicparams[constants.NIC_LINK]
1964
    if nic.nicparams[constants.NIC_LINK]:
1965
      result['NIC_%d_LINK' % idx] = nic.nicparams[constants.NIC_LINK]
1966
    if constants.HV_NIC_TYPE in instance.hvparams:
1967
      result['NIC_%d_FRONTEND_TYPE' % idx] = \
1968
        instance.hvparams[constants.HV_NIC_TYPE]
1969

    
1970
  for source, kind in [(instance.beparams, "BE"), (instance.hvparams, "HV")]:
1971
    for key, value in source.items():
1972
      result["INSTANCE_%s_%s" % (kind, key)] = str(value)
1973

    
1974
  return result
1975

    
1976

    
1977
def BlockdevGrow(disk, amount):
1978
  """Grow a stack of block devices.
1979

1980
  This function is called recursively, with the childrens being the
1981
  first ones to resize.
1982

1983
  @type disk: L{objects.Disk}
1984
  @param disk: the disk to be grown
1985
  @rtype: (status, result)
1986
  @return: a tuple with the status of the operation
1987
      (True/False), and the errors message if status
1988
      is False
1989

1990
  """
1991
  r_dev = _RecursiveFindBD(disk)
1992
  if r_dev is None:
1993
    _Fail("Cannot find block device %s", disk)
1994

    
1995
  try:
1996
    r_dev.Grow(amount)
1997
  except errors.BlockDeviceError, err:
1998
    _Fail("Failed to grow block device: %s", err, exc=True)
1999

    
2000

    
2001
def BlockdevSnapshot(disk):
2002
  """Create a snapshot copy of a block device.
2003

2004
  This function is called recursively, and the snapshot is actually created
2005
  just for the leaf lvm backend device.
2006

2007
  @type disk: L{objects.Disk}
2008
  @param disk: the disk to be snapshotted
2009
  @rtype: string
2010
  @return: snapshot disk path
2011

2012
  """
2013
  if disk.dev_type == constants.LD_DRBD8:
2014
    if not disk.children:
2015
      _Fail("DRBD device '%s' without backing storage cannot be snapshotted",
2016
            disk.unique_id)
2017
    return BlockdevSnapshot(disk.children[0])
2018
  elif disk.dev_type == constants.LD_LV:
2019
    r_dev = _RecursiveFindBD(disk)
2020
    if r_dev is not None:
2021
      # FIXME: choose a saner value for the snapshot size
2022
      # let's stay on the safe side and ask for the full size, for now
2023
      return r_dev.Snapshot(disk.size)
2024
    else:
2025
      _Fail("Cannot find block device %s", disk)
2026
  else:
2027
    _Fail("Cannot snapshot non-lvm block device '%s' of type '%s'",
2028
          disk.unique_id, disk.dev_type)
2029

    
2030

    
2031
def FinalizeExport(instance, snap_disks):
2032
  """Write out the export configuration information.
2033

2034
  @type instance: L{objects.Instance}
2035
  @param instance: the instance which we export, used for
2036
      saving configuration
2037
  @type snap_disks: list of L{objects.Disk}
2038
  @param snap_disks: list of snapshot block devices, which
2039
      will be used to get the actual name of the dump file
2040

2041
  @rtype: None
2042

2043
  """
2044
  destdir = utils.PathJoin(constants.EXPORT_DIR, instance.name + ".new")
2045
  finaldestdir = utils.PathJoin(constants.EXPORT_DIR, instance.name)
2046

    
2047
  config = objects.SerializableConfigParser()
2048

    
2049
  config.add_section(constants.INISECT_EXP)
2050
  config.set(constants.INISECT_EXP, 'version', '0')
2051
  config.set(constants.INISECT_EXP, 'timestamp', '%d' % int(time.time()))
2052
  config.set(constants.INISECT_EXP, 'source', instance.primary_node)
2053
  config.set(constants.INISECT_EXP, 'os', instance.os)
2054
  config.set(constants.INISECT_EXP, 'compression', 'gzip')
2055

    
2056
  config.add_section(constants.INISECT_INS)
2057
  config.set(constants.INISECT_INS, 'name', instance.name)
2058
  config.set(constants.INISECT_INS, 'memory', '%d' %
2059
             instance.beparams[constants.BE_MEMORY])
2060
  config.set(constants.INISECT_INS, 'vcpus', '%d' %
2061
             instance.beparams[constants.BE_VCPUS])
2062
  config.set(constants.INISECT_INS, 'disk_template', instance.disk_template)
2063
  config.set(constants.INISECT_INS, 'hypervisor', instance.hypervisor)
2064

    
2065
  nic_total = 0
2066
  for nic_count, nic in enumerate(instance.nics):
2067
    nic_total += 1
2068
    config.set(constants.INISECT_INS, 'nic%d_mac' %
2069
               nic_count, '%s' % nic.mac)
2070
    config.set(constants.INISECT_INS, 'nic%d_ip' % nic_count, '%s' % nic.ip)
2071
    for param in constants.NICS_PARAMETER_TYPES:
2072
      config.set(constants.INISECT_INS, 'nic%d_%s' % (nic_count, param),
2073
                 '%s' % nic.nicparams.get(param, None))
2074
  # TODO: redundant: on load can read nics until it doesn't exist
2075
  config.set(constants.INISECT_INS, 'nic_count' , '%d' % nic_total)
2076

    
2077
  disk_total = 0
2078
  for disk_count, disk in enumerate(snap_disks):
2079
    if disk:
2080
      disk_total += 1
2081
      config.set(constants.INISECT_INS, 'disk%d_ivname' % disk_count,
2082
                 ('%s' % disk.iv_name))
2083
      config.set(constants.INISECT_INS, 'disk%d_dump' % disk_count,
2084
                 ('%s' % disk.physical_id[1]))
2085
      config.set(constants.INISECT_INS, 'disk%d_size' % disk_count,
2086
                 ('%d' % disk.size))
2087

    
2088
  config.set(constants.INISECT_INS, 'disk_count' , '%d' % disk_total)
2089

    
2090
  # New-style hypervisor/backend parameters
2091

    
2092
  config.add_section(constants.INISECT_HYP)
2093
  for name, value in instance.hvparams.items():
2094
    if name not in constants.HVC_GLOBALS:
2095
      config.set(constants.INISECT_HYP, name, str(value))
2096

    
2097
  config.add_section(constants.INISECT_BEP)
2098
  for name, value in instance.beparams.items():
2099
    config.set(constants.INISECT_BEP, name, str(value))
2100

    
2101
  utils.WriteFile(utils.PathJoin(destdir, constants.EXPORT_CONF_FILE),
2102
                  data=config.Dumps())
2103
  shutil.rmtree(finaldestdir, ignore_errors=True)
2104
  shutil.move(destdir, finaldestdir)
2105

    
2106

    
2107
def ExportInfo(dest):
2108
  """Get export configuration information.
2109

2110
  @type dest: str
2111
  @param dest: directory containing the export
2112

2113
  @rtype: L{objects.SerializableConfigParser}
2114
  @return: a serializable config file containing the
2115
      export info
2116

2117
  """
2118
  cff = utils.PathJoin(dest, constants.EXPORT_CONF_FILE)
2119

    
2120
  config = objects.SerializableConfigParser()
2121
  config.read(cff)
2122

    
2123
  if (not config.has_section(constants.INISECT_EXP) or
2124
      not config.has_section(constants.INISECT_INS)):
2125
    _Fail("Export info file doesn't have the required fields")
2126

    
2127
  return config.Dumps()
2128

    
2129

    
2130
def ListExports():
2131
  """Return a list of exports currently available on this machine.
2132

2133
  @rtype: list
2134
  @return: list of the exports
2135

2136
  """
2137
  if os.path.isdir(constants.EXPORT_DIR):
2138
    return utils.ListVisibleFiles(constants.EXPORT_DIR)
2139
  else:
2140
    _Fail("No exports directory")
2141

    
2142

    
2143
def RemoveExport(export):
2144
  """Remove an existing export from the node.
2145

2146
  @type export: str
2147
  @param export: the name of the export to remove
2148
  @rtype: None
2149

2150
  """
2151
  target = utils.PathJoin(constants.EXPORT_DIR, export)
2152

    
2153
  try:
2154
    shutil.rmtree(target)
2155
  except EnvironmentError, err:
2156
    _Fail("Error while removing the export: %s", err, exc=True)
2157

    
2158

    
2159
def BlockdevRename(devlist):
2160
  """Rename a list of block devices.
2161

2162
  @type devlist: list of tuples
2163
  @param devlist: list of tuples of the form  (disk,
2164
      new_logical_id, new_physical_id); disk is an
2165
      L{objects.Disk} object describing the current disk,
2166
      and new logical_id/physical_id is the name we
2167
      rename it to
2168
  @rtype: boolean
2169
  @return: True if all renames succeeded, False otherwise
2170

2171
  """
2172
  msgs = []
2173
  result = True
2174
  for disk, unique_id in devlist:
2175
    dev = _RecursiveFindBD(disk)
2176
    if dev is None:
2177
      msgs.append("Can't find device %s in rename" % str(disk))
2178
      result = False
2179
      continue
2180
    try:
2181
      old_rpath = dev.dev_path
2182
      dev.Rename(unique_id)
2183
      new_rpath = dev.dev_path
2184
      if old_rpath != new_rpath:
2185
        DevCacheManager.RemoveCache(old_rpath)
2186
        # FIXME: we should add the new cache information here, like:
2187
        # DevCacheManager.UpdateCache(new_rpath, owner, ...)
2188
        # but we don't have the owner here - maybe parse from existing
2189
        # cache? for now, we only lose lvm data when we rename, which
2190
        # is less critical than DRBD or MD
2191
    except errors.BlockDeviceError, err:
2192
      msgs.append("Can't rename device '%s' to '%s': %s" %
2193
                  (dev, unique_id, err))
2194
      logging.exception("Can't rename device '%s' to '%s'", dev, unique_id)
2195
      result = False
2196
  if not result:
2197
    _Fail("; ".join(msgs))
2198

    
2199

    
2200
def _TransformFileStorageDir(file_storage_dir):
2201
  """Checks whether given file_storage_dir is valid.
2202

2203
  Checks wheter the given file_storage_dir is within the cluster-wide
2204
  default file_storage_dir stored in SimpleStore. Only paths under that
2205
  directory are allowed.
2206

2207
  @type file_storage_dir: str
2208
  @param file_storage_dir: the path to check
2209

2210
  @return: the normalized path if valid, None otherwise
2211

2212
  """
2213
  if not constants.ENABLE_FILE_STORAGE:
2214
    _Fail("File storage disabled at configure time")
2215
  cfg = _GetConfig()
2216
  file_storage_dir = os.path.normpath(file_storage_dir)
2217
  base_file_storage_dir = cfg.GetFileStorageDir()
2218
  if (os.path.commonprefix([file_storage_dir, base_file_storage_dir]) !=
2219
      base_file_storage_dir):
2220
    _Fail("File storage directory '%s' is not under base file"
2221
          " storage directory '%s'", file_storage_dir, base_file_storage_dir)
2222
  return file_storage_dir
2223

    
2224

    
2225
def CreateFileStorageDir(file_storage_dir):
2226
  """Create file storage directory.
2227

2228
  @type file_storage_dir: str
2229
  @param file_storage_dir: directory to create
2230

2231
  @rtype: tuple
2232
  @return: tuple with first element a boolean indicating wheter dir
2233
      creation was successful or not
2234

2235
  """
2236
  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
2237
  if os.path.exists(file_storage_dir):
2238
    if not os.path.isdir(file_storage_dir):
2239
      _Fail("Specified storage dir '%s' is not a directory",
2240
            file_storage_dir)
2241
  else:
2242
    try:
2243
      os.makedirs(file_storage_dir, 0750)
2244
    except OSError, err:
2245
      _Fail("Cannot create file storage directory '%s': %s",
2246
            file_storage_dir, err, exc=True)
2247

    
2248

    
2249
def RemoveFileStorageDir(file_storage_dir):
2250
  """Remove file storage directory.
2251

2252
  Remove it only if it's empty. If not log an error and return.
2253

2254
  @type file_storage_dir: str
2255
  @param file_storage_dir: the directory we should cleanup
2256
  @rtype: tuple (success,)
2257
  @return: tuple of one element, C{success}, denoting
2258
      whether the operation was successful
2259

2260
  """
2261
  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
2262
  if os.path.exists(file_storage_dir):
2263
    if not os.path.isdir(file_storage_dir):
2264
      _Fail("Specified Storage directory '%s' is not a directory",
2265
            file_storage_dir)
2266
    # deletes dir only if empty, otherwise we want to fail the rpc call
2267
    try:
2268
      os.rmdir(file_storage_dir)
2269
    except OSError, err:
2270
      _Fail("Cannot remove file storage directory '%s': %s",
2271
            file_storage_dir, err)
2272

    
2273

    
2274
def RenameFileStorageDir(old_file_storage_dir, new_file_storage_dir):
2275
  """Rename the file storage directory.
2276

2277
  @type old_file_storage_dir: str
2278
  @param old_file_storage_dir: the current path
2279
  @type new_file_storage_dir: str
2280
  @param new_file_storage_dir: the name we should rename to
2281
  @rtype: tuple (success,)
2282
  @return: tuple of one element, C{success}, denoting
2283
      whether the operation was successful
2284

2285
  """
2286
  old_file_storage_dir = _TransformFileStorageDir(old_file_storage_dir)
2287
  new_file_storage_dir = _TransformFileStorageDir(new_file_storage_dir)
2288
  if not os.path.exists(new_file_storage_dir):
2289
    if os.path.isdir(old_file_storage_dir):
2290
      try:
2291
        os.rename(old_file_storage_dir, new_file_storage_dir)
2292
      except OSError, err:
2293
        _Fail("Cannot rename '%s' to '%s': %s",
2294
              old_file_storage_dir, new_file_storage_dir, err)
2295
    else:
2296
      _Fail("Specified storage dir '%s' is not a directory",
2297
            old_file_storage_dir)
2298
  else:
2299
    if os.path.exists(old_file_storage_dir):
2300
      _Fail("Cannot rename '%s' to '%s': both locations exist",
2301
            old_file_storage_dir, new_file_storage_dir)
2302

    
2303

    
2304
def _EnsureJobQueueFile(file_name):
2305
  """Checks whether the given filename is in the queue directory.
2306

2307
  @type file_name: str
2308
  @param file_name: the file name we should check
2309
  @rtype: None
2310
  @raises RPCFail: if the file is not valid
2311

2312
  """
2313
  queue_dir = os.path.normpath(constants.QUEUE_DIR)
2314
  result = (os.path.commonprefix([queue_dir, file_name]) == queue_dir)
2315

    
2316
  if not result:
2317
    _Fail("Passed job queue file '%s' does not belong to"
2318
          " the queue directory '%s'", file_name, queue_dir)
2319

    
2320

    
2321
def JobQueueUpdate(file_name, content):
2322
  """Updates a file in the queue directory.
2323

2324
  This is just a wrapper over L{utils.WriteFile}, with proper
2325
  checking.
2326

2327
  @type file_name: str
2328
  @param file_name: the job file name
2329
  @type content: str
2330
  @param content: the new job contents
2331
  @rtype: boolean
2332
  @return: the success of the operation
2333

2334
  """
2335
  _EnsureJobQueueFile(file_name)
2336

    
2337
  # Write and replace the file atomically
2338
  utils.WriteFile(file_name, data=_Decompress(content))
2339

    
2340

    
2341
def JobQueueRename(old, new):
2342
  """Renames a job queue file.
2343

2344
  This is just a wrapper over os.rename with proper checking.
2345

2346
  @type old: str
2347
  @param old: the old (actual) file name
2348
  @type new: str
2349
  @param new: the desired file name
2350
  @rtype: tuple
2351
  @return: the success of the operation and payload
2352

2353
  """
2354
  _EnsureJobQueueFile(old)
2355
  _EnsureJobQueueFile(new)
2356

    
2357
  utils.RenameFile(old, new, mkdir=True)
2358

    
2359

    
2360
def JobQueueSetDrainFlag(drain_flag):
2361
  """Set the drain flag for the queue.
2362

2363
  This will set or unset the queue drain flag.
2364

2365
  @type drain_flag: boolean
2366
  @param drain_flag: if True, will set the drain flag, otherwise reset it.
2367
  @rtype: truple
2368
  @return: always True, None
2369
  @warning: the function always returns True
2370

2371
  """
2372
  if drain_flag:
2373
    utils.WriteFile(constants.JOB_QUEUE_DRAIN_FILE, data="", close=True)
2374
  else:
2375
    utils.RemoveFile(constants.JOB_QUEUE_DRAIN_FILE)
2376

    
2377

    
2378
def BlockdevClose(instance_name, disks):
2379
  """Closes the given block devices.
2380

2381
  This means they will be switched to secondary mode (in case of
2382
  DRBD).
2383

2384
  @param instance_name: if the argument is not empty, the symlinks
2385
      of this instance will be removed
2386
  @type disks: list of L{objects.Disk}
2387
  @param disks: the list of disks to be closed
2388
  @rtype: tuple (success, message)
2389
  @return: a tuple of success and message, where success
2390
      indicates the succes of the operation, and message
2391
      which will contain the error details in case we
2392
      failed
2393

2394
  """
2395
  bdevs = []
2396
  for cf in disks:
2397
    rd = _RecursiveFindBD(cf)
2398
    if rd is None:
2399
      _Fail("Can't find device %s", cf)
2400
    bdevs.append(rd)
2401

    
2402
  msg = []
2403
  for rd in bdevs:
2404
    try:
2405
      rd.Close()
2406
    except errors.BlockDeviceError, err:
2407
      msg.append(str(err))
2408
  if msg:
2409
    _Fail("Can't make devices secondary: %s", ",".join(msg))
2410
  else:
2411
    if instance_name:
2412
      _RemoveBlockDevLinks(instance_name, disks)
2413

    
2414

    
2415
def ValidateHVParams(hvname, hvparams):
2416
  """Validates the given hypervisor parameters.
2417

2418
  @type hvname: string
2419
  @param hvname: the hypervisor name
2420
  @type hvparams: dict
2421
  @param hvparams: the hypervisor parameters to be validated
2422
  @rtype: None
2423

2424
  """
2425
  try:
2426
    hv_type = hypervisor.GetHypervisor(hvname)
2427
    hv_type.ValidateParameters(hvparams)
2428
  except errors.HypervisorError, err:
2429
    _Fail(str(err), log=False)
2430

    
2431

    
2432
def DemoteFromMC():
2433
  """Demotes the current node from master candidate role.
2434

2435
  """
2436
  # try to ensure we're not the master by mistake
2437
  master, myself = ssconf.GetMasterAndMyself()
2438
  if master == myself:
2439
    _Fail("ssconf status shows I'm the master node, will not demote")
2440

    
2441
  result = utils.RunCmd([constants.DAEMON_UTIL, "check", constants.MASTERD])
2442
  if not result.failed:
2443
    _Fail("The master daemon is running, will not demote")
2444

    
2445
  try:
2446
    if os.path.isfile(constants.CLUSTER_CONF_FILE):
2447
      utils.CreateBackup(constants.CLUSTER_CONF_FILE)
2448
  except EnvironmentError, err:
2449
    if err.errno != errno.ENOENT:
2450
      _Fail("Error while backing up cluster file: %s", err, exc=True)
2451

    
2452
  utils.RemoveFile(constants.CLUSTER_CONF_FILE)
2453

    
2454

    
2455
def _GetX509Filenames(cryptodir, name):
2456
  """Returns the full paths for the private key and certificate.
2457

2458
  """
2459
  return (utils.PathJoin(cryptodir, name),
2460
          utils.PathJoin(cryptodir, name, _X509_KEY_FILE),
2461
          utils.PathJoin(cryptodir, name, _X509_CERT_FILE))
2462

    
2463

    
2464
def CreateX509Certificate(validity, cryptodir=constants.CRYPTO_KEYS_DIR):
2465
  """Creates a new X509 certificate for SSL/TLS.
2466

2467
  @type validity: int
2468
  @param validity: Validity in seconds
2469
  @rtype: tuple; (string, string)
2470
  @return: Certificate name and public part
2471

2472
  """
2473
  (key_pem, cert_pem) = \
2474
    utils.GenerateSelfSignedX509Cert(utils.HostInfo.SysName(),
2475
                                     min(validity, _MAX_SSL_CERT_VALIDITY))
2476

    
2477
  cert_dir = tempfile.mkdtemp(dir=cryptodir,
2478
                              prefix="x509-%s-" % utils.TimestampForFilename())
2479
  try:
2480
    name = os.path.basename(cert_dir)
2481
    assert len(name) > 5
2482

    
2483
    (_, key_file, cert_file) = _GetX509Filenames(cryptodir, name)
2484

    
2485
    utils.WriteFile(key_file, mode=0400, data=key_pem)
2486
    utils.WriteFile(cert_file, mode=0400, data=cert_pem)
2487

    
2488
    # Never return private key as it shouldn't leave the node
2489
    return (name, cert_pem)
2490
  except Exception:
2491
    shutil.rmtree(cert_dir, ignore_errors=True)
2492
    raise
2493

    
2494

    
2495
def RemoveX509Certificate(name, cryptodir=constants.CRYPTO_KEYS_DIR):
2496
  """Removes a X509 certificate.
2497

2498
  @type name: string
2499
  @param name: Certificate name
2500

2501
  """
2502
  (cert_dir, key_file, cert_file) = _GetX509Filenames(cryptodir, name)
2503

    
2504
  utils.RemoveFile(key_file)
2505
  utils.RemoveFile(cert_file)
2506

    
2507
  try:
2508
    os.rmdir(cert_dir)
2509
  except EnvironmentError, err:
2510
    _Fail("Cannot remove certificate directory '%s': %s",
2511
          cert_dir, err)
2512

    
2513

    
2514
def _GetImportExportIoCommand(instance, mode, ieio, ieargs):
2515
  """Returns the command for the requested input/output.
2516

2517
  @type instance: L{objects.Instance}
2518
  @param instance: The instance object
2519
  @param mode: Import/export mode
2520
  @param ieio: Input/output type
2521
  @param ieargs: Input/output arguments
2522

2523
  """
2524
  assert mode in (constants.IEM_IMPORT, constants.IEM_EXPORT)
2525

    
2526
  env = None
2527
  prefix = None
2528
  suffix = None
2529

    
2530
  if ieio == constants.IEIO_FILE:
2531
    (filename, ) = ieargs
2532

    
2533
    if not utils.IsNormAbsPath(filename):
2534
      _Fail("Path '%s' is not normalized or absolute", filename)
2535

    
2536
    directory = os.path.normpath(os.path.dirname(filename))
2537

    
2538
    if (os.path.commonprefix([constants.EXPORT_DIR, directory]) !=
2539
        constants.EXPORT_DIR):
2540
      _Fail("File '%s' is not under exports directory '%s'",
2541
            filename, constants.EXPORT_DIR)
2542

    
2543
    # Create directory
2544
    utils.Makedirs(directory, mode=0750)
2545

    
2546
    quoted_filename = utils.ShellQuote(filename)
2547

    
2548
    if mode == constants.IEM_IMPORT:
2549
      suffix = "> %s" % quoted_filename
2550
    elif mode == constants.IEM_EXPORT:
2551
      suffix = "< %s" % quoted_filename
2552

    
2553
  elif ieio == constants.IEIO_RAW_DISK:
2554
    (disk, ) = ieargs
2555

    
2556
    real_disk = _OpenRealBD(disk)
2557

    
2558
    if mode == constants.IEM_IMPORT:
2559
      # we set here a smaller block size as, due to transport buffering, more
2560
      # than 64-128k will mostly ignored; we use nocreat to fail if the device
2561
      # is not already there or we pass a wrong path; we use notrunc to no
2562
      # attempt truncate on an LV device; we use oflag=dsync to not buffer too
2563
      # much memory; this means that at best, we flush every 64k, which will
2564
      # not be very fast
2565
      suffix = utils.BuildShellCmd(("| dd of=%s conv=nocreat,notrunc"
2566
                                    " bs=%s oflag=dsync"),
2567
                                    real_disk.dev_path,
2568
                                    str(64 * 1024))
2569

    
2570
    elif mode == constants.IEM_EXPORT:
2571
      # the block size on the read dd is 1MiB to match our units
2572
      prefix = utils.BuildShellCmd("dd if=%s bs=%s count=%s |",
2573
                                   real_disk.dev_path,
2574
                                   str(1024 * 1024), # 1 MB
2575
                                   str(disk.size))
2576

    
2577
  elif ieio == constants.IEIO_SCRIPT:
2578
    (disk, disk_index, ) = ieargs
2579

    
2580
    assert isinstance(disk_index, (int, long))
2581

    
2582
    real_disk = _OpenRealBD(disk)
2583

    
2584
    inst_os = OSFromDisk(instance.os)
2585
    env = OSEnvironment(instance, inst_os)
2586

    
2587
    if mode == constants.IEM_IMPORT:
2588
      env["IMPORT_DEVICE"] = env["DISK_%d_PATH" % disk_index]
2589
      env["IMPORT_INDEX"] = str(disk_index)
2590
      script = inst_os.import_script
2591

    
2592
    elif mode == constants.IEM_EXPORT:
2593
      env["EXPORT_DEVICE"] = real_disk.dev_path
2594
      env["EXPORT_INDEX"] = str(disk_index)
2595
      script = inst_os.export_script
2596

    
2597
    # TODO: Pass special environment only to script
2598
    script_cmd = utils.BuildShellCmd("( cd %s && %s; )", inst_os.path, script)
2599

    
2600
    if mode == constants.IEM_IMPORT:
2601
      suffix = "| %s" % script_cmd
2602

    
2603
    elif mode == constants.IEM_EXPORT:
2604
      prefix = "%s |" % script_cmd
2605

    
2606
  else:
2607
    _Fail("Invalid %s I/O mode %r", mode, ieio)
2608

    
2609
  return (env, prefix, suffix)
2610

    
2611

    
2612
def _CreateImportExportStatusDir(prefix):
2613
  """Creates status directory for import/export.
2614

2615
  """
2616
  return tempfile.mkdtemp(dir=constants.IMPORT_EXPORT_DIR,
2617
                          prefix=("%s-%s-" %
2618
                                  (prefix, utils.TimestampForFilename())))
2619

    
2620

    
2621
def StartImportExportDaemon(mode, opts, host, port, instance, ieio, ieioargs):
2622
  """Starts an import or export daemon.
2623

2624
  @param mode: Import/output mode
2625
  @type opts: L{objects.ImportExportOptions}
2626
  @param opts: Daemon options
2627
  @type host: string
2628
  @param host: Remote host for export (None for import)
2629
  @type port: int
2630
  @param port: Remote port for export (None for import)
2631
  @type instance: L{objects.Instance}
2632
  @param instance: Instance object
2633
  @param ieio: Input/output type
2634
  @param ieioargs: Input/output arguments
2635

2636
  """
2637
  if mode == constants.IEM_IMPORT:
2638
    prefix = "import"
2639

    
2640
    if not (host is None and port is None):
2641
      _Fail("Can not specify host or port on import")
2642

    
2643
  elif mode == constants.IEM_EXPORT:
2644
    prefix = "export"
2645

    
2646
    if host is None or port is None:
2647
      _Fail("Host and port must be specified for an export")
2648

    
2649
  else:
2650
    _Fail("Invalid mode %r", mode)
2651

    
2652
  if (opts.key_name is None) ^ (opts.ca_pem is None):
2653
    _Fail("Cluster certificate can only be used for both key and CA")
2654

    
2655
  (cmd_env, cmd_prefix, cmd_suffix) = \
2656
    _GetImportExportIoCommand(instance, mode, ieio, ieioargs)
2657

    
2658
  if opts.key_name is None:
2659
    # Use server.pem
2660
    key_path = constants.NODED_CERT_FILE
2661
    cert_path = constants.NODED_CERT_FILE
2662
    assert opts.ca_pem is None
2663
  else:
2664
    (_, key_path, cert_path) = _GetX509Filenames(constants.CRYPTO_KEYS_DIR,
2665
                                                 opts.key_name)
2666
    assert opts.ca_pem is not None
2667

    
2668
  for i in [key_path, cert_path]:
2669
    if not os.path.exists(i):
2670
      _Fail("File '%s' does not exist" % i)
2671

    
2672
  status_dir = _CreateImportExportStatusDir(prefix)
2673
  try:
2674
    status_file = utils.PathJoin(status_dir, _IES_STATUS_FILE)
2675
    pid_file = utils.PathJoin(status_dir, _IES_PID_FILE)
2676
    ca_file = utils.PathJoin(status_dir, _IES_CA_FILE)
2677

    
2678
    if opts.ca_pem is None:
2679
      # Use server.pem
2680
      ca = utils.ReadFile(constants.NODED_CERT_FILE)
2681
    else:
2682
      ca = opts.ca_pem
2683

    
2684
    # Write CA file
2685
    utils.WriteFile(ca_file, data=ca, mode=0400)
2686

    
2687
    cmd = [
2688
      constants.IMPORT_EXPORT_DAEMON,
2689
      status_file, mode,
2690
      "--key=%s" % key_path,
2691
      "--cert=%s" % cert_path,
2692
      "--ca=%s" % ca_file,
2693
      ]
2694

    
2695
    if host:
2696
      cmd.append("--host=%s" % host)
2697

    
2698
    if port:
2699
      cmd.append("--port=%s" % port)
2700

    
2701
    if opts.compress:
2702
      cmd.append("--compress=%s" % opts.compress)
2703

    
2704
    if cmd_prefix:
2705
      cmd.append("--cmd-prefix=%s" % cmd_prefix)
2706

    
2707
    if cmd_suffix:
2708
      cmd.append("--cmd-suffix=%s" % cmd_suffix)
2709

    
2710
    logfile = _InstanceLogName(prefix, instance.os, instance.name)
2711

    
2712
    # TODO: Once _InstanceLogName uses tempfile.mkstemp, StartDaemon has
2713
    # support for receiving a file descriptor for output
2714
    utils.StartDaemon(cmd, env=cmd_env, pidfile=pid_file,
2715
                      output=logfile)
2716

    
2717
    # The import/export name is simply the status directory name
2718
    return os.path.basename(status_dir)
2719

    
2720
  except Exception:
2721
    shutil.rmtree(status_dir, ignore_errors=True)
2722
    raise
2723

    
2724

    
2725
def GetImportExportStatus(names):
2726
  """Returns import/export daemon status.
2727

2728
  @type names: sequence
2729
  @param names: List of names
2730
  @rtype: List of dicts
2731
  @return: Returns a list of the state of each named import/export or None if a
2732
           status couldn't be read
2733

2734
  """
2735
  result = []
2736

    
2737
  for name in names:
2738
    status_file = utils.PathJoin(constants.IMPORT_EXPORT_DIR, name,
2739
                                 _IES_STATUS_FILE)
2740

    
2741
    try:
2742
      data = utils.ReadFile(status_file)
2743
    except EnvironmentError, err:
2744
      if err.errno != errno.ENOENT:
2745
        raise
2746
      data = None
2747

    
2748
    if not data:
2749
      result.append(None)
2750
      continue
2751

    
2752
    result.append(serializer.LoadJson(data))
2753

    
2754
  return result
2755

    
2756

    
2757
def AbortImportExport(name):
2758
  """Sends SIGTERM to a running import/export daemon.
2759

2760
  """
2761
  logging.info("Abort import/export %s", name)
2762

    
2763
  status_dir = utils.PathJoin(constants.IMPORT_EXPORT_DIR, name)
2764
  pid = utils.ReadLockedPidFile(utils.PathJoin(status_dir, _IES_PID_FILE))
2765

    
2766
  if pid:
2767
    logging.info("Import/export %s is running with PID %s, sending SIGTERM",
2768
                 name, pid)
2769
    os.kill(pid, signal.SIGTERM)
2770

    
2771

    
2772
def CleanupImportExport(name):
2773
  """Cleanup after an import or export.
2774

2775
  If the import/export daemon is still running it's killed. Afterwards the
2776
  whole status directory is removed.
2777

2778
  """
2779
  logging.info("Finalizing import/export %s", name)
2780

    
2781
  status_dir = utils.PathJoin(constants.IMPORT_EXPORT_DIR, name)
2782

    
2783
  pid = utils.ReadLockedPidFile(utils.PathJoin(status_dir, _IES_PID_FILE))
2784

    
2785
  if pid:
2786
    logging.info("Import/export %s is still running with PID %s",
2787
                 name, pid)
2788
    utils.KillProcess(pid, waitpid=False)
2789

    
2790
  shutil.rmtree(status_dir, ignore_errors=True)
2791

    
2792

    
2793
def _FindDisks(nodes_ip, disks):
2794
  """Sets the physical ID on disks and returns the block devices.
2795

2796
  """
2797
  # set the correct physical ID
2798
  my_name = utils.HostInfo().name
2799
  for cf in disks:
2800
    cf.SetPhysicalID(my_name, nodes_ip)
2801

    
2802
  bdevs = []
2803

    
2804
  for cf in disks:
2805
    rd = _RecursiveFindBD(cf)
2806
    if rd is None:
2807
      _Fail("Can't find device %s", cf)
2808
    bdevs.append(rd)
2809
  return bdevs
2810

    
2811

    
2812
def DrbdDisconnectNet(nodes_ip, disks):
2813
  """Disconnects the network on a list of drbd devices.
2814

2815
  """
2816
  bdevs = _FindDisks(nodes_ip, disks)
2817

    
2818
  # disconnect disks
2819
  for rd in bdevs:
2820
    try:
2821
      rd.DisconnectNet()
2822
    except errors.BlockDeviceError, err:
2823
      _Fail("Can't change network configuration to standalone mode: %s",
2824
            err, exc=True)
2825

    
2826

    
2827
def DrbdAttachNet(nodes_ip, disks, instance_name, multimaster):
2828
  """Attaches the network on a list of drbd devices.
2829

2830
  """
2831
  bdevs = _FindDisks(nodes_ip, disks)
2832

    
2833
  if multimaster:
2834
    for idx, rd in enumerate(bdevs):
2835
      try:
2836
        _SymlinkBlockDev(instance_name, rd.dev_path, idx)
2837
      except EnvironmentError, err:
2838
        _Fail("Can't create symlink: %s", err)
2839
  # reconnect disks, switch to new master configuration and if
2840
  # needed primary mode
2841
  for rd in bdevs:
2842
    try:
2843
      rd.AttachNet(multimaster)
2844
    except errors.BlockDeviceError, err:
2845
      _Fail("Can't change network configuration: %s", err)
2846

    
2847
  # wait until the disks are connected; we need to retry the re-attach
2848
  # if the device becomes standalone, as this might happen if the one
2849
  # node disconnects and reconnects in a different mode before the
2850
  # other node reconnects; in this case, one or both of the nodes will
2851
  # decide it has wrong configuration and switch to standalone
2852

    
2853
  def _Attach():
2854
    all_connected = True
2855

    
2856
    for rd in bdevs:
2857
      stats = rd.GetProcStatus()
2858

    
2859
      all_connected = (all_connected and
2860
                       (stats.is_connected or stats.is_in_resync))
2861

    
2862
      if stats.is_standalone:
2863
        # peer had different config info and this node became
2864
        # standalone, even though this should not happen with the
2865
        # new staged way of changing disk configs
2866
        try:
2867
          rd.AttachNet(multimaster)
2868
        except errors.BlockDeviceError, err:
2869
          _Fail("Can't change network configuration: %s", err)
2870

    
2871
    if not all_connected:
2872
      raise utils.RetryAgain()
2873

    
2874
  try:
2875
    # Start with a delay of 100 miliseconds and go up to 5 seconds
2876
    utils.Retry(_Attach, (0.1, 1.5, 5.0), 2 * 60)
2877
  except utils.RetryTimeout:
2878
    _Fail("Timeout in disk reconnecting")
2879

    
2880
  if multimaster:
2881
    # change to primary mode
2882
    for rd in bdevs:
2883
      try:
2884
        rd.Open()
2885
      except errors.BlockDeviceError, err:
2886
        _Fail("Can't change to primary mode: %s", err)
2887

    
2888

    
2889
def DrbdWaitSync(nodes_ip, disks):
2890
  """Wait until DRBDs have synchronized.
2891

2892
  """
2893
  def _helper(rd):
2894
    stats = rd.GetProcStatus()
2895
    if not (stats.is_connected or stats.is_in_resync):
2896
      raise utils.RetryAgain()
2897
    return stats
2898

    
2899
  bdevs = _FindDisks(nodes_ip, disks)
2900

    
2901
  min_resync = 100
2902
  alldone = True
2903
  for rd in bdevs:
2904
    try:
2905
      # poll each second for 15 seconds
2906
      stats = utils.Retry(_helper, 1, 15, args=[rd])
2907
    except utils.RetryTimeout:
2908
      stats = rd.GetProcStatus()
2909
      # last check
2910
      if not (stats.is_connected or stats.is_in_resync):
2911
        _Fail("DRBD device %s is not in sync: stats=%s", rd, stats)
2912
    alldone = alldone and (not stats.is_in_resync)
2913
    if stats.sync_percent is not None:
2914
      min_resync = min(min_resync, stats.sync_percent)
2915

    
2916
  return (alldone, min_resync)
2917

    
2918

    
2919
def PowercycleNode(hypervisor_type):
2920
  """Hard-powercycle the node.
2921

2922
  Because we need to return first, and schedule the powercycle in the
2923
  background, we won't be able to report failures nicely.
2924

2925
  """
2926
  hyper = hypervisor.GetHypervisor(hypervisor_type)
2927
  try:
2928
    pid = os.fork()
2929
  except OSError:
2930
    # if we can't fork, we'll pretend that we're in the child process
2931
    pid = 0
2932
  if pid > 0:
2933
    return "Reboot scheduled in 5 seconds"
2934
  # ensure the child is running on ram
2935
  try:
2936
    utils.Mlockall()
2937
  except Exception: # pylint: disable-msg=W0703
2938
    pass
2939
  time.sleep(5)
2940
  hyper.PowercycleNode()
2941

    
2942

    
2943
class HooksRunner(object):
2944
  """Hook runner.
2945

2946
  This class is instantiated on the node side (ganeti-noded) and not
2947
  on the master side.
2948

2949
  """
2950
  def __init__(self, hooks_base_dir=None):
2951
    """Constructor for hooks runner.
2952

2953
    @type hooks_base_dir: str or None
2954
    @param hooks_base_dir: if not None, this overrides the
2955
        L{constants.HOOKS_BASE_DIR} (useful for unittests)
2956

2957
    """
2958
    if hooks_base_dir is None:
2959
      hooks_base_dir = constants.HOOKS_BASE_DIR
2960
    # yeah, _BASE_DIR is not valid for attributes, we use it like a
2961
    # constant
2962
    self._BASE_DIR = hooks_base_dir # pylint: disable-msg=C0103
2963

    
2964
  def RunHooks(self, hpath, phase, env):
2965
    """Run the scripts in the hooks directory.
2966

2967
    @type hpath: str
2968
    @param hpath: the path to the hooks directory which
2969
        holds the scripts
2970
    @type phase: str
2971
    @param phase: either L{constants.HOOKS_PHASE_PRE} or
2972
        L{constants.HOOKS_PHASE_POST}
2973
    @type env: dict
2974
    @param env: dictionary with the environment for the hook
2975
    @rtype: list
2976
    @return: list of 3-element tuples:
2977
      - script path
2978
      - script result, either L{constants.HKR_SUCCESS} or
2979
        L{constants.HKR_FAIL}
2980
      - output of the script
2981

2982
    @raise errors.ProgrammerError: for invalid input
2983
        parameters
2984

2985
    """
2986
    if phase == constants.HOOKS_PHASE_PRE:
2987
      suffix = "pre"
2988
    elif phase == constants.HOOKS_PHASE_POST:
2989
      suffix = "post"
2990
    else:
2991
      _Fail("Unknown hooks phase '%s'", phase)
2992

    
2993

    
2994
    subdir = "%s-%s.d" % (hpath, suffix)
2995
    dir_name = utils.PathJoin(self._BASE_DIR, subdir)
2996

    
2997
    results = []
2998

    
2999
    if not os.path.isdir(dir_name):
3000
      # for non-existing/non-dirs, we simply exit instead of logging a
3001
      # warning at every operation
3002
      return results
3003

    
3004
    runparts_results = utils.RunParts(dir_name, env=env, reset_env=True)
3005

    
3006
    for (relname, relstatus, runresult)  in runparts_results:
3007
      if relstatus == constants.RUNPARTS_SKIP:
3008
        rrval = constants.HKR_SKIP
3009
        output = ""
3010
      elif relstatus == constants.RUNPARTS_ERR:
3011
        rrval = constants.HKR_FAIL
3012
        output = "Hook script execution error: %s" % runresult
3013
      elif relstatus == constants.RUNPARTS_RUN:
3014
        if runresult.failed:
3015
          rrval = constants.HKR_FAIL
3016
        else:
3017
          rrval = constants.HKR_SUCCESS
3018
        output = utils.SafeEncode(runresult.output.strip())
3019
      results.append(("%s/%s" % (subdir, relname), rrval, output))
3020

    
3021
    return results
3022

    
3023

    
3024
class IAllocatorRunner(object):
3025
  """IAllocator runner.
3026

3027
  This class is instantiated on the node side (ganeti-noded) and not on
3028
  the master side.
3029

3030
  """
3031
  @staticmethod
3032
  def Run(name, idata):
3033
    """Run an iallocator script.
3034

3035
    @type name: str
3036
    @param name: the iallocator script name
3037
    @type idata: str
3038
    @param idata: the allocator input data
3039

3040
    @rtype: tuple
3041
    @return: two element tuple of:
3042
       - status
3043
       - either error message or stdout of allocator (for success)
3044

3045
    """
3046
    alloc_script = utils.FindFile(name, constants.IALLOCATOR_SEARCH_PATH,
3047
                                  os.path.isfile)
3048
    if alloc_script is None:
3049
      _Fail("iallocator module '%s' not found in the search path", name)
3050

    
3051
    fd, fin_name = tempfile.mkstemp(prefix="ganeti-iallocator.")
3052
    try:
3053
      os.write(fd, idata)
3054
      os.close(fd)
3055
      result = utils.RunCmd([alloc_script, fin_name])
3056
      if result.failed:
3057
        _Fail("iallocator module '%s' failed: %s, output '%s'",
3058
              name, result.fail_reason, result.output)
3059
    finally:
3060
      os.unlink(fin_name)
3061

    
3062
    return result.stdout
3063

    
3064

    
3065
class DevCacheManager(object):
3066
  """Simple class for managing a cache of block device information.
3067

3068
  """
3069
  _DEV_PREFIX = "/dev/"
3070
  _ROOT_DIR = constants.BDEV_CACHE_DIR
3071

    
3072
  @classmethod
3073
  def _ConvertPath(cls, dev_path):
3074
    """Converts a /dev/name path to the cache file name.
3075

3076
    This replaces slashes with underscores and strips the /dev
3077
    prefix. It then returns the full path to the cache file.
3078

3079
    @type dev_path: str
3080
    @param dev_path: the C{/dev/} path name
3081
    @rtype: str
3082
    @return: the converted path name
3083

3084
    """
3085
    if dev_path.startswith(cls._DEV_PREFIX):
3086
      dev_path = dev_path[len(cls._DEV_PREFIX):]
3087
    dev_path = dev_path.replace("/", "_")
3088
    fpath = utils.PathJoin(cls._ROOT_DIR, "bdev_%s" % dev_path)
3089
    return fpath
3090

    
3091
  @classmethod
3092
  def UpdateCache(cls, dev_path, owner, on_primary, iv_name):
3093
    """Updates the cache information for a given device.
3094

3095
    @type dev_path: str
3096
    @param dev_path: the pathname of the device
3097
    @type owner: str
3098
    @param owner: the owner (instance name) of the device
3099
    @type on_primary: bool
3100
    @param on_primary: whether this is the primary
3101
        node nor not
3102
    @type iv_name: str
3103
    @param iv_name: the instance-visible name of the
3104
        device, as in objects.Disk.iv_name
3105

3106
    @rtype: None
3107

3108
    """
3109
    if dev_path is None:
3110
      logging.error("DevCacheManager.UpdateCache got a None dev_path")
3111
      return
3112
    fpath = cls._ConvertPath(dev_path)
3113
    if on_primary:
3114
      state = "primary"
3115
    else:
3116
      state = "secondary"
3117
    if iv_name is None:
3118
      iv_name = "not_visible"
3119
    fdata = "%s %s %s\n" % (str(owner), state, iv_name)
3120
    try:
3121
      utils.WriteFile(fpath, data=fdata)
3122
    except EnvironmentError, err:
3123
      logging.exception("Can't update bdev cache for %s: %s", dev_path, err)
3124

    
3125
  @classmethod
3126
  def RemoveCache(cls, dev_path):
3127
    """Remove data for a dev_path.
3128

3129
    This is just a wrapper over L{utils.RemoveFile} with a converted
3130
    path name and logging.
3131

3132
    @type dev_path: str
3133
    @param dev_path: the pathname of the device
3134

3135
    @rtype: None
3136

3137
    """
3138
    if dev_path is None:
3139
      logging.error("DevCacheManager.RemoveCache got a None dev_path")
3140
      return
3141
    fpath = cls._ConvertPath(dev_path)
3142
    try:
3143
      utils.RemoveFile(fpath)
3144
    except EnvironmentError, err:
3145
      logging.exception("Can't update bdev cache for %s: %s", dev_path, err)