Statistics
| Branch: | Tag: | Revision:

root / lib / backend.py @ c19f9810

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

27
"""
28

    
29
# pylint: disable-msg=E1103
30

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

    
35

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

    
50
from ganeti import errors
51
from ganeti import utils
52
from ganeti import ssh
53
from ganeti import hypervisor
54
from ganeti import constants
55
from ganeti import bdev
56
from ganeti import objects
57
from ganeti import ssconf
58

    
59

    
60
_BOOT_ID_PATH = "/proc/sys/kernel/random/boot_id"
61

    
62

    
63
class RPCFail(Exception):
64
  """Class denoting RPC failure.
65

66
  Its argument is the error message.
67

68
  """
69

    
70

    
71
def _Fail(msg, *args, **kwargs):
72
  """Log an error and the raise an RPCFail exception.
73

74
  This exception is then handled specially in the ganeti daemon and
75
  turned into a 'failed' return type. As such, this function is a
76
  useful shortcut for logging the error and returning it to the master
77
  daemon.
78

79
  @type msg: string
80
  @param msg: the text of the exception
81
  @raise RPCFail
82

83
  """
84
  if args:
85
    msg = msg % args
86
  if "log" not in kwargs or kwargs["log"]: # if we should log this error
87
    if "exc" in kwargs and kwargs["exc"]:
88
      logging.exception(msg)
89
    else:
90
      logging.error(msg)
91
  raise RPCFail(msg)
92

    
93

    
94
def _GetConfig():
95
  """Simple wrapper to return a SimpleStore.
96

97
  @rtype: L{ssconf.SimpleStore}
98
  @return: a SimpleStore instance
99

100
  """
101
  return ssconf.SimpleStore()
102

    
103

    
104
def _GetSshRunner(cluster_name):
105
  """Simple wrapper to return an SshRunner.
106

107
  @type cluster_name: str
108
  @param cluster_name: the cluster name, which is needed
109
      by the SshRunner constructor
110
  @rtype: L{ssh.SshRunner}
111
  @return: an SshRunner instance
112

113
  """
114
  return ssh.SshRunner(cluster_name)
115

    
116

    
117
def _Decompress(data):
118
  """Unpacks data compressed by the RPC client.
119

120
  @type data: list or tuple
121
  @param data: Data sent by RPC client
122
  @rtype: str
123
  @return: Decompressed data
124

125
  """
126
  assert isinstance(data, (list, tuple))
127
  assert len(data) == 2
128
  (encoding, content) = data
129
  if encoding == constants.RPC_ENCODING_NONE:
130
    return content
131
  elif encoding == constants.RPC_ENCODING_ZLIB_BASE64:
132
    return zlib.decompress(base64.b64decode(content))
133
  else:
134
    raise AssertionError("Unknown data encoding")
135

    
136

    
137
def _CleanDirectory(path, exclude=None):
138
  """Removes all regular files in a directory.
139

140
  @type path: str
141
  @param path: the directory to clean
142
  @type exclude: list
143
  @param exclude: list of files to be excluded, defaults
144
      to the empty list
145

146
  """
147
  if not os.path.isdir(path):
148
    return
149
  if exclude is None:
150
    exclude = []
151
  else:
152
    # Normalize excluded paths
153
    exclude = [os.path.normpath(i) for i in exclude]
154

    
155
  for rel_name in utils.ListVisibleFiles(path):
156
    full_name = os.path.normpath(os.path.join(path, rel_name))
157
    if full_name in exclude:
158
      continue
159
    if os.path.isfile(full_name) and not os.path.islink(full_name):
160
      utils.RemoveFile(full_name)
161

    
162

    
163
def _BuildUploadFileList():
164
  """Build the list of allowed upload files.
165

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

168
  """
169
  allowed_files = set([
170
    constants.CLUSTER_CONF_FILE,
171
    constants.ETC_HOSTS,
172
    constants.SSH_KNOWN_HOSTS_FILE,
173
    constants.VNC_PASSWORD_FILE,
174
    constants.RAPI_CERT_FILE,
175
    constants.RAPI_USERS_FILE,
176
    constants.HMAC_CLUSTER_KEY,
177
    ])
178

    
179
  for hv_name in constants.HYPER_TYPES:
180
    hv_class = hypervisor.GetHypervisorClass(hv_name)
181
    allowed_files.update(hv_class.GetAncillaryFiles())
182

    
183
  return frozenset(allowed_files)
184

    
185

    
186
_ALLOWED_UPLOAD_FILES = _BuildUploadFileList()
187

    
188

    
189
def JobQueuePurge():
190
  """Removes job queue files and archived jobs.
191

192
  @rtype: tuple
193
  @return: True, None
194

195
  """
196
  _CleanDirectory(constants.QUEUE_DIR, exclude=[constants.JOB_QUEUE_LOCK_FILE])
197
  _CleanDirectory(constants.JOB_QUEUE_ARCHIVE_DIR)
198

    
199

    
200
def GetMasterInfo():
201
  """Returns master information.
202

203
  This is an utility function to compute master information, either
204
  for consumption here or from the node daemon.
205

206
  @rtype: tuple
207
  @return: master_netdev, master_ip, master_name
208
  @raise RPCFail: in case of errors
209

210
  """
211
  try:
212
    cfg = _GetConfig()
213
    master_netdev = cfg.GetMasterNetdev()
214
    master_ip = cfg.GetMasterIP()
215
    master_node = cfg.GetMasterNode()
216
  except errors.ConfigurationError, err:
217
    _Fail("Cluster configuration incomplete: %s", err, exc=True)
218
  return (master_netdev, master_ip, master_node)
219

    
220

    
221
def StartMaster(start_daemons, no_voting):
222
  """Activate local node as master node.
223

224
  The function will always try activate the IP address of the master
225
  (unless someone else has it). It will also start the master daemons,
226
  based on the start_daemons parameter.
227

228
  @type start_daemons: boolean
229
  @param start_daemons: whether to also start the master
230
      daemons (ganeti-masterd and ganeti-rapi)
231
  @type no_voting: boolean
232
  @param no_voting: whether to start ganeti-masterd without a node vote
233
      (if start_daemons is True), but still non-interactively
234
  @rtype: None
235

236
  """
237
  # GetMasterInfo will raise an exception if not able to return data
238
  master_netdev, master_ip, _ = GetMasterInfo()
239

    
240
  err_msgs = []
241
  if utils.TcpPing(master_ip, constants.DEFAULT_NODED_PORT):
242
    if utils.OwnIpAddress(master_ip):
243
      # we already have the ip:
244
      logging.debug("Master IP already configured, doing nothing")
245
    else:
246
      msg = "Someone else has the master ip, not activating"
247
      logging.error(msg)
248
      err_msgs.append(msg)
249
  else:
250
    result = utils.RunCmd(["ip", "address", "add", "%s/32" % master_ip,
251
                           "dev", master_netdev, "label",
252
                           "%s:0" % master_netdev])
253
    if result.failed:
254
      msg = "Can't activate master IP: %s" % result.output
255
      logging.error(msg)
256
      err_msgs.append(msg)
257

    
258
    result = utils.RunCmd(["arping", "-q", "-U", "-c 3", "-I", master_netdev,
259
                           "-s", master_ip, master_ip])
260
    # we'll ignore the exit code of arping
261

    
262
  # and now start the master and rapi daemons
263
  if start_daemons:
264
    if no_voting:
265
      masterd_args = "--no-voting --yes-do-it"
266
    else:
267
      masterd_args = ""
268

    
269
    env = {
270
      "EXTRA_MASTERD_ARGS": masterd_args,
271
      }
272

    
273
    result = utils.RunCmd([constants.DAEMON_UTIL, "start-master"], env=env)
274
    if result.failed:
275
      msg = "Can't start Ganeti master: %s" % result.output
276
      logging.error(msg)
277
      err_msgs.append(msg)
278

    
279
  if err_msgs:
280
    _Fail("; ".join(err_msgs))
281

    
282

    
283
def StopMaster(stop_daemons):
284
  """Deactivate this node as master.
285

286
  The function will always try to deactivate the IP address of the
287
  master. It will also stop the master daemons depending on the
288
  stop_daemons parameter.
289

290
  @type stop_daemons: boolean
291
  @param stop_daemons: whether to also stop the master daemons
292
      (ganeti-masterd and ganeti-rapi)
293
  @rtype: None
294

295
  """
296
  # TODO: log and report back to the caller the error failures; we
297
  # need to decide in which case we fail the RPC for this
298

    
299
  # GetMasterInfo will raise an exception if not able to return data
300
  master_netdev, master_ip, _ = GetMasterInfo()
301

    
302
  result = utils.RunCmd(["ip", "address", "del", "%s/32" % master_ip,
303
                         "dev", master_netdev])
304
  if result.failed:
305
    logging.error("Can't remove the master IP, error: %s", result.output)
306
    # but otherwise ignore the failure
307

    
308
  if stop_daemons:
309
    result = utils.RunCmd([constants.DAEMON_UTIL, "stop-master"])
310
    if result.failed:
311
      logging.error("Could not stop Ganeti master, command %s had exitcode %s"
312
                    " and error %s",
313
                    result.cmd, result.exit_code, result.output)
314

    
315

    
316
def AddNode(dsa, dsapub, rsa, rsapub, sshkey, sshpub):
317
  """Joins this node to the cluster.
318

319
  This does the following:
320
      - updates the hostkeys of the machine (rsa and dsa)
321
      - adds the ssh private key to the user
322
      - adds the ssh public key to the users' authorized_keys file
323

324
  @type dsa: str
325
  @param dsa: the DSA private key to write
326
  @type dsapub: str
327
  @param dsapub: the DSA public key to write
328
  @type rsa: str
329
  @param rsa: the RSA private key to write
330
  @type rsapub: str
331
  @param rsapub: the RSA public key to write
332
  @type sshkey: str
333
  @param sshkey: the SSH private key to write
334
  @type sshpub: str
335
  @param sshpub: the SSH public key to write
336
  @rtype: boolean
337
  @return: the success of the operation
338

339
  """
340
  sshd_keys =  [(constants.SSH_HOST_RSA_PRIV, rsa, 0600),
341
                (constants.SSH_HOST_RSA_PUB, rsapub, 0644),
342
                (constants.SSH_HOST_DSA_PRIV, dsa, 0600),
343
                (constants.SSH_HOST_DSA_PUB, dsapub, 0644)]
344
  for name, content, mode in sshd_keys:
345
    utils.WriteFile(name, data=content, mode=mode)
346

    
347
  try:
348
    priv_key, pub_key, auth_keys = ssh.GetUserFiles(constants.GANETI_RUNAS,
349
                                                    mkdir=True)
350
  except errors.OpExecError, err:
351
    _Fail("Error while processing user ssh files: %s", err, exc=True)
352

    
353
  for name, content in [(priv_key, sshkey), (pub_key, sshpub)]:
354
    utils.WriteFile(name, data=content, mode=0600)
355

    
356
  utils.AddAuthorizedKey(auth_keys, sshpub)
357

    
358
  result = utils.RunCmd([constants.DAEMON_UTIL, "reload-ssh-keys"])
359
  if result.failed:
360
    _Fail("Unable to reload SSH keys (command %r, exit code %s, output %r)",
361
          result.cmd, result.exit_code, result.output)
362

    
363

    
364
def LeaveCluster(modify_ssh_setup):
365
  """Cleans up and remove the current node.
366

367
  This function cleans up and prepares the current node to be removed
368
  from the cluster.
369

370
  If processing is successful, then it raises an
371
  L{errors.QuitGanetiException} which is used as a special case to
372
  shutdown the node daemon.
373

374
  @param modify_ssh_setup: boolean
375

376
  """
377
  _CleanDirectory(constants.DATA_DIR)
378
  JobQueuePurge()
379

    
380
  if modify_ssh_setup:
381
    try:
382
      priv_key, pub_key, auth_keys = ssh.GetUserFiles(constants.GANETI_RUNAS)
383

    
384
      utils.RemoveAuthorizedKey(auth_keys, utils.ReadFile(pub_key))
385

    
386
      utils.RemoveFile(priv_key)
387
      utils.RemoveFile(pub_key)
388
    except errors.OpExecError:
389
      logging.exception("Error while processing ssh files")
390

    
391
  try:
392
    utils.RemoveFile(constants.HMAC_CLUSTER_KEY)
393
    utils.RemoveFile(constants.RAPI_CERT_FILE)
394
    utils.RemoveFile(constants.SSL_CERT_FILE)
395
  except: # pylint: disable-msg=W0702
396
    logging.exception("Error while removing cluster secrets")
397

    
398
  result = utils.RunCmd([constants.DAEMON_UTIL, "stop", constants.CONFD])
399
  if result.failed:
400
    logging.error("Command %s failed with exitcode %s and error %s",
401
                  result.cmd, result.exit_code, result.output)
402

    
403
  # Raise a custom exception (handled in ganeti-noded)
404
  raise errors.QuitGanetiException(True, 'Shutdown scheduled')
405

    
406

    
407
def GetNodeInfo(vgname, hypervisor_type):
408
  """Gives back a hash with different information about the node.
409

410
  @type vgname: C{string}
411
  @param vgname: the name of the volume group to ask for disk space information
412
  @type hypervisor_type: C{str}
413
  @param hypervisor_type: the name of the hypervisor to ask for
414
      memory information
415
  @rtype: C{dict}
416
  @return: dictionary with the following keys:
417
      - vg_size is the size of the configured volume group in MiB
418
      - vg_free is the free size of the volume group in MiB
419
      - memory_dom0 is the memory allocated for domain0 in MiB
420
      - memory_free is the currently available (free) ram in MiB
421
      - memory_total is the total number of ram in MiB
422

423
  """
424
  outputarray = {}
425
  vginfo = _GetVGInfo(vgname)
426
  outputarray['vg_size'] = vginfo['vg_size']
427
  outputarray['vg_free'] = vginfo['vg_free']
428

    
429
  hyper = hypervisor.GetHypervisor(hypervisor_type)
430
  hyp_info = hyper.GetNodeInfo()
431
  if hyp_info is not None:
432
    outputarray.update(hyp_info)
433

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

    
436
  return outputarray
437

    
438

    
439
def VerifyNode(what, cluster_name):
440
  """Verify the status of the local node.
441

442
  Based on the input L{what} parameter, various checks are done on the
443
  local node.
444

445
  If the I{filelist} key is present, this list of
446
  files is checksummed and the file/checksum pairs are returned.
447

448
  If the I{nodelist} key is present, we check that we have
449
  connectivity via ssh with the target nodes (and check the hostname
450
  report).
451

452
  If the I{node-net-test} key is present, we check that we have
453
  connectivity to the given nodes via both primary IP and, if
454
  applicable, secondary IPs.
455

456
  @type what: C{dict}
457
  @param what: a dictionary of things to check:
458
      - filelist: list of files for which to compute checksums
459
      - nodelist: list of nodes we should check ssh communication with
460
      - node-net-test: list of nodes we should check node daemon port
461
        connectivity with
462
      - hypervisor: list with hypervisors to run the verify for
463
  @rtype: dict
464
  @return: a dictionary with the same keys as the input dict, and
465
      values representing the result of the checks
466

467
  """
468
  result = {}
469

    
470
  if constants.NV_HYPERVISOR in what:
471
    result[constants.NV_HYPERVISOR] = tmp = {}
472
    for hv_name in what[constants.NV_HYPERVISOR]:
473
      tmp[hv_name] = hypervisor.GetHypervisor(hv_name).Verify()
474

    
475
  if constants.NV_FILELIST in what:
476
    result[constants.NV_FILELIST] = utils.FingerprintFiles(
477
      what[constants.NV_FILELIST])
478

    
479
  if constants.NV_NODELIST in what:
480
    result[constants.NV_NODELIST] = tmp = {}
481
    random.shuffle(what[constants.NV_NODELIST])
482
    for node in what[constants.NV_NODELIST]:
483
      success, message = _GetSshRunner(cluster_name).VerifyNodeHostname(node)
484
      if not success:
485
        tmp[node] = message
486

    
487
  if constants.NV_NODENETTEST in what:
488
    result[constants.NV_NODENETTEST] = tmp = {}
489
    my_name = utils.HostInfo().name
490
    my_pip = my_sip = None
491
    for name, pip, sip in what[constants.NV_NODENETTEST]:
492
      if name == my_name:
493
        my_pip = pip
494
        my_sip = sip
495
        break
496
    if not my_pip:
497
      tmp[my_name] = ("Can't find my own primary/secondary IP"
498
                      " in the node list")
499
    else:
500
      port = utils.GetDaemonPort(constants.NODED)
501
      for name, pip, sip in what[constants.NV_NODENETTEST]:
502
        fail = []
503
        if not utils.TcpPing(pip, port, source=my_pip):
504
          fail.append("primary")
505
        if sip != pip:
506
          if not utils.TcpPing(sip, port, source=my_sip):
507
            fail.append("secondary")
508
        if fail:
509
          tmp[name] = ("failure using the %s interface(s)" %
510
                       " and ".join(fail))
511

    
512
  if constants.NV_LVLIST in what:
513
    result[constants.NV_LVLIST] = GetVolumeList(what[constants.NV_LVLIST])
514

    
515
  if constants.NV_INSTANCELIST in what:
516
    result[constants.NV_INSTANCELIST] = GetInstanceList(
517
      what[constants.NV_INSTANCELIST])
518

    
519
  if constants.NV_VGLIST in what:
520
    result[constants.NV_VGLIST] = utils.ListVolumeGroups()
521

    
522
  if constants.NV_PVLIST in what:
523
    result[constants.NV_PVLIST] = \
524
      bdev.LogicalVolume.GetPVInfo(what[constants.NV_PVLIST],
525
                                   filter_allocatable=False)
526

    
527
  if constants.NV_VERSION in what:
528
    result[constants.NV_VERSION] = (constants.PROTOCOL_VERSION,
529
                                    constants.RELEASE_VERSION)
530

    
531
  if constants.NV_HVINFO in what:
532
    hyper = hypervisor.GetHypervisor(what[constants.NV_HVINFO])
533
    result[constants.NV_HVINFO] = hyper.GetNodeInfo()
534

    
535
  if constants.NV_DRBDLIST in what:
536
    try:
537
      used_minors = bdev.DRBD8.GetUsedDevs().keys()
538
    except errors.BlockDeviceError, err:
539
      logging.warning("Can't get used minors list", exc_info=True)
540
      used_minors = str(err)
541
    result[constants.NV_DRBDLIST] = used_minors
542

    
543
  if constants.NV_NODESETUP in what:
544
    result[constants.NV_NODESETUP] = tmpr = []
545
    if not os.path.isdir("/sys/block") or not os.path.isdir("/sys/class/net"):
546
      tmpr.append("The sysfs filesytem doesn't seem to be mounted"
547
                  " under /sys, missing required directories /sys/block"
548
                  " and /sys/class/net")
549
    if (not os.path.isdir("/proc/sys") or
550
        not os.path.isfile("/proc/sysrq-trigger")):
551
      tmpr.append("The procfs filesystem doesn't seem to be mounted"
552
                  " under /proc, missing required directory /proc/sys and"
553
                  " the file /proc/sysrq-trigger")
554

    
555
  if constants.NV_TIME in what:
556
    result[constants.NV_TIME] = utils.SplitTime(time.time())
557

    
558
  return result
559

    
560

    
561
def GetVolumeList(vg_name):
562
  """Compute list of logical volumes and their size.
563

564
  @type vg_name: str
565
  @param vg_name: the volume group whose LVs we should list
566
  @rtype: dict
567
  @return:
568
      dictionary of all partions (key) with value being a tuple of
569
      their size (in MiB), inactive and online status::
570

571
        {'test1': ('20.06', True, True)}
572

573
      in case of errors, a string is returned with the error
574
      details.
575

576
  """
577
  lvs = {}
578
  sep = '|'
579
  result = utils.RunCmd(["lvs", "--noheadings", "--units=m", "--nosuffix",
580
                         "--separator=%s" % sep,
581
                         "-olv_name,lv_size,lv_attr", vg_name])
582
  if result.failed:
583
    _Fail("Failed to list logical volumes, lvs output: %s", result.output)
584

    
585
  valid_line_re = re.compile("^ *([^|]+)\|([0-9.]+)\|([^|]{6})\|?$")
586
  for line in result.stdout.splitlines():
587
    line = line.strip()
588
    match = valid_line_re.match(line)
589
    if not match:
590
      logging.error("Invalid line returned from lvs output: '%s'", line)
591
      continue
592
    name, size, attr = match.groups()
593
    inactive = attr[4] == '-'
594
    online = attr[5] == 'o'
595
    virtual = attr[0] == 'v'
596
    if virtual:
597
      # we don't want to report such volumes as existing, since they
598
      # don't really hold data
599
      continue
600
    lvs[name] = (size, inactive, online)
601

    
602
  return lvs
603

    
604

    
605
def ListVolumeGroups():
606
  """List the volume groups and their size.
607

608
  @rtype: dict
609
  @return: dictionary with keys volume name and values the
610
      size of the volume
611

612
  """
613
  return utils.ListVolumeGroups()
614

    
615

    
616
def NodeVolumes():
617
  """List all volumes on this node.
618

619
  @rtype: list
620
  @return:
621
    A list of dictionaries, each having four keys:
622
      - name: the logical volume name,
623
      - size: the size of the logical volume
624
      - dev: the physical device on which the LV lives
625
      - vg: the volume group to which it belongs
626

627
    In case of errors, we return an empty list and log the
628
    error.
629

630
    Note that since a logical volume can live on multiple physical
631
    volumes, the resulting list might include a logical volume
632
    multiple times.
633

634
  """
635
  result = utils.RunCmd(["lvs", "--noheadings", "--units=m", "--nosuffix",
636
                         "--separator=|",
637
                         "--options=lv_name,lv_size,devices,vg_name"])
638
  if result.failed:
639
    _Fail("Failed to list logical volumes, lvs output: %s",
640
          result.output)
641

    
642
  def parse_dev(dev):
643
    if '(' in dev:
644
      return dev.split('(')[0]
645
    else:
646
      return dev
647

    
648
  def map_line(line):
649
    return {
650
      'name': line[0].strip(),
651
      'size': line[1].strip(),
652
      'dev': parse_dev(line[2].strip()),
653
      'vg': line[3].strip(),
654
    }
655

    
656
  return [map_line(line.split('|')) for line in result.stdout.splitlines()
657
          if line.count('|') >= 3]
658

    
659

    
660
def BridgesExist(bridges_list):
661
  """Check if a list of bridges exist on the current node.
662

663
  @rtype: boolean
664
  @return: C{True} if all of them exist, C{False} otherwise
665

666
  """
667
  missing = []
668
  for bridge in bridges_list:
669
    if not utils.BridgeExists(bridge):
670
      missing.append(bridge)
671

    
672
  if missing:
673
    _Fail("Missing bridges %s", utils.CommaJoin(missing))
674

    
675

    
676
def GetInstanceList(hypervisor_list):
677
  """Provides a list of instances.
678

679
  @type hypervisor_list: list
680
  @param hypervisor_list: the list of hypervisors to query information
681

682
  @rtype: list
683
  @return: a list of all running instances on the current node
684
    - instance1.example.com
685
    - instance2.example.com
686

687
  """
688
  results = []
689
  for hname in hypervisor_list:
690
    try:
691
      names = hypervisor.GetHypervisor(hname).ListInstances()
692
      results.extend(names)
693
    except errors.HypervisorError, err:
694
      _Fail("Error enumerating instances (hypervisor %s): %s",
695
            hname, err, exc=True)
696

    
697
  return results
698

    
699

    
700
def GetInstanceInfo(instance, hname):
701
  """Gives back the information about an instance as a dictionary.
702

703
  @type instance: string
704
  @param instance: the instance name
705
  @type hname: string
706
  @param hname: the hypervisor type of the instance
707

708
  @rtype: dict
709
  @return: dictionary with the following keys:
710
      - memory: memory size of instance (int)
711
      - state: xen state of instance (string)
712
      - time: cpu time of instance (float)
713

714
  """
715
  output = {}
716

    
717
  iinfo = hypervisor.GetHypervisor(hname).GetInstanceInfo(instance)
718
  if iinfo is not None:
719
    output['memory'] = iinfo[2]
720
    output['state'] = iinfo[4]
721
    output['time'] = iinfo[5]
722

    
723
  return output
724

    
725

    
726
def GetInstanceMigratable(instance):
727
  """Gives whether an instance can be migrated.
728

729
  @type instance: L{objects.Instance}
730
  @param instance: object representing the instance to be checked.
731

732
  @rtype: tuple
733
  @return: tuple of (result, description) where:
734
      - result: whether the instance can be migrated or not
735
      - description: a description of the issue, if relevant
736

737
  """
738
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
739
  iname = instance.name
740
  if iname not in hyper.ListInstances():
741
    _Fail("Instance %s is not running", iname)
742

    
743
  for idx in range(len(instance.disks)):
744
    link_name = _GetBlockDevSymlinkPath(iname, idx)
745
    if not os.path.islink(link_name):
746
      _Fail("Instance %s was not restarted since ganeti 1.2.5", iname)
747

    
748

    
749
def GetAllInstancesInfo(hypervisor_list):
750
  """Gather data about all instances.
751

752
  This is the equivalent of L{GetInstanceInfo}, except that it
753
  computes data for all instances at once, thus being faster if one
754
  needs data about more than one instance.
755

756
  @type hypervisor_list: list
757
  @param hypervisor_list: list of hypervisors to query for instance data
758

759
  @rtype: dict
760
  @return: dictionary of instance: data, with data having the following keys:
761
      - memory: memory size of instance (int)
762
      - state: xen state of instance (string)
763
      - time: cpu time of instance (float)
764
      - vcpus: the number of vcpus
765

766
  """
767
  output = {}
768

    
769
  for hname in hypervisor_list:
770
    iinfo = hypervisor.GetHypervisor(hname).GetAllInstancesInfo()
771
    if iinfo:
772
      for name, _, memory, vcpus, state, times in iinfo:
773
        value = {
774
          'memory': memory,
775
          'vcpus': vcpus,
776
          'state': state,
777
          'time': times,
778
          }
779
        if name in output:
780
          # we only check static parameters, like memory and vcpus,
781
          # and not state and time which can change between the
782
          # invocations of the different hypervisors
783
          for key in 'memory', 'vcpus':
784
            if value[key] != output[name][key]:
785
              _Fail("Instance %s is running twice"
786
                    " with different parameters", name)
787
        output[name] = value
788

    
789
  return output
790

    
791

    
792
def InstanceOsAdd(instance, reinstall):
793
  """Add an OS to an instance.
794

795
  @type instance: L{objects.Instance}
796
  @param instance: Instance whose OS is to be installed
797
  @type reinstall: boolean
798
  @param reinstall: whether this is an instance reinstall
799
  @rtype: None
800

801
  """
802
  inst_os = OSFromDisk(instance.os)
803

    
804
  create_env = OSEnvironment(instance, inst_os)
805
  if reinstall:
806
    create_env['INSTANCE_REINSTALL'] = "1"
807

    
808
  logfile = "%s/add-%s-%s-%d.log" % (constants.LOG_OS_DIR, instance.os,
809
                                     instance.name, int(time.time()))
810

    
811
  result = utils.RunCmd([inst_os.create_script], env=create_env,
812
                        cwd=inst_os.path, output=logfile,)
813
  if result.failed:
814
    logging.error("os create command '%s' returned error: %s, logfile: %s,"
815
                  " output: %s", result.cmd, result.fail_reason, logfile,
816
                  result.output)
817
    lines = [utils.SafeEncode(val)
818
             for val in utils.TailFile(logfile, lines=20)]
819
    _Fail("OS create script failed (%s), last lines in the"
820
          " log file:\n%s", result.fail_reason, "\n".join(lines), log=False)
821

    
822

    
823
def RunRenameInstance(instance, old_name):
824
  """Run the OS rename script for an instance.
825

826
  @type instance: L{objects.Instance}
827
  @param instance: Instance whose OS is to be installed
828
  @type old_name: string
829
  @param old_name: previous instance name
830
  @rtype: boolean
831
  @return: the success of the operation
832

833
  """
834
  inst_os = OSFromDisk(instance.os)
835

    
836
  rename_env = OSEnvironment(instance, inst_os)
837
  rename_env['OLD_INSTANCE_NAME'] = old_name
838

    
839
  logfile = "%s/rename-%s-%s-%s-%d.log" % (constants.LOG_OS_DIR, instance.os,
840
                                           old_name,
841
                                           instance.name, int(time.time()))
842

    
843
  result = utils.RunCmd([inst_os.rename_script], env=rename_env,
844
                        cwd=inst_os.path, output=logfile)
845

    
846
  if result.failed:
847
    logging.error("os create command '%s' returned error: %s output: %s",
848
                  result.cmd, result.fail_reason, result.output)
849
    lines = [utils.SafeEncode(val)
850
             for val in utils.TailFile(logfile, lines=20)]
851
    _Fail("OS rename script failed (%s), last lines in the"
852
          " log file:\n%s", result.fail_reason, "\n".join(lines), log=False)
853

    
854

    
855
def _GetVGInfo(vg_name):
856
  """Get information about the volume group.
857

858
  @type vg_name: str
859
  @param vg_name: the volume group which we query
860
  @rtype: dict
861
  @return:
862
    A dictionary with the following keys:
863
      - C{vg_size} is the total size of the volume group in MiB
864
      - C{vg_free} is the free size of the volume group in MiB
865
      - C{pv_count} are the number of physical disks in that VG
866

867
    If an error occurs during gathering of data, we return the same dict
868
    with keys all set to None.
869

870
  """
871
  retdic = dict.fromkeys(["vg_size", "vg_free", "pv_count"])
872

    
873
  retval = utils.RunCmd(["vgs", "-ovg_size,vg_free,pv_count", "--noheadings",
874
                         "--nosuffix", "--units=m", "--separator=:", vg_name])
875

    
876
  if retval.failed:
877
    logging.error("volume group %s not present", vg_name)
878
    return retdic
879
  valarr = retval.stdout.strip().rstrip(':').split(':')
880
  if len(valarr) == 3:
881
    try:
882
      retdic = {
883
        "vg_size": int(round(float(valarr[0]), 0)),
884
        "vg_free": int(round(float(valarr[1]), 0)),
885
        "pv_count": int(valarr[2]),
886
        }
887
    except ValueError, err:
888
      logging.exception("Fail to parse vgs output: %s", err)
889
  else:
890
    logging.error("vgs output has the wrong number of fields (expected"
891
                  " three): %s", str(valarr))
892
  return retdic
893

    
894

    
895
def _GetBlockDevSymlinkPath(instance_name, idx):
896
  return os.path.join(constants.DISK_LINKS_DIR,
897
                      "%s:%d" % (instance_name, idx))
898

    
899

    
900
def _SymlinkBlockDev(instance_name, device_path, idx):
901
  """Set up symlinks to a instance's block device.
902

903
  This is an auxiliary function run when an instance is start (on the primary
904
  node) or when an instance is migrated (on the target node).
905

906

907
  @param instance_name: the name of the target instance
908
  @param device_path: path of the physical block device, on the node
909
  @param idx: the disk index
910
  @return: absolute path to the disk's symlink
911

912
  """
913
  link_name = _GetBlockDevSymlinkPath(instance_name, idx)
914
  try:
915
    os.symlink(device_path, link_name)
916
  except OSError, err:
917
    if err.errno == errno.EEXIST:
918
      if (not os.path.islink(link_name) or
919
          os.readlink(link_name) != device_path):
920
        os.remove(link_name)
921
        os.symlink(device_path, link_name)
922
    else:
923
      raise
924

    
925
  return link_name
926

    
927

    
928
def _RemoveBlockDevLinks(instance_name, disks):
929
  """Remove the block device symlinks belonging to the given instance.
930

931
  """
932
  for idx, _ in enumerate(disks):
933
    link_name = _GetBlockDevSymlinkPath(instance_name, idx)
934
    if os.path.islink(link_name):
935
      try:
936
        os.remove(link_name)
937
      except OSError:
938
        logging.exception("Can't remove symlink '%s'", link_name)
939

    
940

    
941
def _GatherAndLinkBlockDevs(instance):
942
  """Set up an instance's block device(s).
943

944
  This is run on the primary node at instance startup. The block
945
  devices must be already assembled.
946

947
  @type instance: L{objects.Instance}
948
  @param instance: the instance whose disks we shoul assemble
949
  @rtype: list
950
  @return: list of (disk_object, device_path)
951

952
  """
953
  block_devices = []
954
  for idx, disk in enumerate(instance.disks):
955
    device = _RecursiveFindBD(disk)
956
    if device is None:
957
      raise errors.BlockDeviceError("Block device '%s' is not set up." %
958
                                    str(disk))
959
    device.Open()
960
    try:
961
      link_name = _SymlinkBlockDev(instance.name, device.dev_path, idx)
962
    except OSError, e:
963
      raise errors.BlockDeviceError("Cannot create block device symlink: %s" %
964
                                    e.strerror)
965

    
966
    block_devices.append((disk, link_name))
967

    
968
  return block_devices
969

    
970

    
971
def StartInstance(instance):
972
  """Start an instance.
973

974
  @type instance: L{objects.Instance}
975
  @param instance: the instance object
976
  @rtype: None
977

978
  """
979
  running_instances = GetInstanceList([instance.hypervisor])
980

    
981
  if instance.name in running_instances:
982
    logging.info("Instance %s already running, not starting", instance.name)
983
    return
984

    
985
  try:
986
    block_devices = _GatherAndLinkBlockDevs(instance)
987
    hyper = hypervisor.GetHypervisor(instance.hypervisor)
988
    hyper.StartInstance(instance, block_devices)
989
  except errors.BlockDeviceError, err:
990
    _Fail("Block device error: %s", err, exc=True)
991
  except errors.HypervisorError, err:
992
    _RemoveBlockDevLinks(instance.name, instance.disks)
993
    _Fail("Hypervisor error: %s", err, exc=True)
994

    
995

    
996
def InstanceShutdown(instance, timeout):
997
  """Shut an instance down.
998

999
  @note: this functions uses polling with a hardcoded timeout.
1000

1001
  @type instance: L{objects.Instance}
1002
  @param instance: the instance object
1003
  @type timeout: integer
1004
  @param timeout: maximum timeout for soft shutdown
1005
  @rtype: None
1006

1007
  """
1008
  hv_name = instance.hypervisor
1009
  hyper = hypervisor.GetHypervisor(hv_name)
1010
  iname = instance.name
1011

    
1012
  if instance.name not in hyper.ListInstances():
1013
    logging.info("Instance %s not running, doing nothing", iname)
1014
    return
1015

    
1016
  class _TryShutdown:
1017
    def __init__(self):
1018
      self.tried_once = False
1019

    
1020
    def __call__(self):
1021
      if iname not in hyper.ListInstances():
1022
        return
1023

    
1024
      try:
1025
        hyper.StopInstance(instance, retry=self.tried_once)
1026
      except errors.HypervisorError, err:
1027
        if iname not in hyper.ListInstances():
1028
          # if the instance is no longer existing, consider this a
1029
          # success and go to cleanup
1030
          return
1031

    
1032
        _Fail("Failed to stop instance %s: %s", iname, err)
1033

    
1034
      self.tried_once = True
1035

    
1036
      raise utils.RetryAgain()
1037

    
1038
  try:
1039
    utils.Retry(_TryShutdown(), 5, timeout)
1040
  except utils.RetryTimeout:
1041
    # the shutdown did not succeed
1042
    logging.error("Shutdown of '%s' unsuccessful, forcing", iname)
1043

    
1044
    try:
1045
      hyper.StopInstance(instance, force=True)
1046
    except errors.HypervisorError, err:
1047
      if iname in hyper.ListInstances():
1048
        # only raise an error if the instance still exists, otherwise
1049
        # the error could simply be "instance ... unknown"!
1050
        _Fail("Failed to force stop instance %s: %s", iname, err)
1051

    
1052
    time.sleep(1)
1053

    
1054
    if iname in hyper.ListInstances():
1055
      _Fail("Could not shutdown instance %s even by destroy", iname)
1056

    
1057
  _RemoveBlockDevLinks(iname, instance.disks)
1058

    
1059

    
1060
def InstanceReboot(instance, reboot_type, shutdown_timeout):
1061
  """Reboot an instance.
1062

1063
  @type instance: L{objects.Instance}
1064
  @param instance: the instance object to reboot
1065
  @type reboot_type: str
1066
  @param reboot_type: the type of reboot, one the following
1067
    constants:
1068
      - L{constants.INSTANCE_REBOOT_SOFT}: only reboot the
1069
        instance OS, do not recreate the VM
1070
      - L{constants.INSTANCE_REBOOT_HARD}: tear down and
1071
        restart the VM (at the hypervisor level)
1072
      - the other reboot type (L{constants.INSTANCE_REBOOT_FULL}) is
1073
        not accepted here, since that mode is handled differently, in
1074
        cmdlib, and translates into full stop and start of the
1075
        instance (instead of a call_instance_reboot RPC)
1076
  @type shutdown_timeout: integer
1077
  @param shutdown_timeout: maximum timeout for soft shutdown
1078
  @rtype: None
1079

1080
  """
1081
  running_instances = GetInstanceList([instance.hypervisor])
1082

    
1083
  if instance.name not in running_instances:
1084
    _Fail("Cannot reboot instance %s that is not running", instance.name)
1085

    
1086
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1087
  if reboot_type == constants.INSTANCE_REBOOT_SOFT:
1088
    try:
1089
      hyper.RebootInstance(instance)
1090
    except errors.HypervisorError, err:
1091
      _Fail("Failed to soft reboot instance %s: %s", instance.name, err)
1092
  elif reboot_type == constants.INSTANCE_REBOOT_HARD:
1093
    try:
1094
      InstanceShutdown(instance, shutdown_timeout)
1095
      return StartInstance(instance)
1096
    except errors.HypervisorError, err:
1097
      _Fail("Failed to hard reboot instance %s: %s", instance.name, err)
1098
  else:
1099
    _Fail("Invalid reboot_type received: %s", reboot_type)
1100

    
1101

    
1102
def MigrationInfo(instance):
1103
  """Gather information about an instance to be migrated.
1104

1105
  @type instance: L{objects.Instance}
1106
  @param instance: the instance definition
1107

1108
  """
1109
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1110
  try:
1111
    info = hyper.MigrationInfo(instance)
1112
  except errors.HypervisorError, err:
1113
    _Fail("Failed to fetch migration information: %s", err, exc=True)
1114
  return info
1115

    
1116

    
1117
def AcceptInstance(instance, info, target):
1118
  """Prepare the node to accept an instance.
1119

1120
  @type instance: L{objects.Instance}
1121
  @param instance: the instance definition
1122
  @type info: string/data (opaque)
1123
  @param info: migration information, from the source node
1124
  @type target: string
1125
  @param target: target host (usually ip), on this node
1126

1127
  """
1128
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1129
  try:
1130
    hyper.AcceptInstance(instance, info, target)
1131
  except errors.HypervisorError, err:
1132
    _Fail("Failed to accept instance: %s", err, exc=True)
1133

    
1134

    
1135
def FinalizeMigration(instance, info, success):
1136
  """Finalize any preparation to accept an instance.
1137

1138
  @type instance: L{objects.Instance}
1139
  @param instance: the instance definition
1140
  @type info: string/data (opaque)
1141
  @param info: migration information, from the source node
1142
  @type success: boolean
1143
  @param success: whether the migration was a success or a failure
1144

1145
  """
1146
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1147
  try:
1148
    hyper.FinalizeMigration(instance, info, success)
1149
  except errors.HypervisorError, err:
1150
    _Fail("Failed to finalize migration: %s", err, exc=True)
1151

    
1152

    
1153
def MigrateInstance(instance, target, live):
1154
  """Migrates an instance to another node.
1155

1156
  @type instance: L{objects.Instance}
1157
  @param instance: the instance definition
1158
  @type target: string
1159
  @param target: the target node name
1160
  @type live: boolean
1161
  @param live: whether the migration should be done live or not (the
1162
      interpretation of this parameter is left to the hypervisor)
1163
  @rtype: tuple
1164
  @return: a tuple of (success, msg) where:
1165
      - succes is a boolean denoting the success/failure of the operation
1166
      - msg is a string with details in case of failure
1167

1168
  """
1169
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1170

    
1171
  try:
1172
    hyper.MigrateInstance(instance, target, live)
1173
  except errors.HypervisorError, err:
1174
    _Fail("Failed to migrate instance: %s", err, exc=True)
1175

    
1176

    
1177
def BlockdevCreate(disk, size, owner, on_primary, info):
1178
  """Creates a block device for an instance.
1179

1180
  @type disk: L{objects.Disk}
1181
  @param disk: the object describing the disk we should create
1182
  @type size: int
1183
  @param size: the size of the physical underlying device, in MiB
1184
  @type owner: str
1185
  @param owner: the name of the instance for which disk is created,
1186
      used for device cache data
1187
  @type on_primary: boolean
1188
  @param on_primary:  indicates if it is the primary node or not
1189
  @type info: string
1190
  @param info: string that will be sent to the physical device
1191
      creation, used for example to set (LVM) tags on LVs
1192

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

1197
  """
1198
  # TODO: remove the obsolete 'size' argument
1199
  # pylint: disable-msg=W0613
1200
  clist = []
1201
  if disk.children:
1202
    for child in disk.children:
1203
      try:
1204
        crdev = _RecursiveAssembleBD(child, owner, on_primary)
1205
      except errors.BlockDeviceError, err:
1206
        _Fail("Can't assemble device %s: %s", child, err)
1207
      if on_primary or disk.AssembleOnSecondary():
1208
        # we need the children open in case the device itself has to
1209
        # be assembled
1210
        try:
1211
          # pylint: disable-msg=E1103
1212
          crdev.Open()
1213
        except errors.BlockDeviceError, err:
1214
          _Fail("Can't make child '%s' read-write: %s", child, err)
1215
      clist.append(crdev)
1216

    
1217
  try:
1218
    device = bdev.Create(disk.dev_type, disk.physical_id, clist, disk.size)
1219
  except errors.BlockDeviceError, err:
1220
    _Fail("Can't create block device: %s", err)
1221

    
1222
  if on_primary or disk.AssembleOnSecondary():
1223
    try:
1224
      device.Assemble()
1225
    except errors.BlockDeviceError, err:
1226
      _Fail("Can't assemble device after creation, unusual event: %s", err)
1227
    device.SetSyncSpeed(constants.SYNC_SPEED)
1228
    if on_primary or disk.OpenOnSecondary():
1229
      try:
1230
        device.Open(force=True)
1231
      except errors.BlockDeviceError, err:
1232
        _Fail("Can't make device r/w after creation, unusual event: %s", err)
1233
    DevCacheManager.UpdateCache(device.dev_path, owner,
1234
                                on_primary, disk.iv_name)
1235

    
1236
  device.SetInfo(info)
1237

    
1238
  return device.unique_id
1239

    
1240

    
1241
def BlockdevRemove(disk):
1242
  """Remove a block device.
1243

1244
  @note: This is intended to be called recursively.
1245

1246
  @type disk: L{objects.Disk}
1247
  @param disk: the disk object we should remove
1248
  @rtype: boolean
1249
  @return: the success of the operation
1250

1251
  """
1252
  msgs = []
1253
  try:
1254
    rdev = _RecursiveFindBD(disk)
1255
  except errors.BlockDeviceError, err:
1256
    # probably can't attach
1257
    logging.info("Can't attach to device %s in remove", disk)
1258
    rdev = None
1259
  if rdev is not None:
1260
    r_path = rdev.dev_path
1261
    try:
1262
      rdev.Remove()
1263
    except errors.BlockDeviceError, err:
1264
      msgs.append(str(err))
1265
    if not msgs:
1266
      DevCacheManager.RemoveCache(r_path)
1267

    
1268
  if disk.children:
1269
    for child in disk.children:
1270
      try:
1271
        BlockdevRemove(child)
1272
      except RPCFail, err:
1273
        msgs.append(str(err))
1274

    
1275
  if msgs:
1276
    _Fail("; ".join(msgs))
1277

    
1278

    
1279
def _RecursiveAssembleBD(disk, owner, as_primary):
1280
  """Activate a block device for an instance.
1281

1282
  This is run on the primary and secondary nodes for an instance.
1283

1284
  @note: this function is called recursively.
1285

1286
  @type disk: L{objects.Disk}
1287
  @param disk: the disk we try to assemble
1288
  @type owner: str
1289
  @param owner: the name of the instance which owns the disk
1290
  @type as_primary: boolean
1291
  @param as_primary: if we should make the block device
1292
      read/write
1293

1294
  @return: the assembled device or None (in case no device
1295
      was assembled)
1296
  @raise errors.BlockDeviceError: in case there is an error
1297
      during the activation of the children or the device
1298
      itself
1299

1300
  """
1301
  children = []
1302
  if disk.children:
1303
    mcn = disk.ChildrenNeeded()
1304
    if mcn == -1:
1305
      mcn = 0 # max number of Nones allowed
1306
    else:
1307
      mcn = len(disk.children) - mcn # max number of Nones
1308
    for chld_disk in disk.children:
1309
      try:
1310
        cdev = _RecursiveAssembleBD(chld_disk, owner, as_primary)
1311
      except errors.BlockDeviceError, err:
1312
        if children.count(None) >= mcn:
1313
          raise
1314
        cdev = None
1315
        logging.error("Error in child activation (but continuing): %s",
1316
                      str(err))
1317
      children.append(cdev)
1318

    
1319
  if as_primary or disk.AssembleOnSecondary():
1320
    r_dev = bdev.Assemble(disk.dev_type, disk.physical_id, children, disk.size)
1321
    r_dev.SetSyncSpeed(constants.SYNC_SPEED)
1322
    result = r_dev
1323
    if as_primary or disk.OpenOnSecondary():
1324
      r_dev.Open()
1325
    DevCacheManager.UpdateCache(r_dev.dev_path, owner,
1326
                                as_primary, disk.iv_name)
1327

    
1328
  else:
1329
    result = True
1330
  return result
1331

    
1332

    
1333
def BlockdevAssemble(disk, owner, as_primary):
1334
  """Activate a block device for an instance.
1335

1336
  This is a wrapper over _RecursiveAssembleBD.
1337

1338
  @rtype: str or boolean
1339
  @return: a C{/dev/...} path for primary nodes, and
1340
      C{True} for secondary nodes
1341

1342
  """
1343
  try:
1344
    result = _RecursiveAssembleBD(disk, owner, as_primary)
1345
    if isinstance(result, bdev.BlockDev):
1346
      # pylint: disable-msg=E1103
1347
      result = result.dev_path
1348
  except errors.BlockDeviceError, err:
1349
    _Fail("Error while assembling disk: %s", err, exc=True)
1350

    
1351
  return result
1352

    
1353

    
1354
def BlockdevShutdown(disk):
1355
  """Shut down a block device.
1356

1357
  First, if the device is assembled (Attach() is successful), then
1358
  the device is shutdown. Then the children of the device are
1359
  shutdown.
1360

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

1365
  @type disk: L{objects.Disk}
1366
  @param disk: the description of the disk we should
1367
      shutdown
1368
  @rtype: None
1369

1370
  """
1371
  msgs = []
1372
  r_dev = _RecursiveFindBD(disk)
1373
  if r_dev is not None:
1374
    r_path = r_dev.dev_path
1375
    try:
1376
      r_dev.Shutdown()
1377
      DevCacheManager.RemoveCache(r_path)
1378
    except errors.BlockDeviceError, err:
1379
      msgs.append(str(err))
1380

    
1381
  if disk.children:
1382
    for child in disk.children:
1383
      try:
1384
        BlockdevShutdown(child)
1385
      except RPCFail, err:
1386
        msgs.append(str(err))
1387

    
1388
  if msgs:
1389
    _Fail("; ".join(msgs))
1390

    
1391

    
1392
def BlockdevAddchildren(parent_cdev, new_cdevs):
1393
  """Extend a mirrored block device.
1394

1395
  @type parent_cdev: L{objects.Disk}
1396
  @param parent_cdev: the disk to which we should add children
1397
  @type new_cdevs: list of L{objects.Disk}
1398
  @param new_cdevs: the list of children which we should add
1399
  @rtype: None
1400

1401
  """
1402
  parent_bdev = _RecursiveFindBD(parent_cdev)
1403
  if parent_bdev is None:
1404
    _Fail("Can't find parent device '%s' in add children", parent_cdev)
1405
  new_bdevs = [_RecursiveFindBD(disk) for disk in new_cdevs]
1406
  if new_bdevs.count(None) > 0:
1407
    _Fail("Can't find new device(s) to add: %s:%s", new_bdevs, new_cdevs)
1408
  parent_bdev.AddChildren(new_bdevs)
1409

    
1410

    
1411
def BlockdevRemovechildren(parent_cdev, new_cdevs):
1412
  """Shrink a mirrored block device.
1413

1414
  @type parent_cdev: L{objects.Disk}
1415
  @param parent_cdev: the disk from which we should remove children
1416
  @type new_cdevs: list of L{objects.Disk}
1417
  @param new_cdevs: the list of children which we should remove
1418
  @rtype: None
1419

1420
  """
1421
  parent_bdev = _RecursiveFindBD(parent_cdev)
1422
  if parent_bdev is None:
1423
    _Fail("Can't find parent device '%s' in remove children", parent_cdev)
1424
  devs = []
1425
  for disk in new_cdevs:
1426
    rpath = disk.StaticDevPath()
1427
    if rpath is None:
1428
      bd = _RecursiveFindBD(disk)
1429
      if bd is None:
1430
        _Fail("Can't find device %s while removing children", disk)
1431
      else:
1432
        devs.append(bd.dev_path)
1433
    else:
1434
      devs.append(rpath)
1435
  parent_bdev.RemoveChildren(devs)
1436

    
1437

    
1438
def BlockdevGetmirrorstatus(disks):
1439
  """Get the mirroring status of a list of devices.
1440

1441
  @type disks: list of L{objects.Disk}
1442
  @param disks: the list of disks which we should query
1443
  @rtype: disk
1444
  @return:
1445
      a list of (mirror_done, estimated_time) tuples, which
1446
      are the result of L{bdev.BlockDev.CombinedSyncStatus}
1447
  @raise errors.BlockDeviceError: if any of the disks cannot be
1448
      found
1449

1450
  """
1451
  stats = []
1452
  for dsk in disks:
1453
    rbd = _RecursiveFindBD(dsk)
1454
    if rbd is None:
1455
      _Fail("Can't find device %s", dsk)
1456

    
1457
    stats.append(rbd.CombinedSyncStatus())
1458

    
1459
  return stats
1460

    
1461

    
1462
def _RecursiveFindBD(disk):
1463
  """Check if a device is activated.
1464

1465
  If so, return information about the real device.
1466

1467
  @type disk: L{objects.Disk}
1468
  @param disk: the disk object we need to find
1469

1470
  @return: None if the device can't be found,
1471
      otherwise the device instance
1472

1473
  """
1474
  children = []
1475
  if disk.children:
1476
    for chdisk in disk.children:
1477
      children.append(_RecursiveFindBD(chdisk))
1478

    
1479
  return bdev.FindDevice(disk.dev_type, disk.physical_id, children, disk.size)
1480

    
1481

    
1482
def BlockdevFind(disk):
1483
  """Check if a device is activated.
1484

1485
  If it is, return information about the real device.
1486

1487
  @type disk: L{objects.Disk}
1488
  @param disk: the disk to find
1489
  @rtype: None or objects.BlockDevStatus
1490
  @return: None if the disk cannot be found, otherwise a the current
1491
           information
1492

1493
  """
1494
  try:
1495
    rbd = _RecursiveFindBD(disk)
1496
  except errors.BlockDeviceError, err:
1497
    _Fail("Failed to find device: %s", err, exc=True)
1498

    
1499
  if rbd is None:
1500
    return None
1501

    
1502
  return rbd.GetSyncStatus()
1503

    
1504

    
1505
def BlockdevGetsize(disks):
1506
  """Computes the size of the given disks.
1507

1508
  If a disk is not found, returns None instead.
1509

1510
  @type disks: list of L{objects.Disk}
1511
  @param disks: the list of disk to compute the size for
1512
  @rtype: list
1513
  @return: list with elements None if the disk cannot be found,
1514
      otherwise the size
1515

1516
  """
1517
  result = []
1518
  for cf in disks:
1519
    try:
1520
      rbd = _RecursiveFindBD(cf)
1521
    except errors.BlockDeviceError:
1522
      result.append(None)
1523
      continue
1524
    if rbd is None:
1525
      result.append(None)
1526
    else:
1527
      result.append(rbd.GetActualSize())
1528
  return result
1529

    
1530

    
1531
def BlockdevExport(disk, dest_node, dest_path, cluster_name):
1532
  """Export a block device to a remote node.
1533

1534
  @type disk: L{objects.Disk}
1535
  @param disk: the description of the disk to export
1536
  @type dest_node: str
1537
  @param dest_node: the destination node to export to
1538
  @type dest_path: str
1539
  @param dest_path: the destination path on the target node
1540
  @type cluster_name: str
1541
  @param cluster_name: the cluster name, needed for SSH hostalias
1542
  @rtype: None
1543

1544
  """
1545
  real_disk = _RecursiveFindBD(disk)
1546
  if real_disk is None:
1547
    _Fail("Block device '%s' is not set up", disk)
1548

    
1549
  real_disk.Open()
1550

    
1551
  # the block size on the read dd is 1MiB to match our units
1552
  expcmd = utils.BuildShellCmd("set -e; set -o pipefail; "
1553
                               "dd if=%s bs=1048576 count=%s",
1554
                               real_disk.dev_path, str(disk.size))
1555

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

    
1565
  remotecmd = _GetSshRunner(cluster_name).BuildCmd(dest_node,
1566
                                                   constants.GANETI_RUNAS,
1567
                                                   destcmd)
1568

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

    
1572
  result = utils.RunCmd(["bash", "-c", command])
1573

    
1574
  if result.failed:
1575
    _Fail("Disk copy command '%s' returned error: %s"
1576
          " output: %s", command, result.fail_reason, result.output)
1577

    
1578

    
1579
def UploadFile(file_name, data, mode, uid, gid, atime, mtime):
1580
  """Write a file to the filesystem.
1581

1582
  This allows the master to overwrite(!) a file. It will only perform
1583
  the operation if the file belongs to a list of configuration files.
1584

1585
  @type file_name: str
1586
  @param file_name: the target file name
1587
  @type data: str
1588
  @param data: the new contents of the file
1589
  @type mode: int
1590
  @param mode: the mode to give the file (can be None)
1591
  @type uid: int
1592
  @param uid: the owner of the file (can be -1 for default)
1593
  @type gid: int
1594
  @param gid: the group of the file (can be -1 for default)
1595
  @type atime: float
1596
  @param atime: the atime to set on the file (can be None)
1597
  @type mtime: float
1598
  @param mtime: the mtime to set on the file (can be None)
1599
  @rtype: None
1600

1601
  """
1602
  if not os.path.isabs(file_name):
1603
    _Fail("Filename passed to UploadFile is not absolute: '%s'", file_name)
1604

    
1605
  if file_name not in _ALLOWED_UPLOAD_FILES:
1606
    _Fail("Filename passed to UploadFile not in allowed upload targets: '%s'",
1607
          file_name)
1608

    
1609
  raw_data = _Decompress(data)
1610

    
1611
  utils.WriteFile(file_name, data=raw_data, mode=mode, uid=uid, gid=gid,
1612
                  atime=atime, mtime=mtime)
1613

    
1614

    
1615
def WriteSsconfFiles(values):
1616
  """Update all ssconf files.
1617

1618
  Wrapper around the SimpleStore.WriteFiles.
1619

1620
  """
1621
  ssconf.SimpleStore().WriteFiles(values)
1622

    
1623

    
1624
def _ErrnoOrStr(err):
1625
  """Format an EnvironmentError exception.
1626

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

1631
  @type err: L{EnvironmentError}
1632
  @param err: the exception to format
1633

1634
  """
1635
  if hasattr(err, 'errno'):
1636
    detail = errno.errorcode[err.errno]
1637
  else:
1638
    detail = str(err)
1639
  return detail
1640

    
1641

    
1642
def _OSOndiskAPIVersion(os_dir):
1643
  """Compute and return the API version of a given OS.
1644

1645
  This function will try to read the API version of the OS residing in
1646
  the 'os_dir' directory.
1647

1648
  @type os_dir: str
1649
  @param os_dir: the directory in which we should look for the OS
1650
  @rtype: tuple
1651
  @return: tuple (status, data) with status denoting the validity and
1652
      data holding either the vaid versions or an error message
1653

1654
  """
1655
  api_file = os.path.sep.join([os_dir, constants.OS_API_FILE])
1656

    
1657
  try:
1658
    st = os.stat(api_file)
1659
  except EnvironmentError, err:
1660
    return False, ("Required file '%s' not found under path %s: %s" %
1661
                   (constants.OS_API_FILE, os_dir, _ErrnoOrStr(err)))
1662

    
1663
  if not stat.S_ISREG(stat.S_IFMT(st.st_mode)):
1664
    return False, ("File '%s' in %s is not a regular file" %
1665
                   (constants.OS_API_FILE, os_dir))
1666

    
1667
  try:
1668
    api_versions = utils.ReadFile(api_file).splitlines()
1669
  except EnvironmentError, err:
1670
    return False, ("Error while reading the API version file at %s: %s" %
1671
                   (api_file, _ErrnoOrStr(err)))
1672

    
1673
  try:
1674
    api_versions = [int(version.strip()) for version in api_versions]
1675
  except (TypeError, ValueError), err:
1676
    return False, ("API version(s) can't be converted to integer: %s" %
1677
                   str(err))
1678

    
1679
  return True, api_versions
1680

    
1681

    
1682
def DiagnoseOS(top_dirs=None):
1683
  """Compute the validity for all OSes.
1684

1685
  @type top_dirs: list
1686
  @param top_dirs: the list of directories in which to
1687
      search (if not given defaults to
1688
      L{constants.OS_SEARCH_PATH})
1689
  @rtype: list of L{objects.OS}
1690
  @return: a list of tuples (name, path, status, diagnose, variants)
1691
      for all (potential) OSes under all search paths, where:
1692
          - name is the (potential) OS name
1693
          - path is the full path to the OS
1694
          - status True/False is the validity of the OS
1695
          - diagnose is the error message for an invalid OS, otherwise empty
1696
          - variants is a list of supported OS variants, if any
1697

1698
  """
1699
  if top_dirs is None:
1700
    top_dirs = constants.OS_SEARCH_PATH
1701

    
1702
  result = []
1703
  for dir_name in top_dirs:
1704
    if os.path.isdir(dir_name):
1705
      try:
1706
        f_names = utils.ListVisibleFiles(dir_name)
1707
      except EnvironmentError, err:
1708
        logging.exception("Can't list the OS directory %s: %s", dir_name, err)
1709
        break
1710
      for name in f_names:
1711
        os_path = os.path.sep.join([dir_name, name])
1712
        status, os_inst = _TryOSFromDisk(name, base_dir=dir_name)
1713
        if status:
1714
          diagnose = ""
1715
          variants = os_inst.supported_variants
1716
        else:
1717
          diagnose = os_inst
1718
          variants = []
1719
        result.append((name, os_path, status, diagnose, variants))
1720

    
1721
  return result
1722

    
1723

    
1724
def _TryOSFromDisk(name, base_dir=None):
1725
  """Create an OS instance from disk.
1726

1727
  This function will return an OS instance if the given name is a
1728
  valid OS name.
1729

1730
  @type base_dir: string
1731
  @keyword base_dir: Base directory containing OS installations.
1732
                     Defaults to a search in all the OS_SEARCH_PATH dirs.
1733
  @rtype: tuple
1734
  @return: success and either the OS instance if we find a valid one,
1735
      or error message
1736

1737
  """
1738
  if base_dir is None:
1739
    os_dir = utils.FindFile(name, constants.OS_SEARCH_PATH, os.path.isdir)
1740
  else:
1741
    os_dir = utils.FindFile(name, [base_dir], os.path.isdir)
1742

    
1743
  if os_dir is None:
1744
    return False, "Directory for OS %s not found in search path" % name
1745

    
1746
  status, api_versions = _OSOndiskAPIVersion(os_dir)
1747
  if not status:
1748
    # push the error up
1749
    return status, api_versions
1750

    
1751
  if not constants.OS_API_VERSIONS.intersection(api_versions):
1752
    return False, ("API version mismatch for path '%s': found %s, want %s." %
1753
                   (os_dir, api_versions, constants.OS_API_VERSIONS))
1754

    
1755
  # OS Files dictionary, we will populate it with the absolute path names
1756
  os_files = dict.fromkeys(constants.OS_SCRIPTS)
1757

    
1758
  if max(api_versions) >= constants.OS_API_V15:
1759
    os_files[constants.OS_VARIANTS_FILE] = ''
1760

    
1761
  for filename in os_files:
1762
    os_files[filename] = os.path.sep.join([os_dir, filename])
1763

    
1764
    try:
1765
      st = os.stat(os_files[filename])
1766
    except EnvironmentError, err:
1767
      return False, ("File '%s' under path '%s' is missing (%s)" %
1768
                     (filename, os_dir, _ErrnoOrStr(err)))
1769

    
1770
    if not stat.S_ISREG(stat.S_IFMT(st.st_mode)):
1771
      return False, ("File '%s' under path '%s' is not a regular file" %
1772
                     (filename, os_dir))
1773

    
1774
    if filename in constants.OS_SCRIPTS:
1775
      if stat.S_IMODE(st.st_mode) & stat.S_IXUSR != stat.S_IXUSR:
1776
        return False, ("File '%s' under path '%s' is not executable" %
1777
                       (filename, os_dir))
1778

    
1779
  variants = None
1780
  if constants.OS_VARIANTS_FILE in os_files:
1781
    variants_file = os_files[constants.OS_VARIANTS_FILE]
1782
    try:
1783
      variants = utils.ReadFile(variants_file).splitlines()
1784
    except EnvironmentError, err:
1785
      return False, ("Error while reading the OS variants file at %s: %s" %
1786
                     (variants_file, _ErrnoOrStr(err)))
1787
    if not variants:
1788
      return False, ("No supported os variant found")
1789

    
1790
  os_obj = objects.OS(name=name, path=os_dir,
1791
                      create_script=os_files[constants.OS_SCRIPT_CREATE],
1792
                      export_script=os_files[constants.OS_SCRIPT_EXPORT],
1793
                      import_script=os_files[constants.OS_SCRIPT_IMPORT],
1794
                      rename_script=os_files[constants.OS_SCRIPT_RENAME],
1795
                      supported_variants=variants,
1796
                      api_versions=api_versions)
1797
  return True, os_obj
1798

    
1799

    
1800
def OSFromDisk(name, base_dir=None):
1801
  """Create an OS instance from disk.
1802

1803
  This function will return an OS instance if the given name is a
1804
  valid OS name. Otherwise, it will raise an appropriate
1805
  L{RPCFail} exception, detailing why this is not a valid OS.
1806

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

1810
  @type base_dir: string
1811
  @keyword base_dir: Base directory containing OS installations.
1812
                     Defaults to a search in all the OS_SEARCH_PATH dirs.
1813
  @rtype: L{objects.OS}
1814
  @return: the OS instance if we find a valid one
1815
  @raise RPCFail: if we don't find a valid OS
1816

1817
  """
1818
  name_only = name.split("+", 1)[0]
1819
  status, payload = _TryOSFromDisk(name_only, base_dir)
1820

    
1821
  if not status:
1822
    _Fail(payload)
1823

    
1824
  return payload
1825

    
1826

    
1827
def OSEnvironment(instance, inst_os, debug=0):
1828
  """Calculate the environment for an os script.
1829

1830
  @type instance: L{objects.Instance}
1831
  @param instance: target instance for the os script run
1832
  @type inst_os: L{objects.OS}
1833
  @param inst_os: operating system for which the environment is being built
1834
  @type debug: integer
1835
  @param debug: debug level (0 or 1, for OS Api 10)
1836
  @rtype: dict
1837
  @return: dict of environment variables
1838
  @raise errors.BlockDeviceError: if the block device
1839
      cannot be found
1840

1841
  """
1842
  result = {}
1843
  api_version = \
1844
    max(constants.OS_API_VERSIONS.intersection(inst_os.api_versions))
1845
  result['OS_API_VERSION'] = '%d' % api_version
1846
  result['INSTANCE_NAME'] = instance.name
1847
  result['INSTANCE_OS'] = instance.os
1848
  result['HYPERVISOR'] = instance.hypervisor
1849
  result['DISK_COUNT'] = '%d' % len(instance.disks)
1850
  result['NIC_COUNT'] = '%d' % len(instance.nics)
1851
  result['DEBUG_LEVEL'] = '%d' % debug
1852
  if api_version >= constants.OS_API_V15:
1853
    try:
1854
      variant = instance.os.split('+', 1)[1]
1855
    except IndexError:
1856
      variant = inst_os.supported_variants[0]
1857
    result['OS_VARIANT'] = variant
1858
  for idx, disk in enumerate(instance.disks):
1859
    real_disk = _RecursiveFindBD(disk)
1860
    if real_disk is None:
1861
      raise errors.BlockDeviceError("Block device '%s' is not set up" %
1862
                                    str(disk))
1863
    real_disk.Open()
1864
    result['DISK_%d_PATH' % idx] = real_disk.dev_path
1865
    result['DISK_%d_ACCESS' % idx] = disk.mode
1866
    if constants.HV_DISK_TYPE in instance.hvparams:
1867
      result['DISK_%d_FRONTEND_TYPE' % idx] = \
1868
        instance.hvparams[constants.HV_DISK_TYPE]
1869
    if disk.dev_type in constants.LDS_BLOCK:
1870
      result['DISK_%d_BACKEND_TYPE' % idx] = 'block'
1871
    elif disk.dev_type == constants.LD_FILE:
1872
      result['DISK_%d_BACKEND_TYPE' % idx] = \
1873
        'file:%s' % disk.physical_id[0]
1874
  for idx, nic in enumerate(instance.nics):
1875
    result['NIC_%d_MAC' % idx] = nic.mac
1876
    if nic.ip:
1877
      result['NIC_%d_IP' % idx] = nic.ip
1878
    result['NIC_%d_MODE' % idx] = nic.nicparams[constants.NIC_MODE]
1879
    if nic.nicparams[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED:
1880
      result['NIC_%d_BRIDGE' % idx] = nic.nicparams[constants.NIC_LINK]
1881
    if nic.nicparams[constants.NIC_LINK]:
1882
      result['NIC_%d_LINK' % idx] = nic.nicparams[constants.NIC_LINK]
1883
    if constants.HV_NIC_TYPE in instance.hvparams:
1884
      result['NIC_%d_FRONTEND_TYPE' % idx] = \
1885
        instance.hvparams[constants.HV_NIC_TYPE]
1886

    
1887
  for source, kind in [(instance.beparams, "BE"), (instance.hvparams, "HV")]:
1888
    for key, value in source.items():
1889
      result["INSTANCE_%s_%s" % (kind, key)] = str(value)
1890

    
1891
  return result
1892

    
1893
def BlockdevGrow(disk, amount):
1894
  """Grow a stack of block devices.
1895

1896
  This function is called recursively, with the childrens being the
1897
  first ones to resize.
1898

1899
  @type disk: L{objects.Disk}
1900
  @param disk: the disk to be grown
1901
  @rtype: (status, result)
1902
  @return: a tuple with the status of the operation
1903
      (True/False), and the errors message if status
1904
      is False
1905

1906
  """
1907
  r_dev = _RecursiveFindBD(disk)
1908
  if r_dev is None:
1909
    _Fail("Cannot find block device %s", disk)
1910

    
1911
  try:
1912
    r_dev.Grow(amount)
1913
  except errors.BlockDeviceError, err:
1914
    _Fail("Failed to grow block device: %s", err, exc=True)
1915

    
1916

    
1917
def BlockdevSnapshot(disk):
1918
  """Create a snapshot copy of a block device.
1919

1920
  This function is called recursively, and the snapshot is actually created
1921
  just for the leaf lvm backend device.
1922

1923
  @type disk: L{objects.Disk}
1924
  @param disk: the disk to be snapshotted
1925
  @rtype: string
1926
  @return: snapshot disk path
1927

1928
  """
1929
  if disk.children:
1930
    if len(disk.children) == 1:
1931
      # only one child, let's recurse on it
1932
      return BlockdevSnapshot(disk.children[0])
1933
    else:
1934
      # more than one child, choose one that matches
1935
      for child in disk.children:
1936
        if child.size == disk.size:
1937
          # return implies breaking the loop
1938
          return BlockdevSnapshot(child)
1939
  elif disk.dev_type == constants.LD_LV:
1940
    r_dev = _RecursiveFindBD(disk)
1941
    if r_dev is not None:
1942
      # let's stay on the safe side and ask for the full size, for now
1943
      return r_dev.Snapshot(disk.size)
1944
    else:
1945
      _Fail("Cannot find block device %s", disk)
1946
  else:
1947
    _Fail("Cannot snapshot non-lvm block device '%s' of type '%s'",
1948
          disk.unique_id, disk.dev_type)
1949

    
1950

    
1951
def ExportSnapshot(disk, dest_node, instance, cluster_name, idx):
1952
  """Export a block device snapshot to a remote node.
1953

1954
  @type disk: L{objects.Disk}
1955
  @param disk: the description of the disk to export
1956
  @type dest_node: str
1957
  @param dest_node: the destination node to export to
1958
  @type instance: L{objects.Instance}
1959
  @param instance: the instance object to whom the disk belongs
1960
  @type cluster_name: str
1961
  @param cluster_name: the cluster name, needed for SSH hostalias
1962
  @type idx: int
1963
  @param idx: the index of the disk in the instance's disk list,
1964
      used to export to the OS scripts environment
1965
  @rtype: None
1966

1967
  """
1968
  inst_os = OSFromDisk(instance.os)
1969
  export_env = OSEnvironment(instance, inst_os)
1970

    
1971
  export_script = inst_os.export_script
1972

    
1973
  logfile = "%s/exp-%s-%s-%s.log" % (constants.LOG_OS_DIR, inst_os.name,
1974
                                     instance.name, int(time.time()))
1975
  if not os.path.exists(constants.LOG_OS_DIR):
1976
    os.mkdir(constants.LOG_OS_DIR, 0750)
1977
  real_disk = _RecursiveFindBD(disk)
1978
  if real_disk is None:
1979
    _Fail("Block device '%s' is not set up", disk)
1980

    
1981
  real_disk.Open()
1982

    
1983
  export_env['EXPORT_DEVICE'] = real_disk.dev_path
1984
  export_env['EXPORT_INDEX'] = str(idx)
1985

    
1986
  destdir = os.path.join(constants.EXPORT_DIR, instance.name + ".new")
1987
  destfile = disk.physical_id[1]
1988

    
1989
  # the target command is built out of three individual commands,
1990
  # which are joined by pipes; we check each individual command for
1991
  # valid parameters
1992
  expcmd = utils.BuildShellCmd("set -e; set -o pipefail; cd %s; %s 2>%s",
1993
                               inst_os.path, export_script, logfile)
1994

    
1995
  comprcmd = "gzip"
1996

    
1997
  destcmd = utils.BuildShellCmd("mkdir -p %s && cat > %s/%s",
1998
                                destdir, destdir, destfile)
1999
  remotecmd = _GetSshRunner(cluster_name).BuildCmd(dest_node,
2000
                                                   constants.GANETI_RUNAS,
2001
                                                   destcmd)
2002

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

    
2006
  result = utils.RunCmd(["bash", "-c", command], env=export_env)
2007

    
2008
  if result.failed:
2009
    _Fail("OS snapshot export command '%s' returned error: %s"
2010
          " output: %s", command, result.fail_reason, result.output)
2011

    
2012

    
2013
def FinalizeExport(instance, snap_disks):
2014
  """Write out the export configuration information.
2015

2016
  @type instance: L{objects.Instance}
2017
  @param instance: the instance which we export, used for
2018
      saving configuration
2019
  @type snap_disks: list of L{objects.Disk}
2020
  @param snap_disks: list of snapshot block devices, which
2021
      will be used to get the actual name of the dump file
2022

2023
  @rtype: None
2024

2025
  """
2026
  destdir = os.path.join(constants.EXPORT_DIR, instance.name + ".new")
2027
  finaldestdir = os.path.join(constants.EXPORT_DIR, instance.name)
2028

    
2029
  config = objects.SerializableConfigParser()
2030

    
2031
  config.add_section(constants.INISECT_EXP)
2032
  config.set(constants.INISECT_EXP, 'version', '0')
2033
  config.set(constants.INISECT_EXP, 'timestamp', '%d' % int(time.time()))
2034
  config.set(constants.INISECT_EXP, 'source', instance.primary_node)
2035
  config.set(constants.INISECT_EXP, 'os', instance.os)
2036
  config.set(constants.INISECT_EXP, 'compression', 'gzip')
2037

    
2038
  config.add_section(constants.INISECT_INS)
2039
  config.set(constants.INISECT_INS, 'name', instance.name)
2040
  config.set(constants.INISECT_INS, 'memory', '%d' %
2041
             instance.beparams[constants.BE_MEMORY])
2042
  config.set(constants.INISECT_INS, 'vcpus', '%d' %
2043
             instance.beparams[constants.BE_VCPUS])
2044
  config.set(constants.INISECT_INS, 'disk_template', instance.disk_template)
2045

    
2046
  nic_total = 0
2047
  for nic_count, nic in enumerate(instance.nics):
2048
    nic_total += 1
2049
    config.set(constants.INISECT_INS, 'nic%d_mac' %
2050
               nic_count, '%s' % nic.mac)
2051
    config.set(constants.INISECT_INS, 'nic%d_ip' % nic_count, '%s' % nic.ip)
2052
    config.set(constants.INISECT_INS, 'nic%d_bridge' % nic_count,
2053
               '%s' % nic.bridge)
2054
  # TODO: redundant: on load can read nics until it doesn't exist
2055
  config.set(constants.INISECT_INS, 'nic_count' , '%d' % nic_total)
2056

    
2057
  disk_total = 0
2058
  for disk_count, disk in enumerate(snap_disks):
2059
    if disk:
2060
      disk_total += 1
2061
      config.set(constants.INISECT_INS, 'disk%d_ivname' % disk_count,
2062
                 ('%s' % disk.iv_name))
2063
      config.set(constants.INISECT_INS, 'disk%d_dump' % disk_count,
2064
                 ('%s' % disk.physical_id[1]))
2065
      config.set(constants.INISECT_INS, 'disk%d_size' % disk_count,
2066
                 ('%d' % disk.size))
2067

    
2068
  config.set(constants.INISECT_INS, 'disk_count' , '%d' % disk_total)
2069

    
2070
  utils.WriteFile(os.path.join(destdir, constants.EXPORT_CONF_FILE),
2071
                  data=config.Dumps())
2072
  shutil.rmtree(finaldestdir, True)
2073
  shutil.move(destdir, finaldestdir)
2074

    
2075

    
2076
def ExportInfo(dest):
2077
  """Get export configuration information.
2078

2079
  @type dest: str
2080
  @param dest: directory containing the export
2081

2082
  @rtype: L{objects.SerializableConfigParser}
2083
  @return: a serializable config file containing the
2084
      export info
2085

2086
  """
2087
  cff = os.path.join(dest, constants.EXPORT_CONF_FILE)
2088

    
2089
  config = objects.SerializableConfigParser()
2090
  config.read(cff)
2091

    
2092
  if (not config.has_section(constants.INISECT_EXP) or
2093
      not config.has_section(constants.INISECT_INS)):
2094
    _Fail("Export info file doesn't have the required fields")
2095

    
2096
  return config.Dumps()
2097

    
2098

    
2099
def ImportOSIntoInstance(instance, src_node, src_images, cluster_name):
2100
  """Import an os image into an instance.
2101

2102
  @type instance: L{objects.Instance}
2103
  @param instance: instance to import the disks into
2104
  @type src_node: string
2105
  @param src_node: source node for the disk images
2106
  @type src_images: list of string
2107
  @param src_images: absolute paths of the disk images
2108
  @rtype: list of boolean
2109
  @return: each boolean represent the success of importing the n-th disk
2110

2111
  """
2112
  inst_os = OSFromDisk(instance.os)
2113
  import_env = OSEnvironment(instance, inst_os)
2114
  import_script = inst_os.import_script
2115

    
2116
  logfile = "%s/import-%s-%s-%s.log" % (constants.LOG_OS_DIR, instance.os,
2117
                                        instance.name, int(time.time()))
2118
  if not os.path.exists(constants.LOG_OS_DIR):
2119
    os.mkdir(constants.LOG_OS_DIR, 0750)
2120

    
2121
  comprcmd = "gunzip"
2122
  impcmd = utils.BuildShellCmd("(cd %s; %s >%s 2>&1)", inst_os.path,
2123
                               import_script, logfile)
2124

    
2125
  final_result = []
2126
  for idx, image in enumerate(src_images):
2127
    if image:
2128
      destcmd = utils.BuildShellCmd('cat %s', image)
2129
      remotecmd = _GetSshRunner(cluster_name).BuildCmd(src_node,
2130
                                                       constants.GANETI_RUNAS,
2131
                                                       destcmd)
2132
      command = '|'.join([utils.ShellQuoteArgs(remotecmd), comprcmd, impcmd])
2133
      import_env['IMPORT_DEVICE'] = import_env['DISK_%d_PATH' % idx]
2134
      import_env['IMPORT_INDEX'] = str(idx)
2135
      result = utils.RunCmd(command, env=import_env)
2136
      if result.failed:
2137
        logging.error("Disk import command '%s' returned error: %s"
2138
                      " output: %s", command, result.fail_reason,
2139
                      result.output)
2140
        final_result.append("error importing disk %d: %s, %s" %
2141
                            (idx, result.fail_reason, result.output[-100]))
2142

    
2143
  if final_result:
2144
    _Fail("; ".join(final_result), log=False)
2145

    
2146

    
2147
def ListExports():
2148
  """Return a list of exports currently available on this machine.
2149

2150
  @rtype: list
2151
  @return: list of the exports
2152

2153
  """
2154
  if os.path.isdir(constants.EXPORT_DIR):
2155
    return utils.ListVisibleFiles(constants.EXPORT_DIR)
2156
  else:
2157
    _Fail("No exports directory")
2158

    
2159

    
2160
def RemoveExport(export):
2161
  """Remove an existing export from the node.
2162

2163
  @type export: str
2164
  @param export: the name of the export to remove
2165
  @rtype: None
2166

2167
  """
2168
  target = os.path.join(constants.EXPORT_DIR, export)
2169

    
2170
  try:
2171
    shutil.rmtree(target)
2172
  except EnvironmentError, err:
2173
    _Fail("Error while removing the export: %s", err, exc=True)
2174

    
2175

    
2176
def BlockdevRename(devlist):
2177
  """Rename a list of block devices.
2178

2179
  @type devlist: list of tuples
2180
  @param devlist: list of tuples of the form  (disk,
2181
      new_logical_id, new_physical_id); disk is an
2182
      L{objects.Disk} object describing the current disk,
2183
      and new logical_id/physical_id is the name we
2184
      rename it to
2185
  @rtype: boolean
2186
  @return: True if all renames succeeded, False otherwise
2187

2188
  """
2189
  msgs = []
2190
  result = True
2191
  for disk, unique_id in devlist:
2192
    dev = _RecursiveFindBD(disk)
2193
    if dev is None:
2194
      msgs.append("Can't find device %s in rename" % str(disk))
2195
      result = False
2196
      continue
2197
    try:
2198
      old_rpath = dev.dev_path
2199
      dev.Rename(unique_id)
2200
      new_rpath = dev.dev_path
2201
      if old_rpath != new_rpath:
2202
        DevCacheManager.RemoveCache(old_rpath)
2203
        # FIXME: we should add the new cache information here, like:
2204
        # DevCacheManager.UpdateCache(new_rpath, owner, ...)
2205
        # but we don't have the owner here - maybe parse from existing
2206
        # cache? for now, we only lose lvm data when we rename, which
2207
        # is less critical than DRBD or MD
2208
    except errors.BlockDeviceError, err:
2209
      msgs.append("Can't rename device '%s' to '%s': %s" %
2210
                  (dev, unique_id, err))
2211
      logging.exception("Can't rename device '%s' to '%s'", dev, unique_id)
2212
      result = False
2213
  if not result:
2214
    _Fail("; ".join(msgs))
2215

    
2216

    
2217
def _TransformFileStorageDir(file_storage_dir):
2218
  """Checks whether given file_storage_dir is valid.
2219

2220
  Checks wheter the given file_storage_dir is within the cluster-wide
2221
  default file_storage_dir stored in SimpleStore. Only paths under that
2222
  directory are allowed.
2223

2224
  @type file_storage_dir: str
2225
  @param file_storage_dir: the path to check
2226

2227
  @return: the normalized path if valid, None otherwise
2228

2229
  """
2230
  cfg = _GetConfig()
2231
  file_storage_dir = os.path.normpath(file_storage_dir)
2232
  base_file_storage_dir = cfg.GetFileStorageDir()
2233
  if (not os.path.commonprefix([file_storage_dir, base_file_storage_dir]) ==
2234
      base_file_storage_dir):
2235
    _Fail("File storage directory '%s' is not under base file"
2236
          " storage directory '%s'", file_storage_dir, base_file_storage_dir)
2237
  return file_storage_dir
2238

    
2239

    
2240
def CreateFileStorageDir(file_storage_dir):
2241
  """Create file storage directory.
2242

2243
  @type file_storage_dir: str
2244
  @param file_storage_dir: directory to create
2245

2246
  @rtype: tuple
2247
  @return: tuple with first element a boolean indicating wheter dir
2248
      creation was successful or not
2249

2250
  """
2251
  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
2252
  if os.path.exists(file_storage_dir):
2253
    if not os.path.isdir(file_storage_dir):
2254
      _Fail("Specified storage dir '%s' is not a directory",
2255
            file_storage_dir)
2256
  else:
2257
    try:
2258
      os.makedirs(file_storage_dir, 0750)
2259
    except OSError, err:
2260
      _Fail("Cannot create file storage directory '%s': %s",
2261
            file_storage_dir, err, exc=True)
2262

    
2263

    
2264
def RemoveFileStorageDir(file_storage_dir):
2265
  """Remove file storage directory.
2266

2267
  Remove it only if it's empty. If not log an error and return.
2268

2269
  @type file_storage_dir: str
2270
  @param file_storage_dir: the directory we should cleanup
2271
  @rtype: tuple (success,)
2272
  @return: tuple of one element, C{success}, denoting
2273
      whether the operation was successful
2274

2275
  """
2276
  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
2277
  if os.path.exists(file_storage_dir):
2278
    if not os.path.isdir(file_storage_dir):
2279
      _Fail("Specified Storage directory '%s' is not a directory",
2280
            file_storage_dir)
2281
    # deletes dir only if empty, otherwise we want to fail the rpc call
2282
    try:
2283
      os.rmdir(file_storage_dir)
2284
    except OSError, err:
2285
      _Fail("Cannot remove file storage directory '%s': %s",
2286
            file_storage_dir, err)
2287

    
2288

    
2289
def RenameFileStorageDir(old_file_storage_dir, new_file_storage_dir):
2290
  """Rename the file storage directory.
2291

2292
  @type old_file_storage_dir: str
2293
  @param old_file_storage_dir: the current path
2294
  @type new_file_storage_dir: str
2295
  @param new_file_storage_dir: the name we should rename to
2296
  @rtype: tuple (success,)
2297
  @return: tuple of one element, C{success}, denoting
2298
      whether the operation was successful
2299

2300
  """
2301
  old_file_storage_dir = _TransformFileStorageDir(old_file_storage_dir)
2302
  new_file_storage_dir = _TransformFileStorageDir(new_file_storage_dir)
2303
  if not os.path.exists(new_file_storage_dir):
2304
    if os.path.isdir(old_file_storage_dir):
2305
      try:
2306
        os.rename(old_file_storage_dir, new_file_storage_dir)
2307
      except OSError, err:
2308
        _Fail("Cannot rename '%s' to '%s': %s",
2309
              old_file_storage_dir, new_file_storage_dir, err)
2310
    else:
2311
      _Fail("Specified storage dir '%s' is not a directory",
2312
            old_file_storage_dir)
2313
  else:
2314
    if os.path.exists(old_file_storage_dir):
2315
      _Fail("Cannot rename '%s' to '%s': both locations exist",
2316
            old_file_storage_dir, new_file_storage_dir)
2317

    
2318

    
2319
def _EnsureJobQueueFile(file_name):
2320
  """Checks whether the given filename is in the queue directory.
2321

2322
  @type file_name: str
2323
  @param file_name: the file name we should check
2324
  @rtype: None
2325
  @raises RPCFail: if the file is not valid
2326

2327
  """
2328
  queue_dir = os.path.normpath(constants.QUEUE_DIR)
2329
  result = (os.path.commonprefix([queue_dir, file_name]) == queue_dir)
2330

    
2331
  if not result:
2332
    _Fail("Passed job queue file '%s' does not belong to"
2333
          " the queue directory '%s'", file_name, queue_dir)
2334

    
2335

    
2336
def JobQueueUpdate(file_name, content):
2337
  """Updates a file in the queue directory.
2338

2339
  This is just a wrapper over L{utils.WriteFile}, with proper
2340
  checking.
2341

2342
  @type file_name: str
2343
  @param file_name: the job file name
2344
  @type content: str
2345
  @param content: the new job contents
2346
  @rtype: boolean
2347
  @return: the success of the operation
2348

2349
  """
2350
  _EnsureJobQueueFile(file_name)
2351

    
2352
  # Write and replace the file atomically
2353
  utils.WriteFile(file_name, data=_Decompress(content))
2354

    
2355

    
2356
def JobQueueRename(old, new):
2357
  """Renames a job queue file.
2358

2359
  This is just a wrapper over os.rename with proper checking.
2360

2361
  @type old: str
2362
  @param old: the old (actual) file name
2363
  @type new: str
2364
  @param new: the desired file name
2365
  @rtype: tuple
2366
  @return: the success of the operation and payload
2367

2368
  """
2369
  _EnsureJobQueueFile(old)
2370
  _EnsureJobQueueFile(new)
2371

    
2372
  utils.RenameFile(old, new, mkdir=True)
2373

    
2374

    
2375
def JobQueueSetDrainFlag(drain_flag):
2376
  """Set the drain flag for the queue.
2377

2378
  This will set or unset the queue drain flag.
2379

2380
  @type drain_flag: boolean
2381
  @param drain_flag: if True, will set the drain flag, otherwise reset it.
2382
  @rtype: truple
2383
  @return: always True, None
2384
  @warning: the function always returns True
2385

2386
  """
2387
  if drain_flag:
2388
    utils.WriteFile(constants.JOB_QUEUE_DRAIN_FILE, data="", close=True)
2389
  else:
2390
    utils.RemoveFile(constants.JOB_QUEUE_DRAIN_FILE)
2391

    
2392

    
2393
def BlockdevClose(instance_name, disks):
2394
  """Closes the given block devices.
2395

2396
  This means they will be switched to secondary mode (in case of
2397
  DRBD).
2398

2399
  @param instance_name: if the argument is not empty, the symlinks
2400
      of this instance will be removed
2401
  @type disks: list of L{objects.Disk}
2402
  @param disks: the list of disks to be closed
2403
  @rtype: tuple (success, message)
2404
  @return: a tuple of success and message, where success
2405
      indicates the succes of the operation, and message
2406
      which will contain the error details in case we
2407
      failed
2408

2409
  """
2410
  bdevs = []
2411
  for cf in disks:
2412
    rd = _RecursiveFindBD(cf)
2413
    if rd is None:
2414
      _Fail("Can't find device %s", cf)
2415
    bdevs.append(rd)
2416

    
2417
  msg = []
2418
  for rd in bdevs:
2419
    try:
2420
      rd.Close()
2421
    except errors.BlockDeviceError, err:
2422
      msg.append(str(err))
2423
  if msg:
2424
    _Fail("Can't make devices secondary: %s", ",".join(msg))
2425
  else:
2426
    if instance_name:
2427
      _RemoveBlockDevLinks(instance_name, disks)
2428

    
2429

    
2430
def ValidateHVParams(hvname, hvparams):
2431
  """Validates the given hypervisor parameters.
2432

2433
  @type hvname: string
2434
  @param hvname: the hypervisor name
2435
  @type hvparams: dict
2436
  @param hvparams: the hypervisor parameters to be validated
2437
  @rtype: None
2438

2439
  """
2440
  try:
2441
    hv_type = hypervisor.GetHypervisor(hvname)
2442
    hv_type.ValidateParameters(hvparams)
2443
  except errors.HypervisorError, err:
2444
    _Fail(str(err), log=False)
2445

    
2446

    
2447
def DemoteFromMC():
2448
  """Demotes the current node from master candidate role.
2449

2450
  """
2451
  # try to ensure we're not the master by mistake
2452
  master, myself = ssconf.GetMasterAndMyself()
2453
  if master == myself:
2454
    _Fail("ssconf status shows I'm the master node, will not demote")
2455

    
2456
  result = utils.RunCmd([constants.DAEMON_UTIL, "check", constants.MASTERD])
2457
  if not result.failed:
2458
    _Fail("The master daemon is running, will not demote")
2459

    
2460
  try:
2461
    if os.path.isfile(constants.CLUSTER_CONF_FILE):
2462
      utils.CreateBackup(constants.CLUSTER_CONF_FILE)
2463
  except EnvironmentError, err:
2464
    if err.errno != errno.ENOENT:
2465
      _Fail("Error while backing up cluster file: %s", err, exc=True)
2466

    
2467
  utils.RemoveFile(constants.CLUSTER_CONF_FILE)
2468

    
2469

    
2470
def _FindDisks(nodes_ip, disks):
2471
  """Sets the physical ID on disks and returns the block devices.
2472

2473
  """
2474
  # set the correct physical ID
2475
  my_name = utils.HostInfo().name
2476
  for cf in disks:
2477
    cf.SetPhysicalID(my_name, nodes_ip)
2478

    
2479
  bdevs = []
2480

    
2481
  for cf in disks:
2482
    rd = _RecursiveFindBD(cf)
2483
    if rd is None:
2484
      _Fail("Can't find device %s", cf)
2485
    bdevs.append(rd)
2486
  return bdevs
2487

    
2488

    
2489
def DrbdDisconnectNet(nodes_ip, disks):
2490
  """Disconnects the network on a list of drbd devices.
2491

2492
  """
2493
  bdevs = _FindDisks(nodes_ip, disks)
2494

    
2495
  # disconnect disks
2496
  for rd in bdevs:
2497
    try:
2498
      rd.DisconnectNet()
2499
    except errors.BlockDeviceError, err:
2500
      _Fail("Can't change network configuration to standalone mode: %s",
2501
            err, exc=True)
2502

    
2503

    
2504
def DrbdAttachNet(nodes_ip, disks, instance_name, multimaster):
2505
  """Attaches the network on a list of drbd devices.
2506

2507
  """
2508
  bdevs = _FindDisks(nodes_ip, disks)
2509

    
2510
  if multimaster:
2511
    for idx, rd in enumerate(bdevs):
2512
      try:
2513
        _SymlinkBlockDev(instance_name, rd.dev_path, idx)
2514
      except EnvironmentError, err:
2515
        _Fail("Can't create symlink: %s", err)
2516
  # reconnect disks, switch to new master configuration and if
2517
  # needed primary mode
2518
  for rd in bdevs:
2519
    try:
2520
      rd.AttachNet(multimaster)
2521
    except errors.BlockDeviceError, err:
2522
      _Fail("Can't change network configuration: %s", err)
2523

    
2524
  # wait until the disks are connected; we need to retry the re-attach
2525
  # if the device becomes standalone, as this might happen if the one
2526
  # node disconnects and reconnects in a different mode before the
2527
  # other node reconnects; in this case, one or both of the nodes will
2528
  # decide it has wrong configuration and switch to standalone
2529

    
2530
  def _Attach():
2531
    all_connected = True
2532

    
2533
    for rd in bdevs:
2534
      stats = rd.GetProcStatus()
2535

    
2536
      all_connected = (all_connected and
2537
                       (stats.is_connected or stats.is_in_resync))
2538

    
2539
      if stats.is_standalone:
2540
        # peer had different config info and this node became
2541
        # standalone, even though this should not happen with the
2542
        # new staged way of changing disk configs
2543
        try:
2544
          rd.AttachNet(multimaster)
2545
        except errors.BlockDeviceError, err:
2546
          _Fail("Can't change network configuration: %s", err)
2547

    
2548
    if not all_connected:
2549
      raise utils.RetryAgain()
2550

    
2551
  try:
2552
    # Start with a delay of 100 miliseconds and go up to 5 seconds
2553
    utils.Retry(_Attach, (0.1, 1.5, 5.0), 2 * 60)
2554
  except utils.RetryTimeout:
2555
    _Fail("Timeout in disk reconnecting")
2556

    
2557
  if multimaster:
2558
    # change to primary mode
2559
    for rd in bdevs:
2560
      try:
2561
        rd.Open()
2562
      except errors.BlockDeviceError, err:
2563
        _Fail("Can't change to primary mode: %s", err)
2564

    
2565

    
2566
def DrbdWaitSync(nodes_ip, disks):
2567
  """Wait until DRBDs have synchronized.
2568

2569
  """
2570
  def _helper(rd):
2571
    stats = rd.GetProcStatus()
2572
    if not (stats.is_connected or stats.is_in_resync):
2573
      raise utils.RetryAgain()
2574
    return stats
2575

    
2576
  bdevs = _FindDisks(nodes_ip, disks)
2577

    
2578
  min_resync = 100
2579
  alldone = True
2580
  for rd in bdevs:
2581
    try:
2582
      # poll each second for 15 seconds
2583
      stats = utils.Retry(_helper, 1, 15, args=[rd])
2584
    except utils.RetryTimeout:
2585
      stats = rd.GetProcStatus()
2586
      # last check
2587
      if not (stats.is_connected or stats.is_in_resync):
2588
        _Fail("DRBD device %s is not in sync: stats=%s", rd, stats)
2589
    alldone = alldone and (not stats.is_in_resync)
2590
    if stats.sync_percent is not None:
2591
      min_resync = min(min_resync, stats.sync_percent)
2592

    
2593
  return (alldone, min_resync)
2594

    
2595

    
2596
def PowercycleNode(hypervisor_type):
2597
  """Hard-powercycle the node.
2598

2599
  Because we need to return first, and schedule the powercycle in the
2600
  background, we won't be able to report failures nicely.
2601

2602
  """
2603
  hyper = hypervisor.GetHypervisor(hypervisor_type)
2604
  try:
2605
    pid = os.fork()
2606
  except OSError:
2607
    # if we can't fork, we'll pretend that we're in the child process
2608
    pid = 0
2609
  if pid > 0:
2610
    return "Reboot scheduled in 5 seconds"
2611
  time.sleep(5)
2612
  hyper.PowercycleNode()
2613

    
2614

    
2615
class HooksRunner(object):
2616
  """Hook runner.
2617

2618
  This class is instantiated on the node side (ganeti-noded) and not
2619
  on the master side.
2620

2621
  """
2622
  def __init__(self, hooks_base_dir=None):
2623
    """Constructor for hooks runner.
2624

2625
    @type hooks_base_dir: str or None
2626
    @param hooks_base_dir: if not None, this overrides the
2627
        L{constants.HOOKS_BASE_DIR} (useful for unittests)
2628

2629
    """
2630
    if hooks_base_dir is None:
2631
      hooks_base_dir = constants.HOOKS_BASE_DIR
2632
    # yeah, _BASE_DIR is not valid for attributes, we use it like a
2633
    # constant
2634
    self._BASE_DIR = hooks_base_dir # pylint: disable-msg=C0103
2635

    
2636
  @staticmethod
2637
  def ExecHook(script, env):
2638
    """Exec one hook script.
2639

2640
    @type script: str
2641
    @param script: the full path to the script
2642
    @type env: dict
2643
    @param env: the environment with which to exec the script
2644
    @rtype: tuple (success, message)
2645
    @return: a tuple of success and message, where success
2646
        indicates the succes of the operation, and message
2647
        which will contain the error details in case we
2648
        failed
2649

2650
    """
2651
    # exec the process using subprocess and log the output
2652
    fdstdin = None
2653
    try:
2654
      fdstdin = open("/dev/null", "r")
2655
      child = subprocess.Popen([script], stdin=fdstdin, stdout=subprocess.PIPE,
2656
                               stderr=subprocess.STDOUT, close_fds=True,
2657
                               shell=False, cwd="/", env=env)
2658
      output = ""
2659
      try:
2660
        output = child.stdout.read(4096)
2661
        child.stdout.close()
2662
      except EnvironmentError, err:
2663
        output += "Hook script error: %s" % str(err)
2664

    
2665
      while True:
2666
        try:
2667
          result = child.wait()
2668
          break
2669
        except EnvironmentError, err:
2670
          if err.errno == errno.EINTR:
2671
            continue
2672
          raise
2673
    finally:
2674
      # try not to leak fds
2675
      for fd in (fdstdin, ):
2676
        if fd is not None:
2677
          try:
2678
            fd.close()
2679
          except EnvironmentError, err:
2680
            # just log the error
2681
            #logging.exception("Error while closing fd %s", fd)
2682
            pass
2683

    
2684
    return result == 0, utils.SafeEncode(output.strip())
2685

    
2686
  def RunHooks(self, hpath, phase, env):
2687
    """Run the scripts in the hooks directory.
2688

2689
    @type hpath: str
2690
    @param hpath: the path to the hooks directory which
2691
        holds the scripts
2692
    @type phase: str
2693
    @param phase: either L{constants.HOOKS_PHASE_PRE} or
2694
        L{constants.HOOKS_PHASE_POST}
2695
    @type env: dict
2696
    @param env: dictionary with the environment for the hook
2697
    @rtype: list
2698
    @return: list of 3-element tuples:
2699
      - script path
2700
      - script result, either L{constants.HKR_SUCCESS} or
2701
        L{constants.HKR_FAIL}
2702
      - output of the script
2703

2704
    @raise errors.ProgrammerError: for invalid input
2705
        parameters
2706

2707
    """
2708
    if phase == constants.HOOKS_PHASE_PRE:
2709
      suffix = "pre"
2710
    elif phase == constants.HOOKS_PHASE_POST:
2711
      suffix = "post"
2712
    else:
2713
      _Fail("Unknown hooks phase '%s'", phase)
2714

    
2715
    rr = []
2716

    
2717
    subdir = "%s-%s.d" % (hpath, suffix)
2718
    dir_name = "%s/%s" % (self._BASE_DIR, subdir)
2719
    try:
2720
      dir_contents = utils.ListVisibleFiles(dir_name)
2721
    except OSError:
2722
      # FIXME: must log output in case of failures
2723
      return rr
2724

    
2725
    # we use the standard python sort order,
2726
    # so 00name is the recommended naming scheme
2727
    dir_contents.sort()
2728
    for relname in dir_contents:
2729
      fname = os.path.join(dir_name, relname)
2730
      if not (os.path.isfile(fname) and os.access(fname, os.X_OK) and
2731
              constants.EXT_PLUGIN_MASK.match(relname) is not None):
2732
        rrval = constants.HKR_SKIP
2733
        output = ""
2734
      else:
2735
        result, output = self.ExecHook(fname, env)
2736
        if not result:
2737
          rrval = constants.HKR_FAIL
2738
        else:
2739
          rrval = constants.HKR_SUCCESS
2740
      rr.append(("%s/%s" % (subdir, relname), rrval, output))
2741

    
2742
    return rr
2743

    
2744

    
2745
class IAllocatorRunner(object):
2746
  """IAllocator runner.
2747

2748
  This class is instantiated on the node side (ganeti-noded) and not on
2749
  the master side.
2750

2751
  """
2752
  @staticmethod
2753
  def Run(name, idata):
2754
    """Run an iallocator script.
2755

2756
    @type name: str
2757
    @param name: the iallocator script name
2758
    @type idata: str
2759
    @param idata: the allocator input data
2760

2761
    @rtype: tuple
2762
    @return: two element tuple of:
2763
       - status
2764
       - either error message or stdout of allocator (for success)
2765

2766
    """
2767
    alloc_script = utils.FindFile(name, constants.IALLOCATOR_SEARCH_PATH,
2768
                                  os.path.isfile)
2769
    if alloc_script is None:
2770
      _Fail("iallocator module '%s' not found in the search path", name)
2771

    
2772
    fd, fin_name = tempfile.mkstemp(prefix="ganeti-iallocator.")
2773
    try:
2774
      os.write(fd, idata)
2775
      os.close(fd)
2776
      result = utils.RunCmd([alloc_script, fin_name])
2777
      if result.failed:
2778
        _Fail("iallocator module '%s' failed: %s, output '%s'",
2779
              name, result.fail_reason, result.output)
2780
    finally:
2781
      os.unlink(fin_name)
2782

    
2783
    return result.stdout
2784

    
2785

    
2786
class DevCacheManager(object):
2787
  """Simple class for managing a cache of block device information.
2788

2789
  """
2790
  _DEV_PREFIX = "/dev/"
2791
  _ROOT_DIR = constants.BDEV_CACHE_DIR
2792

    
2793
  @classmethod
2794
  def _ConvertPath(cls, dev_path):
2795
    """Converts a /dev/name path to the cache file name.
2796

2797
    This replaces slashes with underscores and strips the /dev
2798
    prefix. It then returns the full path to the cache file.
2799

2800
    @type dev_path: str
2801
    @param dev_path: the C{/dev/} path name
2802
    @rtype: str
2803
    @return: the converted path name
2804

2805
    """
2806
    if dev_path.startswith(cls._DEV_PREFIX):
2807
      dev_path = dev_path[len(cls._DEV_PREFIX):]
2808
    dev_path = dev_path.replace("/", "_")
2809
    fpath = "%s/bdev_%s" % (cls._ROOT_DIR, dev_path)
2810
    return fpath
2811

    
2812
  @classmethod
2813
  def UpdateCache(cls, dev_path, owner, on_primary, iv_name):
2814
    """Updates the cache information for a given device.
2815

2816
    @type dev_path: str
2817
    @param dev_path: the pathname of the device
2818
    @type owner: str
2819
    @param owner: the owner (instance name) of the device
2820
    @type on_primary: bool
2821
    @param on_primary: whether this is the primary
2822
        node nor not
2823
    @type iv_name: str
2824
    @param iv_name: the instance-visible name of the
2825
        device, as in objects.Disk.iv_name
2826

2827
    @rtype: None
2828

2829
    """
2830
    if dev_path is None:
2831
      logging.error("DevCacheManager.UpdateCache got a None dev_path")
2832
      return
2833
    fpath = cls._ConvertPath(dev_path)
2834
    if on_primary:
2835
      state = "primary"
2836
    else:
2837
      state = "secondary"
2838
    if iv_name is None:
2839
      iv_name = "not_visible"
2840
    fdata = "%s %s %s\n" % (str(owner), state, iv_name)
2841
    try:
2842
      utils.WriteFile(fpath, data=fdata)
2843
    except EnvironmentError, err:
2844
      logging.exception("Can't update bdev cache for %s: %s", dev_path, err)
2845

    
2846
  @classmethod
2847
  def RemoveCache(cls, dev_path):
2848
    """Remove data for a dev_path.
2849

2850
    This is just a wrapper over L{utils.RemoveFile} with a converted
2851
    path name and logging.
2852

2853
    @type dev_path: str
2854
    @param dev_path: the pathname of the device
2855

2856
    @rtype: None
2857

2858
    """
2859
    if dev_path is None:
2860
      logging.error("DevCacheManager.RemoveCache got a None dev_path")
2861
      return
2862
    fpath = cls._ConvertPath(dev_path)
2863
    try:
2864
      utils.RemoveFile(fpath)
2865
    except EnvironmentError, err:
2866
      logging.exception("Can't update bdev cache for %s: %s", dev_path, err)