Statistics
| Branch: | Tag: | Revision:

root / lib / backend.py @ 96e0d5cc

History | View | Annotate | Download (110.9 kB)

1
#
2
#
3

    
4
# Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011 Google Inc.
5
#
6
# This program is free software; you can redistribute it and/or modify
7
# it under the terms of the GNU General Public License as published by
8
# the Free Software Foundation; either version 2 of the License, or
9
# (at your option) any later version.
10
#
11
# This program is distributed in the hope that it will be useful, but
12
# WITHOUT ANY WARRANTY; without even the implied warranty of
13
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
14
# General Public License for more details.
15
#
16
# You should have received a copy of the GNU General Public License
17
# along with this program; if not, write to the Free Software
18
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
19
# 02110-1301, USA.
20

    
21

    
22
"""Functions used by the node daemon
23

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

29
"""
30

    
31
# pylint: disable=E1103
32

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

    
37

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

    
52
from ganeti import errors
53
from ganeti import utils
54
from ganeti import ssh
55
from ganeti import hypervisor
56
from ganeti import constants
57
from ganeti import bdev
58
from ganeti import objects
59
from ganeti import ssconf
60
from ganeti import serializer
61
from ganeti import netutils
62
from ganeti import runtime
63
from ganeti import mcpu
64

    
65

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

    
80
#: Valid LVS output line regex
81
_LVSLINE_REGEX = re.compile("^ *([^|]+)\|([^|]+)\|([0-9.]+)\|([^|]{6})\|?$")
82

    
83

    
84
class RPCFail(Exception):
85
  """Class denoting RPC failure.
86

87
  Its argument is the error message.
88

89
  """
90

    
91

    
92
def _Fail(msg, *args, **kwargs):
93
  """Log an error and the raise an RPCFail exception.
94

95
  This exception is then handled specially in the ganeti daemon and
96
  turned into a 'failed' return type. As such, this function is a
97
  useful shortcut for logging the error and returning it to the master
98
  daemon.
99

100
  @type msg: string
101
  @param msg: the text of the exception
102
  @raise RPCFail
103

104
  """
105
  if args:
106
    msg = msg % args
107
  if "log" not in kwargs or kwargs["log"]: # if we should log this error
108
    if "exc" in kwargs and kwargs["exc"]:
109
      logging.exception(msg)
110
    else:
111
      logging.error(msg)
112
  raise RPCFail(msg)
113

    
114

    
115
def _GetConfig():
116
  """Simple wrapper to return a SimpleStore.
117

118
  @rtype: L{ssconf.SimpleStore}
119
  @return: a SimpleStore instance
120

121
  """
122
  return ssconf.SimpleStore()
123

    
124

    
125
def _GetSshRunner(cluster_name):
126
  """Simple wrapper to return an SshRunner.
127

128
  @type cluster_name: str
129
  @param cluster_name: the cluster name, which is needed
130
      by the SshRunner constructor
131
  @rtype: L{ssh.SshRunner}
132
  @return: an SshRunner instance
133

134
  """
135
  return ssh.SshRunner(cluster_name)
136

    
137

    
138
def _Decompress(data):
139
  """Unpacks data compressed by the RPC client.
140

141
  @type data: list or tuple
142
  @param data: Data sent by RPC client
143
  @rtype: str
144
  @return: Decompressed data
145

146
  """
147
  assert isinstance(data, (list, tuple))
148
  assert len(data) == 2
149
  (encoding, content) = data
150
  if encoding == constants.RPC_ENCODING_NONE:
151
    return content
152
  elif encoding == constants.RPC_ENCODING_ZLIB_BASE64:
153
    return zlib.decompress(base64.b64decode(content))
154
  else:
155
    raise AssertionError("Unknown data encoding")
156

    
157

    
158
def _CleanDirectory(path, exclude=None):
159
  """Removes all regular files in a directory.
160

161
  @type path: str
162
  @param path: the directory to clean
163
  @type exclude: list
164
  @param exclude: list of files to be excluded, defaults
165
      to the empty list
166

167
  """
168
  if path not in _ALLOWED_CLEAN_DIRS:
169
    _Fail("Path passed to _CleanDirectory not in allowed clean targets: '%s'",
170
          path)
171

    
172
  if not os.path.isdir(path):
173
    return
174
  if exclude is None:
175
    exclude = []
176
  else:
177
    # Normalize excluded paths
178
    exclude = [os.path.normpath(i) for i in exclude]
179

    
180
  for rel_name in utils.ListVisibleFiles(path):
181
    full_name = utils.PathJoin(path, rel_name)
182
    if full_name in exclude:
183
      continue
184
    if os.path.isfile(full_name) and not os.path.islink(full_name):
185
      utils.RemoveFile(full_name)
186

    
187

    
188
def _BuildUploadFileList():
189
  """Build the list of allowed upload files.
190

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

193
  """
194
  allowed_files = set([
195
    constants.CLUSTER_CONF_FILE,
196
    constants.ETC_HOSTS,
197
    constants.SSH_KNOWN_HOSTS_FILE,
198
    constants.VNC_PASSWORD_FILE,
199
    constants.RAPI_CERT_FILE,
200
    constants.SPICE_CERT_FILE,
201
    constants.SPICE_CACERT_FILE,
202
    constants.RAPI_USERS_FILE,
203
    constants.CONFD_HMAC_KEY,
204
    constants.CLUSTER_DOMAIN_SECRET_FILE,
205
    ])
206

    
207
  for hv_name in constants.HYPER_TYPES:
208
    hv_class = hypervisor.GetHypervisorClass(hv_name)
209
    allowed_files.update(hv_class.GetAncillaryFiles()[0])
210

    
211
  return frozenset(allowed_files)
212

    
213

    
214
_ALLOWED_UPLOAD_FILES = _BuildUploadFileList()
215

    
216

    
217
def JobQueuePurge():
218
  """Removes job queue files and archived jobs.
219

220
  @rtype: tuple
221
  @return: True, None
222

223
  """
224
  _CleanDirectory(constants.QUEUE_DIR, exclude=[constants.JOB_QUEUE_LOCK_FILE])
225
  _CleanDirectory(constants.JOB_QUEUE_ARCHIVE_DIR)
226

    
227

    
228
def GetMasterInfo():
229
  """Returns master information.
230

231
  This is an utility function to compute master information, either
232
  for consumption here or from the node daemon.
233

234
  @rtype: tuple
235
  @return: master_netdev, master_ip, master_name, primary_ip_family,
236
    master_netmask
237
  @raise RPCFail: in case of errors
238

239
  """
240
  try:
241
    cfg = _GetConfig()
242
    master_netdev = cfg.GetMasterNetdev()
243
    master_ip = cfg.GetMasterIP()
244
    master_netmask = cfg.GetMasterNetmask()
245
    master_node = cfg.GetMasterNode()
246
    primary_ip_family = cfg.GetPrimaryIPFamily()
247
  except errors.ConfigurationError, err:
248
    _Fail("Cluster configuration incomplete: %s", err, exc=True)
249
  return (master_netdev, master_ip, master_node, primary_ip_family,
250
      master_netmask)
251

    
252

    
253
def RunLocalHooks(hook_opcode, hooks_path, env_builder_fn):
254
  """Decorator that runs hooks before and after the decorated function.
255

256
  @type hook_opcode: string
257
  @param hook_opcode: opcode of the hook
258
  @type hooks_path: string
259
  @param hooks_path: path of the hooks
260
  @type env_builder_fn: function
261
  @param env_builder_fn: function that returns a dictionary containing the
262
    environment variables for the hooks.
263
  @raise RPCFail: in case of pre-hook failure
264

265
  """
266
  def decorator(fn):
267
    def wrapper(*args, **kwargs):
268
      _, myself = ssconf.GetMasterAndMyself()
269
      nodes = ([myself], [myself])  # these hooks run locally
270

    
271
      cfg = _GetConfig()
272
      hr = HooksRunner()
273
      hm = mcpu.HooksMaster(hook_opcode, hooks_path, nodes, hr.RunLocalHooks,
274
                            None, env_builder_fn, logging.warning,
275
                            cfg.GetClusterName(), cfg.GetMasterNode())
276

    
277
      hm.RunPhase(constants.HOOKS_PHASE_PRE)
278
      result = fn(*args, **kwargs)
279
      hm.RunPhase(constants.HOOKS_PHASE_POST)
280

    
281
      return result
282
    return wrapper
283
  return decorator
284

    
285

    
286
def _BuildMasterIpHookEnv():
287
  """Builds environment variables for master IP hooks.
288

289
  """
290
  cfg = _GetConfig()
291
  env = {
292
    "MASTER_NETDEV": cfg.GetMasterNetdev(),
293
    "MASTER_IP": cfg.GetMasterIP(),
294
  }
295

    
296
  return env
297

    
298

    
299
@RunLocalHooks(constants.FAKE_OP_MASTER_TURNUP, "master-ip-turnup",
300
               _BuildMasterIpHookEnv)
301
def ActivateMasterIp(master_ip, master_netmask, master_netdev, family):
302
  """Activate the IP address of the master daemon.
303

304
  @param master_ip: the master IP
305
  @param master_netmask: the master IP netmask
306
  @param master_netdev: the master network device
307
  @param family: the IP family
308

309
  """
310
  # GetMasterInfo will raise an exception if not able to return data
311
  master_netdev, master_ip, _, family, master_netmask = GetMasterInfo()
312

    
313
  err_msg = None
314
  if netutils.TcpPing(master_ip, constants.DEFAULT_NODED_PORT):
315
    if netutils.IPAddress.Own(master_ip):
316
      # we already have the ip:
317
      logging.debug("Master IP already configured, doing nothing")
318
    else:
319
      err_msg = "Someone else has the master ip, not activating"
320
      logging.error(err_msg)
321
  else:
322
    ipcls = netutils.IPAddress.GetClassFromIpFamily(family)
323

    
324
    result = utils.RunCmd([constants.IP_COMMAND_PATH, "address", "add",
325
                           "%s/%s" % (master_ip, master_netmask),
326
                           "dev", master_netdev, "label",
327
                           "%s:0" % master_netdev])
328
    if result.failed:
329
      err_msg = "Can't activate master IP: %s" % result.output
330
      logging.error(err_msg)
331

    
332
    else:
333
      # we ignore the exit code of the following cmds
334
      if ipcls == netutils.IP4Address:
335
        utils.RunCmd(["arping", "-q", "-U", "-c 3", "-I", master_netdev, "-s",
336
                      master_ip, master_ip])
337
      elif ipcls == netutils.IP6Address:
338
        try:
339
          utils.RunCmd(["ndisc6", "-q", "-r 3", master_ip, master_netdev])
340
        except errors.OpExecError:
341
          # TODO: Better error reporting
342
          logging.warning("Can't execute ndisc6, please install if missing")
343

    
344
  if err_msg:
345
    _Fail(err_msg)
346

    
347

    
348
def StartMasterDaemons(no_voting):
349
  """Activate local node as master node.
350

351
  The function will start the master daemons (ganeti-masterd and ganeti-rapi).
352

353
  @type no_voting: boolean
354
  @param no_voting: whether to start ganeti-masterd without a node vote
355
      but still non-interactively
356
  @rtype: None
357

358
  """
359

    
360
  if no_voting:
361
    masterd_args = "--no-voting --yes-do-it"
362
  else:
363
    masterd_args = ""
364

    
365
  env = {
366
    "EXTRA_MASTERD_ARGS": masterd_args,
367
    }
368

    
369
  result = utils.RunCmd([constants.DAEMON_UTIL, "start-master"], env=env)
370
  if result.failed:
371
    msg = "Can't start Ganeti master: %s" % result.output
372
    logging.error(msg)
373
    _Fail(msg)
374

    
375

    
376
@RunLocalHooks(constants.FAKE_OP_MASTER_TURNDOWN, "master-ip-turndown",
377
               _BuildMasterIpHookEnv)
378
def DeactivateMasterIp(master_ip, master_netmask, master_netdev):
379
  """Deactivate the master IP on this node.
380

381
  @param master_ip: the master IP
382
  @param master_netmask: the master IP netmask
383
  @param master_netdev: the master network device
384

385
  """
386
  # TODO: log and report back to the caller the error failures; we
387
  # need to decide in which case we fail the RPC for this
388

    
389
  result = utils.RunCmd([constants.IP_COMMAND_PATH, "address", "del",
390
                         "%s/%s" % (master_ip, master_netmask),
391
                         "dev", master_netdev])
392
  if result.failed:
393
    logging.error("Can't remove the master IP, error: %s", result.output)
394
    # but otherwise ignore the failure
395

    
396

    
397
def StopMasterDaemons():
398
  """Stop the master daemons on this node.
399

400
  Stop the master daemons (ganeti-masterd and ganeti-rapi) on this node.
401

402
  @rtype: None
403

404
  """
405
  # TODO: log and report back to the caller the error failures; we
406
  # need to decide in which case we fail the RPC for this
407

    
408
  result = utils.RunCmd([constants.DAEMON_UTIL, "stop-master"])
409
  if result.failed:
410
    logging.error("Could not stop Ganeti master, command %s had exitcode %s"
411
                  " and error %s",
412
                  result.cmd, result.exit_code, result.output)
413

    
414

    
415
def ChangeMasterNetmask(netmask):
416
  """Change the netmask of the master IP.
417

418
  """
419
  master_netdev, master_ip, _, _, old_netmask = GetMasterInfo()
420
  if old_netmask == netmask:
421
    return
422

    
423
  result = utils.RunCmd([constants.IP_COMMAND_PATH, "address", "add",
424
                         "%s/%s" % (master_ip, netmask),
425
                         "dev", master_netdev, "label",
426
                         "%s:0" % master_netdev])
427
  if result.failed:
428
    _Fail("Could not change the master IP netmask")
429

    
430
  result = utils.RunCmd([constants.IP_COMMAND_PATH, "address", "del",
431
                         "%s/%s" % (master_ip, old_netmask),
432
                         "dev", master_netdev, "label",
433
                         "%s:0" % master_netdev])
434
  if result.failed:
435
    _Fail("Could not change the master IP netmask")
436

    
437

    
438
def EtcHostsModify(mode, host, ip):
439
  """Modify a host entry in /etc/hosts.
440

441
  @param mode: The mode to operate. Either add or remove entry
442
  @param host: The host to operate on
443
  @param ip: The ip associated with the entry
444

445
  """
446
  if mode == constants.ETC_HOSTS_ADD:
447
    if not ip:
448
      RPCFail("Mode 'add' needs 'ip' parameter, but parameter not"
449
              " present")
450
    utils.AddHostToEtcHosts(host, ip)
451
  elif mode == constants.ETC_HOSTS_REMOVE:
452
    if ip:
453
      RPCFail("Mode 'remove' does not allow 'ip' parameter, but"
454
              " parameter is present")
455
    utils.RemoveHostFromEtcHosts(host)
456
  else:
457
    RPCFail("Mode not supported")
458

    
459

    
460
def LeaveCluster(modify_ssh_setup):
461
  """Cleans up and remove the current node.
462

463
  This function cleans up and prepares the current node to be removed
464
  from the cluster.
465

466
  If processing is successful, then it raises an
467
  L{errors.QuitGanetiException} which is used as a special case to
468
  shutdown the node daemon.
469

470
  @param modify_ssh_setup: boolean
471

472
  """
473
  _CleanDirectory(constants.DATA_DIR)
474
  _CleanDirectory(constants.CRYPTO_KEYS_DIR)
475
  JobQueuePurge()
476

    
477
  if modify_ssh_setup:
478
    try:
479
      priv_key, pub_key, auth_keys = ssh.GetUserFiles(constants.GANETI_RUNAS)
480

    
481
      utils.RemoveAuthorizedKey(auth_keys, utils.ReadFile(pub_key))
482

    
483
      utils.RemoveFile(priv_key)
484
      utils.RemoveFile(pub_key)
485
    except errors.OpExecError:
486
      logging.exception("Error while processing ssh files")
487

    
488
  try:
489
    utils.RemoveFile(constants.CONFD_HMAC_KEY)
490
    utils.RemoveFile(constants.RAPI_CERT_FILE)
491
    utils.RemoveFile(constants.SPICE_CERT_FILE)
492
    utils.RemoveFile(constants.SPICE_CACERT_FILE)
493
    utils.RemoveFile(constants.NODED_CERT_FILE)
494
  except: # pylint: disable=W0702
495
    logging.exception("Error while removing cluster secrets")
496

    
497
  result = utils.RunCmd([constants.DAEMON_UTIL, "stop", constants.CONFD])
498
  if result.failed:
499
    logging.error("Command %s failed with exitcode %s and error %s",
500
                  result.cmd, result.exit_code, result.output)
501

    
502
  # Raise a custom exception (handled in ganeti-noded)
503
  raise errors.QuitGanetiException(True, "Shutdown scheduled")
504

    
505

    
506
def GetNodeInfo(vgname, hypervisor_type):
507
  """Gives back a hash with different information about the node.
508

509
  @type vgname: C{string}
510
  @param vgname: the name of the volume group to ask for disk space information
511
  @type hypervisor_type: C{str}
512
  @param hypervisor_type: the name of the hypervisor to ask for
513
      memory information
514
  @rtype: C{dict}
515
  @return: dictionary with the following keys:
516
      - vg_size is the size of the configured volume group in MiB
517
      - vg_free is the free size of the volume group in MiB
518
      - memory_dom0 is the memory allocated for domain0 in MiB
519
      - memory_free is the currently available (free) ram in MiB
520
      - memory_total is the total number of ram in MiB
521
      - hv_version: the hypervisor version, if available
522

523
  """
524
  outputarray = {}
525

    
526
  if vgname is not None:
527
    vginfo = bdev.LogicalVolume.GetVGInfo([vgname])
528
    vg_free = vg_size = None
529
    if vginfo:
530
      vg_free = int(round(vginfo[0][0], 0))
531
      vg_size = int(round(vginfo[0][1], 0))
532
    outputarray["vg_size"] = vg_size
533
    outputarray["vg_free"] = vg_free
534

    
535
  if hypervisor_type is not None:
536
    hyper = hypervisor.GetHypervisor(hypervisor_type)
537
    hyp_info = hyper.GetNodeInfo()
538
    if hyp_info is not None:
539
      outputarray.update(hyp_info)
540

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

    
543
  return outputarray
544

    
545

    
546
def VerifyNode(what, cluster_name):
547
  """Verify the status of the local node.
548

549
  Based on the input L{what} parameter, various checks are done on the
550
  local node.
551

552
  If the I{filelist} key is present, this list of
553
  files is checksummed and the file/checksum pairs are returned.
554

555
  If the I{nodelist} key is present, we check that we have
556
  connectivity via ssh with the target nodes (and check the hostname
557
  report).
558

559
  If the I{node-net-test} key is present, we check that we have
560
  connectivity to the given nodes via both primary IP and, if
561
  applicable, secondary IPs.
562

563
  @type what: C{dict}
564
  @param what: a dictionary of things to check:
565
      - filelist: list of files for which to compute checksums
566
      - nodelist: list of nodes we should check ssh communication with
567
      - node-net-test: list of nodes we should check node daemon port
568
        connectivity with
569
      - hypervisor: list with hypervisors to run the verify for
570
  @rtype: dict
571
  @return: a dictionary with the same keys as the input dict, and
572
      values representing the result of the checks
573

574
  """
575
  result = {}
576
  my_name = netutils.Hostname.GetSysName()
577
  port = netutils.GetDaemonPort(constants.NODED)
578
  vm_capable = my_name not in what.get(constants.NV_VMNODES, [])
579

    
580
  if constants.NV_HYPERVISOR in what and vm_capable:
581
    result[constants.NV_HYPERVISOR] = tmp = {}
582
    for hv_name in what[constants.NV_HYPERVISOR]:
583
      try:
584
        val = hypervisor.GetHypervisor(hv_name).Verify()
585
      except errors.HypervisorError, err:
586
        val = "Error while checking hypervisor: %s" % str(err)
587
      tmp[hv_name] = val
588

    
589
  if constants.NV_HVPARAMS in what and vm_capable:
590
    result[constants.NV_HVPARAMS] = tmp = []
591
    for source, hv_name, hvparms in what[constants.NV_HVPARAMS]:
592
      try:
593
        logging.info("Validating hv %s, %s", hv_name, hvparms)
594
        hypervisor.GetHypervisor(hv_name).ValidateParameters(hvparms)
595
      except errors.HypervisorError, err:
596
        tmp.append((source, hv_name, str(err)))
597

    
598
  if constants.NV_FILELIST in what:
599
    result[constants.NV_FILELIST] = utils.FingerprintFiles(
600
      what[constants.NV_FILELIST])
601

    
602
  if constants.NV_NODELIST in what:
603
    (nodes, bynode) = what[constants.NV_NODELIST]
604

    
605
    # Add nodes from other groups (different for each node)
606
    try:
607
      nodes.extend(bynode[my_name])
608
    except KeyError:
609
      pass
610

    
611
    # Use a random order
612
    random.shuffle(nodes)
613

    
614
    # Try to contact all nodes
615
    val = {}
616
    for node in nodes:
617
      success, message = _GetSshRunner(cluster_name).VerifyNodeHostname(node)
618
      if not success:
619
        val[node] = message
620

    
621
    result[constants.NV_NODELIST] = val
622

    
623
  if constants.NV_NODENETTEST in what:
624
    result[constants.NV_NODENETTEST] = tmp = {}
625
    my_pip = my_sip = None
626
    for name, pip, sip in what[constants.NV_NODENETTEST]:
627
      if name == my_name:
628
        my_pip = pip
629
        my_sip = sip
630
        break
631
    if not my_pip:
632
      tmp[my_name] = ("Can't find my own primary/secondary IP"
633
                      " in the node list")
634
    else:
635
      for name, pip, sip in what[constants.NV_NODENETTEST]:
636
        fail = []
637
        if not netutils.TcpPing(pip, port, source=my_pip):
638
          fail.append("primary")
639
        if sip != pip:
640
          if not netutils.TcpPing(sip, port, source=my_sip):
641
            fail.append("secondary")
642
        if fail:
643
          tmp[name] = ("failure using the %s interface(s)" %
644
                       " and ".join(fail))
645

    
646
  if constants.NV_MASTERIP in what:
647
    # FIXME: add checks on incoming data structures (here and in the
648
    # rest of the function)
649
    master_name, master_ip = what[constants.NV_MASTERIP]
650
    if master_name == my_name:
651
      source = constants.IP4_ADDRESS_LOCALHOST
652
    else:
653
      source = None
654
    result[constants.NV_MASTERIP] = netutils.TcpPing(master_ip, port,
655
                                                  source=source)
656

    
657
  if constants.NV_OOB_PATHS in what:
658
    result[constants.NV_OOB_PATHS] = tmp = []
659
    for path in what[constants.NV_OOB_PATHS]:
660
      try:
661
        st = os.stat(path)
662
      except OSError, err:
663
        tmp.append("error stating out of band helper: %s" % err)
664
      else:
665
        if stat.S_ISREG(st.st_mode):
666
          if stat.S_IMODE(st.st_mode) & stat.S_IXUSR:
667
            tmp.append(None)
668
          else:
669
            tmp.append("out of band helper %s is not executable" % path)
670
        else:
671
          tmp.append("out of band helper %s is not a file" % path)
672

    
673
  if constants.NV_LVLIST in what and vm_capable:
674
    try:
675
      val = GetVolumeList(utils.ListVolumeGroups().keys())
676
    except RPCFail, err:
677
      val = str(err)
678
    result[constants.NV_LVLIST] = val
679

    
680
  if constants.NV_INSTANCELIST in what and vm_capable:
681
    # GetInstanceList can fail
682
    try:
683
      val = GetInstanceList(what[constants.NV_INSTANCELIST])
684
    except RPCFail, err:
685
      val = str(err)
686
    result[constants.NV_INSTANCELIST] = val
687

    
688
  if constants.NV_VGLIST in what and vm_capable:
689
    result[constants.NV_VGLIST] = utils.ListVolumeGroups()
690

    
691
  if constants.NV_PVLIST in what and vm_capable:
692
    result[constants.NV_PVLIST] = \
693
      bdev.LogicalVolume.GetPVInfo(what[constants.NV_PVLIST],
694
                                   filter_allocatable=False)
695

    
696
  if constants.NV_VERSION in what:
697
    result[constants.NV_VERSION] = (constants.PROTOCOL_VERSION,
698
                                    constants.RELEASE_VERSION)
699

    
700
  if constants.NV_HVINFO in what and vm_capable:
701
    hyper = hypervisor.GetHypervisor(what[constants.NV_HVINFO])
702
    result[constants.NV_HVINFO] = hyper.GetNodeInfo()
703

    
704
  if constants.NV_DRBDLIST in what and vm_capable:
705
    try:
706
      used_minors = bdev.DRBD8.GetUsedDevs().keys()
707
    except errors.BlockDeviceError, err:
708
      logging.warning("Can't get used minors list", exc_info=True)
709
      used_minors = str(err)
710
    result[constants.NV_DRBDLIST] = used_minors
711

    
712
  if constants.NV_DRBDHELPER in what and vm_capable:
713
    status = True
714
    try:
715
      payload = bdev.BaseDRBD.GetUsermodeHelper()
716
    except errors.BlockDeviceError, err:
717
      logging.error("Can't get DRBD usermode helper: %s", str(err))
718
      status = False
719
      payload = str(err)
720
    result[constants.NV_DRBDHELPER] = (status, payload)
721

    
722
  if constants.NV_NODESETUP in what:
723
    result[constants.NV_NODESETUP] = tmpr = []
724
    if not os.path.isdir("/sys/block") or not os.path.isdir("/sys/class/net"):
725
      tmpr.append("The sysfs filesytem doesn't seem to be mounted"
726
                  " under /sys, missing required directories /sys/block"
727
                  " and /sys/class/net")
728
    if (not os.path.isdir("/proc/sys") or
729
        not os.path.isfile("/proc/sysrq-trigger")):
730
      tmpr.append("The procfs filesystem doesn't seem to be mounted"
731
                  " under /proc, missing required directory /proc/sys and"
732
                  " the file /proc/sysrq-trigger")
733

    
734
  if constants.NV_TIME in what:
735
    result[constants.NV_TIME] = utils.SplitTime(time.time())
736

    
737
  if constants.NV_OSLIST in what and vm_capable:
738
    result[constants.NV_OSLIST] = DiagnoseOS()
739

    
740
  if constants.NV_BRIDGES in what and vm_capable:
741
    result[constants.NV_BRIDGES] = [bridge
742
                                    for bridge in what[constants.NV_BRIDGES]
743
                                    if not utils.BridgeExists(bridge)]
744
  return result
745

    
746

    
747
def GetBlockDevSizes(devices):
748
  """Return the size of the given block devices
749

750
  @type devices: list
751
  @param devices: list of block device nodes to query
752
  @rtype: dict
753
  @return:
754
    dictionary of all block devices under /dev (key). The value is their
755
    size in MiB.
756

757
    {'/dev/disk/by-uuid/123456-12321231-312312-312': 124}
758

759
  """
760
  DEV_PREFIX = "/dev/"
761
  blockdevs = {}
762

    
763
  for devpath in devices:
764
    if not utils.IsBelowDir(DEV_PREFIX, devpath):
765
      continue
766

    
767
    try:
768
      st = os.stat(devpath)
769
    except EnvironmentError, err:
770
      logging.warning("Error stat()'ing device %s: %s", devpath, str(err))
771
      continue
772

    
773
    if stat.S_ISBLK(st.st_mode):
774
      result = utils.RunCmd(["blockdev", "--getsize64", devpath])
775
      if result.failed:
776
        # We don't want to fail, just do not list this device as available
777
        logging.warning("Cannot get size for block device %s", devpath)
778
        continue
779

    
780
      size = int(result.stdout) / (1024 * 1024)
781
      blockdevs[devpath] = size
782
  return blockdevs
783

    
784

    
785
def GetVolumeList(vg_names):
786
  """Compute list of logical volumes and their size.
787

788
  @type vg_names: list
789
  @param vg_names: the volume groups whose LVs we should list, or
790
      empty for all volume groups
791
  @rtype: dict
792
  @return:
793
      dictionary of all partions (key) with value being a tuple of
794
      their size (in MiB), inactive and online status::
795

796
        {'xenvg/test1': ('20.06', True, True)}
797

798
      in case of errors, a string is returned with the error
799
      details.
800

801
  """
802
  lvs = {}
803
  sep = "|"
804
  if not vg_names:
805
    vg_names = []
806
  result = utils.RunCmd(["lvs", "--noheadings", "--units=m", "--nosuffix",
807
                         "--separator=%s" % sep,
808
                         "-ovg_name,lv_name,lv_size,lv_attr"] + vg_names)
809
  if result.failed:
810
    _Fail("Failed to list logical volumes, lvs output: %s", result.output)
811

    
812
  for line in result.stdout.splitlines():
813
    line = line.strip()
814
    match = _LVSLINE_REGEX.match(line)
815
    if not match:
816
      logging.error("Invalid line returned from lvs output: '%s'", line)
817
      continue
818
    vg_name, name, size, attr = match.groups()
819
    inactive = attr[4] == "-"
820
    online = attr[5] == "o"
821
    virtual = attr[0] == "v"
822
    if virtual:
823
      # we don't want to report such volumes as existing, since they
824
      # don't really hold data
825
      continue
826
    lvs[vg_name + "/" + name] = (size, inactive, online)
827

    
828
  return lvs
829

    
830

    
831
def ListVolumeGroups():
832
  """List the volume groups and their size.
833

834
  @rtype: dict
835
  @return: dictionary with keys volume name and values the
836
      size of the volume
837

838
  """
839
  return utils.ListVolumeGroups()
840

    
841

    
842
def NodeVolumes():
843
  """List all volumes on this node.
844

845
  @rtype: list
846
  @return:
847
    A list of dictionaries, each having four keys:
848
      - name: the logical volume name,
849
      - size: the size of the logical volume
850
      - dev: the physical device on which the LV lives
851
      - vg: the volume group to which it belongs
852

853
    In case of errors, we return an empty list and log the
854
    error.
855

856
    Note that since a logical volume can live on multiple physical
857
    volumes, the resulting list might include a logical volume
858
    multiple times.
859

860
  """
861
  result = utils.RunCmd(["lvs", "--noheadings", "--units=m", "--nosuffix",
862
                         "--separator=|",
863
                         "--options=lv_name,lv_size,devices,vg_name"])
864
  if result.failed:
865
    _Fail("Failed to list logical volumes, lvs output: %s",
866
          result.output)
867

    
868
  def parse_dev(dev):
869
    return dev.split("(")[0]
870

    
871
  def handle_dev(dev):
872
    return [parse_dev(x) for x in dev.split(",")]
873

    
874
  def map_line(line):
875
    line = [v.strip() for v in line]
876
    return [{"name": line[0], "size": line[1],
877
             "dev": dev, "vg": line[3]} for dev in handle_dev(line[2])]
878

    
879
  all_devs = []
880
  for line in result.stdout.splitlines():
881
    if line.count("|") >= 3:
882
      all_devs.extend(map_line(line.split("|")))
883
    else:
884
      logging.warning("Strange line in the output from lvs: '%s'", line)
885
  return all_devs
886

    
887

    
888
def BridgesExist(bridges_list):
889
  """Check if a list of bridges exist on the current node.
890

891
  @rtype: boolean
892
  @return: C{True} if all of them exist, C{False} otherwise
893

894
  """
895
  missing = []
896
  for bridge in bridges_list:
897
    if not utils.BridgeExists(bridge):
898
      missing.append(bridge)
899

    
900
  if missing:
901
    _Fail("Missing bridges %s", utils.CommaJoin(missing))
902

    
903

    
904
def GetInstanceList(hypervisor_list):
905
  """Provides a list of instances.
906

907
  @type hypervisor_list: list
908
  @param hypervisor_list: the list of hypervisors to query information
909

910
  @rtype: list
911
  @return: a list of all running instances on the current node
912
    - instance1.example.com
913
    - instance2.example.com
914

915
  """
916
  results = []
917
  for hname in hypervisor_list:
918
    try:
919
      names = hypervisor.GetHypervisor(hname).ListInstances()
920
      results.extend(names)
921
    except errors.HypervisorError, err:
922
      _Fail("Error enumerating instances (hypervisor %s): %s",
923
            hname, err, exc=True)
924

    
925
  return results
926

    
927

    
928
def GetInstanceInfo(instance, hname):
929
  """Gives back the information about an instance as a dictionary.
930

931
  @type instance: string
932
  @param instance: the instance name
933
  @type hname: string
934
  @param hname: the hypervisor type of the instance
935

936
  @rtype: dict
937
  @return: dictionary with the following keys:
938
      - memory: memory size of instance (int)
939
      - state: xen state of instance (string)
940
      - time: cpu time of instance (float)
941

942
  """
943
  output = {}
944

    
945
  iinfo = hypervisor.GetHypervisor(hname).GetInstanceInfo(instance)
946
  if iinfo is not None:
947
    output["memory"] = iinfo[2]
948
    output["state"] = iinfo[4]
949
    output["time"] = iinfo[5]
950

    
951
  return output
952

    
953

    
954
def GetInstanceMigratable(instance):
955
  """Gives whether an instance can be migrated.
956

957
  @type instance: L{objects.Instance}
958
  @param instance: object representing the instance to be checked.
959

960
  @rtype: tuple
961
  @return: tuple of (result, description) where:
962
      - result: whether the instance can be migrated or not
963
      - description: a description of the issue, if relevant
964

965
  """
966
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
967
  iname = instance.name
968
  if iname not in hyper.ListInstances():
969
    _Fail("Instance %s is not running", iname)
970

    
971
  for idx in range(len(instance.disks)):
972
    link_name = _GetBlockDevSymlinkPath(iname, idx)
973
    if not os.path.islink(link_name):
974
      logging.warning("Instance %s is missing symlink %s for disk %d",
975
                      iname, link_name, idx)
976

    
977

    
978
def GetAllInstancesInfo(hypervisor_list):
979
  """Gather data about all instances.
980

981
  This is the equivalent of L{GetInstanceInfo}, except that it
982
  computes data for all instances at once, thus being faster if one
983
  needs data about more than one instance.
984

985
  @type hypervisor_list: list
986
  @param hypervisor_list: list of hypervisors to query for instance data
987

988
  @rtype: dict
989
  @return: dictionary of instance: data, with data having the following keys:
990
      - memory: memory size of instance (int)
991
      - state: xen state of instance (string)
992
      - time: cpu time of instance (float)
993
      - vcpus: the number of vcpus
994

995
  """
996
  output = {}
997

    
998
  for hname in hypervisor_list:
999
    iinfo = hypervisor.GetHypervisor(hname).GetAllInstancesInfo()
1000
    if iinfo:
1001
      for name, _, memory, vcpus, state, times in iinfo:
1002
        value = {
1003
          "memory": memory,
1004
          "vcpus": vcpus,
1005
          "state": state,
1006
          "time": times,
1007
          }
1008
        if name in output:
1009
          # we only check static parameters, like memory and vcpus,
1010
          # and not state and time which can change between the
1011
          # invocations of the different hypervisors
1012
          for key in "memory", "vcpus":
1013
            if value[key] != output[name][key]:
1014
              _Fail("Instance %s is running twice"
1015
                    " with different parameters", name)
1016
        output[name] = value
1017

    
1018
  return output
1019

    
1020

    
1021
def _InstanceLogName(kind, os_name, instance, component):
1022
  """Compute the OS log filename for a given instance and operation.
1023

1024
  The instance name and os name are passed in as strings since not all
1025
  operations have these as part of an instance object.
1026

1027
  @type kind: string
1028
  @param kind: the operation type (e.g. add, import, etc.)
1029
  @type os_name: string
1030
  @param os_name: the os name
1031
  @type instance: string
1032
  @param instance: the name of the instance being imported/added/etc.
1033
  @type component: string or None
1034
  @param component: the name of the component of the instance being
1035
      transferred
1036

1037
  """
1038
  # TODO: Use tempfile.mkstemp to create unique filename
1039
  if component:
1040
    assert "/" not in component
1041
    c_msg = "-%s" % component
1042
  else:
1043
    c_msg = ""
1044
  base = ("%s-%s-%s%s-%s.log" %
1045
          (kind, os_name, instance, c_msg, utils.TimestampForFilename()))
1046
  return utils.PathJoin(constants.LOG_OS_DIR, base)
1047

    
1048

    
1049
def InstanceOsAdd(instance, reinstall, debug):
1050
  """Add an OS to an instance.
1051

1052
  @type instance: L{objects.Instance}
1053
  @param instance: Instance whose OS is to be installed
1054
  @type reinstall: boolean
1055
  @param reinstall: whether this is an instance reinstall
1056
  @type debug: integer
1057
  @param debug: debug level, passed to the OS scripts
1058
  @rtype: None
1059

1060
  """
1061
  inst_os = OSFromDisk(instance.os)
1062

    
1063
  create_env = OSEnvironment(instance, inst_os, debug)
1064
  if reinstall:
1065
    create_env["INSTANCE_REINSTALL"] = "1"
1066

    
1067
  logfile = _InstanceLogName("add", instance.os, instance.name, None)
1068

    
1069
  result = utils.RunCmd([inst_os.create_script], env=create_env,
1070
                        cwd=inst_os.path, output=logfile, reset_env=True)
1071
  if result.failed:
1072
    logging.error("os create command '%s' returned error: %s, logfile: %s,"
1073
                  " output: %s", result.cmd, result.fail_reason, logfile,
1074
                  result.output)
1075
    lines = [utils.SafeEncode(val)
1076
             for val in utils.TailFile(logfile, lines=20)]
1077
    _Fail("OS create script failed (%s), last lines in the"
1078
          " log file:\n%s", result.fail_reason, "\n".join(lines), log=False)
1079

    
1080

    
1081
def RunRenameInstance(instance, old_name, debug):
1082
  """Run the OS rename script for an instance.
1083

1084
  @type instance: L{objects.Instance}
1085
  @param instance: Instance whose OS is to be installed
1086
  @type old_name: string
1087
  @param old_name: previous instance name
1088
  @type debug: integer
1089
  @param debug: debug level, passed to the OS scripts
1090
  @rtype: boolean
1091
  @return: the success of the operation
1092

1093
  """
1094
  inst_os = OSFromDisk(instance.os)
1095

    
1096
  rename_env = OSEnvironment(instance, inst_os, debug)
1097
  rename_env["OLD_INSTANCE_NAME"] = old_name
1098

    
1099
  logfile = _InstanceLogName("rename", instance.os,
1100
                             "%s-%s" % (old_name, instance.name), None)
1101

    
1102
  result = utils.RunCmd([inst_os.rename_script], env=rename_env,
1103
                        cwd=inst_os.path, output=logfile, reset_env=True)
1104

    
1105
  if result.failed:
1106
    logging.error("os create command '%s' returned error: %s output: %s",
1107
                  result.cmd, result.fail_reason, result.output)
1108
    lines = [utils.SafeEncode(val)
1109
             for val in utils.TailFile(logfile, lines=20)]
1110
    _Fail("OS rename script failed (%s), last lines in the"
1111
          " log file:\n%s", result.fail_reason, "\n".join(lines), log=False)
1112

    
1113

    
1114
def _GetBlockDevSymlinkPath(instance_name, idx):
1115
  return utils.PathJoin(constants.DISK_LINKS_DIR, "%s%s%d" %
1116
                        (instance_name, constants.DISK_SEPARATOR, idx))
1117

    
1118

    
1119
def _SymlinkBlockDev(instance_name, device_path, idx):
1120
  """Set up symlinks to a instance's block device.
1121

1122
  This is an auxiliary function run when an instance is start (on the primary
1123
  node) or when an instance is migrated (on the target node).
1124

1125

1126
  @param instance_name: the name of the target instance
1127
  @param device_path: path of the physical block device, on the node
1128
  @param idx: the disk index
1129
  @return: absolute path to the disk's symlink
1130

1131
  """
1132
  link_name = _GetBlockDevSymlinkPath(instance_name, idx)
1133
  try:
1134
    os.symlink(device_path, link_name)
1135
  except OSError, err:
1136
    if err.errno == errno.EEXIST:
1137
      if (not os.path.islink(link_name) or
1138
          os.readlink(link_name) != device_path):
1139
        os.remove(link_name)
1140
        os.symlink(device_path, link_name)
1141
    else:
1142
      raise
1143

    
1144
  return link_name
1145

    
1146

    
1147
def _RemoveBlockDevLinks(instance_name, disks):
1148
  """Remove the block device symlinks belonging to the given instance.
1149

1150
  """
1151
  for idx, _ in enumerate(disks):
1152
    link_name = _GetBlockDevSymlinkPath(instance_name, idx)
1153
    if os.path.islink(link_name):
1154
      try:
1155
        os.remove(link_name)
1156
      except OSError:
1157
        logging.exception("Can't remove symlink '%s'", link_name)
1158

    
1159

    
1160
def _GatherAndLinkBlockDevs(instance):
1161
  """Set up an instance's block device(s).
1162

1163
  This is run on the primary node at instance startup. The block
1164
  devices must be already assembled.
1165

1166
  @type instance: L{objects.Instance}
1167
  @param instance: the instance whose disks we shoul assemble
1168
  @rtype: list
1169
  @return: list of (disk_object, device_path)
1170

1171
  """
1172
  block_devices = []
1173
  for idx, disk in enumerate(instance.disks):
1174
    device = _RecursiveFindBD(disk)
1175
    if device is None:
1176
      raise errors.BlockDeviceError("Block device '%s' is not set up." %
1177
                                    str(disk))
1178
    device.Open()
1179
    try:
1180
      link_name = _SymlinkBlockDev(instance.name, device.dev_path, idx)
1181
    except OSError, e:
1182
      raise errors.BlockDeviceError("Cannot create block device symlink: %s" %
1183
                                    e.strerror)
1184

    
1185
    block_devices.append((disk, link_name))
1186

    
1187
  return block_devices
1188

    
1189

    
1190
def StartInstance(instance, startup_paused):
1191
  """Start an instance.
1192

1193
  @type instance: L{objects.Instance}
1194
  @param instance: the instance object
1195
  @type startup_paused: bool
1196
  @param instance: pause instance at startup?
1197
  @rtype: None
1198

1199
  """
1200
  running_instances = GetInstanceList([instance.hypervisor])
1201

    
1202
  if instance.name in running_instances:
1203
    logging.info("Instance %s already running, not starting", instance.name)
1204
    return
1205

    
1206
  try:
1207
    block_devices = _GatherAndLinkBlockDevs(instance)
1208
    hyper = hypervisor.GetHypervisor(instance.hypervisor)
1209
    hyper.StartInstance(instance, block_devices, startup_paused)
1210
  except errors.BlockDeviceError, err:
1211
    _Fail("Block device error: %s", err, exc=True)
1212
  except errors.HypervisorError, err:
1213
    _RemoveBlockDevLinks(instance.name, instance.disks)
1214
    _Fail("Hypervisor error: %s", err, exc=True)
1215

    
1216

    
1217
def InstanceShutdown(instance, timeout):
1218
  """Shut an instance down.
1219

1220
  @note: this functions uses polling with a hardcoded timeout.
1221

1222
  @type instance: L{objects.Instance}
1223
  @param instance: the instance object
1224
  @type timeout: integer
1225
  @param timeout: maximum timeout for soft shutdown
1226
  @rtype: None
1227

1228
  """
1229
  hv_name = instance.hypervisor
1230
  hyper = hypervisor.GetHypervisor(hv_name)
1231
  iname = instance.name
1232

    
1233
  if instance.name not in hyper.ListInstances():
1234
    logging.info("Instance %s not running, doing nothing", iname)
1235
    return
1236

    
1237
  class _TryShutdown:
1238
    def __init__(self):
1239
      self.tried_once = False
1240

    
1241
    def __call__(self):
1242
      if iname not in hyper.ListInstances():
1243
        return
1244

    
1245
      try:
1246
        hyper.StopInstance(instance, retry=self.tried_once)
1247
      except errors.HypervisorError, err:
1248
        if iname not in hyper.ListInstances():
1249
          # if the instance is no longer existing, consider this a
1250
          # success and go to cleanup
1251
          return
1252

    
1253
        _Fail("Failed to stop instance %s: %s", iname, err)
1254

    
1255
      self.tried_once = True
1256

    
1257
      raise utils.RetryAgain()
1258

    
1259
  try:
1260
    utils.Retry(_TryShutdown(), 5, timeout)
1261
  except utils.RetryTimeout:
1262
    # the shutdown did not succeed
1263
    logging.error("Shutdown of '%s' unsuccessful, forcing", iname)
1264

    
1265
    try:
1266
      hyper.StopInstance(instance, force=True)
1267
    except errors.HypervisorError, err:
1268
      if iname in hyper.ListInstances():
1269
        # only raise an error if the instance still exists, otherwise
1270
        # the error could simply be "instance ... unknown"!
1271
        _Fail("Failed to force stop instance %s: %s", iname, err)
1272

    
1273
    time.sleep(1)
1274

    
1275
    if iname in hyper.ListInstances():
1276
      _Fail("Could not shutdown instance %s even by destroy", iname)
1277

    
1278
  try:
1279
    hyper.CleanupInstance(instance.name)
1280
  except errors.HypervisorError, err:
1281
    logging.warning("Failed to execute post-shutdown cleanup step: %s", err)
1282

    
1283
  _RemoveBlockDevLinks(iname, instance.disks)
1284

    
1285

    
1286
def InstanceReboot(instance, reboot_type, shutdown_timeout):
1287
  """Reboot an instance.
1288

1289
  @type instance: L{objects.Instance}
1290
  @param instance: the instance object to reboot
1291
  @type reboot_type: str
1292
  @param reboot_type: the type of reboot, one the following
1293
    constants:
1294
      - L{constants.INSTANCE_REBOOT_SOFT}: only reboot the
1295
        instance OS, do not recreate the VM
1296
      - L{constants.INSTANCE_REBOOT_HARD}: tear down and
1297
        restart the VM (at the hypervisor level)
1298
      - the other reboot type (L{constants.INSTANCE_REBOOT_FULL}) is
1299
        not accepted here, since that mode is handled differently, in
1300
        cmdlib, and translates into full stop and start of the
1301
        instance (instead of a call_instance_reboot RPC)
1302
  @type shutdown_timeout: integer
1303
  @param shutdown_timeout: maximum timeout for soft shutdown
1304
  @rtype: None
1305

1306
  """
1307
  running_instances = GetInstanceList([instance.hypervisor])
1308

    
1309
  if instance.name not in running_instances:
1310
    _Fail("Cannot reboot instance %s that is not running", instance.name)
1311

    
1312
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1313
  if reboot_type == constants.INSTANCE_REBOOT_SOFT:
1314
    try:
1315
      hyper.RebootInstance(instance)
1316
    except errors.HypervisorError, err:
1317
      _Fail("Failed to soft reboot instance %s: %s", instance.name, err)
1318
  elif reboot_type == constants.INSTANCE_REBOOT_HARD:
1319
    try:
1320
      InstanceShutdown(instance, shutdown_timeout)
1321
      return StartInstance(instance, False)
1322
    except errors.HypervisorError, err:
1323
      _Fail("Failed to hard reboot instance %s: %s", instance.name, err)
1324
  else:
1325
    _Fail("Invalid reboot_type received: %s", reboot_type)
1326

    
1327

    
1328
def MigrationInfo(instance):
1329
  """Gather information about an instance to be migrated.
1330

1331
  @type instance: L{objects.Instance}
1332
  @param instance: the instance definition
1333

1334
  """
1335
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1336
  try:
1337
    info = hyper.MigrationInfo(instance)
1338
  except errors.HypervisorError, err:
1339
    _Fail("Failed to fetch migration information: %s", err, exc=True)
1340
  return info
1341

    
1342

    
1343
def AcceptInstance(instance, info, target):
1344
  """Prepare the node to accept an instance.
1345

1346
  @type instance: L{objects.Instance}
1347
  @param instance: the instance definition
1348
  @type info: string/data (opaque)
1349
  @param info: migration information, from the source node
1350
  @type target: string
1351
  @param target: target host (usually ip), on this node
1352

1353
  """
1354
  # TODO: why is this required only for DTS_EXT_MIRROR?
1355
  if instance.disk_template in constants.DTS_EXT_MIRROR:
1356
    # Create the symlinks, as the disks are not active
1357
    # in any way
1358
    try:
1359
      _GatherAndLinkBlockDevs(instance)
1360
    except errors.BlockDeviceError, err:
1361
      _Fail("Block device error: %s", err, exc=True)
1362

    
1363
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1364
  try:
1365
    hyper.AcceptInstance(instance, info, target)
1366
  except errors.HypervisorError, err:
1367
    if instance.disk_template in constants.DTS_EXT_MIRROR:
1368
      _RemoveBlockDevLinks(instance.name, instance.disks)
1369
    _Fail("Failed to accept instance: %s", err, exc=True)
1370

    
1371

    
1372
def FinalizeMigrationDst(instance, info, success):
1373
  """Finalize any preparation to accept an instance.
1374

1375
  @type instance: L{objects.Instance}
1376
  @param instance: the instance definition
1377
  @type info: string/data (opaque)
1378
  @param info: migration information, from the source node
1379
  @type success: boolean
1380
  @param success: whether the migration was a success or a failure
1381

1382
  """
1383
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1384
  try:
1385
    hyper.FinalizeMigrationDst(instance, info, success)
1386
  except errors.HypervisorError, err:
1387
    _Fail("Failed to finalize migration on the target node: %s", err, exc=True)
1388

    
1389

    
1390
def MigrateInstance(instance, target, live):
1391
  """Migrates an instance to another node.
1392

1393
  @type instance: L{objects.Instance}
1394
  @param instance: the instance definition
1395
  @type target: string
1396
  @param target: the target node name
1397
  @type live: boolean
1398
  @param live: whether the migration should be done live or not (the
1399
      interpretation of this parameter is left to the hypervisor)
1400
  @raise RPCFail: if migration fails for some reason
1401

1402
  """
1403
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1404

    
1405
  try:
1406
    hyper.MigrateInstance(instance, target, live)
1407
  except errors.HypervisorError, err:
1408
    _Fail("Failed to migrate instance: %s", err, exc=True)
1409

    
1410

    
1411
def FinalizeMigrationSource(instance, success, live):
1412
  """Finalize the instance migration on the source node.
1413

1414
  @type instance: L{objects.Instance}
1415
  @param instance: the instance definition of the migrated instance
1416
  @type success: bool
1417
  @param success: whether the migration succeeded or not
1418
  @type live: bool
1419
  @param live: whether the user requested a live migration or not
1420
  @raise RPCFail: If the execution fails for some reason
1421

1422
  """
1423
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1424

    
1425
  try:
1426
    hyper.FinalizeMigrationSource(instance, success, live)
1427
  except Exception, err:  # pylint: disable=W0703
1428
    _Fail("Failed to finalize the migration on the source node: %s", err,
1429
          exc=True)
1430

    
1431

    
1432
def GetMigrationStatus(instance):
1433
  """Get the migration status
1434

1435
  @type instance: L{objects.Instance}
1436
  @param instance: the instance that is being migrated
1437
  @rtype: L{objects.MigrationStatus}
1438
  @return: the status of the current migration (one of
1439
           L{constants.HV_MIGRATION_VALID_STATUSES}), plus any additional
1440
           progress info that can be retrieved from the hypervisor
1441
  @raise RPCFail: If the migration status cannot be retrieved
1442

1443
  """
1444
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1445
  try:
1446
    return hyper.GetMigrationStatus(instance)
1447
  except Exception, err:  # pylint: disable=W0703
1448
    _Fail("Failed to get migration status: %s", err, exc=True)
1449

    
1450

    
1451
def BlockdevCreate(disk, size, owner, on_primary, info):
1452
  """Creates a block device for an instance.
1453

1454
  @type disk: L{objects.Disk}
1455
  @param disk: the object describing the disk we should create
1456
  @type size: int
1457
  @param size: the size of the physical underlying device, in MiB
1458
  @type owner: str
1459
  @param owner: the name of the instance for which disk is created,
1460
      used for device cache data
1461
  @type on_primary: boolean
1462
  @param on_primary:  indicates if it is the primary node or not
1463
  @type info: string
1464
  @param info: string that will be sent to the physical device
1465
      creation, used for example to set (LVM) tags on LVs
1466

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

1471
  """
1472
  # TODO: remove the obsolete "size" argument
1473
  # pylint: disable=W0613
1474
  clist = []
1475
  if disk.children:
1476
    for child in disk.children:
1477
      try:
1478
        crdev = _RecursiveAssembleBD(child, owner, on_primary)
1479
      except errors.BlockDeviceError, err:
1480
        _Fail("Can't assemble device %s: %s", child, err)
1481
      if on_primary or disk.AssembleOnSecondary():
1482
        # we need the children open in case the device itself has to
1483
        # be assembled
1484
        try:
1485
          # pylint: disable=E1103
1486
          crdev.Open()
1487
        except errors.BlockDeviceError, err:
1488
          _Fail("Can't make child '%s' read-write: %s", child, err)
1489
      clist.append(crdev)
1490

    
1491
  try:
1492
    device = bdev.Create(disk.dev_type, disk.physical_id, clist, disk.size)
1493
  except errors.BlockDeviceError, err:
1494
    _Fail("Can't create block device: %s", err)
1495

    
1496
  if on_primary or disk.AssembleOnSecondary():
1497
    try:
1498
      device.Assemble()
1499
    except errors.BlockDeviceError, err:
1500
      _Fail("Can't assemble device after creation, unusual event: %s", err)
1501
    device.SetSyncSpeed(constants.SYNC_SPEED)
1502
    if on_primary or disk.OpenOnSecondary():
1503
      try:
1504
        device.Open(force=True)
1505
      except errors.BlockDeviceError, err:
1506
        _Fail("Can't make device r/w after creation, unusual event: %s", err)
1507
    DevCacheManager.UpdateCache(device.dev_path, owner,
1508
                                on_primary, disk.iv_name)
1509

    
1510
  device.SetInfo(info)
1511

    
1512
  return device.unique_id
1513

    
1514

    
1515
def _WipeDevice(path, offset, size):
1516
  """This function actually wipes the device.
1517

1518
  @param path: The path to the device to wipe
1519
  @param offset: The offset in MiB in the file
1520
  @param size: The size in MiB to write
1521

1522
  """
1523
  cmd = [constants.DD_CMD, "if=/dev/zero", "seek=%d" % offset,
1524
         "bs=%d" % constants.WIPE_BLOCK_SIZE, "oflag=direct", "of=%s" % path,
1525
         "count=%d" % size]
1526
  result = utils.RunCmd(cmd)
1527

    
1528
  if result.failed:
1529
    _Fail("Wipe command '%s' exited with error: %s; output: %s", result.cmd,
1530
          result.fail_reason, result.output)
1531

    
1532

    
1533
def BlockdevWipe(disk, offset, size):
1534
  """Wipes a block device.
1535

1536
  @type disk: L{objects.Disk}
1537
  @param disk: the disk object we want to wipe
1538
  @type offset: int
1539
  @param offset: The offset in MiB in the file
1540
  @type size: int
1541
  @param size: The size in MiB to write
1542

1543
  """
1544
  try:
1545
    rdev = _RecursiveFindBD(disk)
1546
  except errors.BlockDeviceError:
1547
    rdev = None
1548

    
1549
  if not rdev:
1550
    _Fail("Cannot execute wipe for device %s: device not found", disk.iv_name)
1551

    
1552
  # Do cross verify some of the parameters
1553
  if offset > rdev.size:
1554
    _Fail("Offset is bigger than device size")
1555
  if (offset + size) > rdev.size:
1556
    _Fail("The provided offset and size to wipe is bigger than device size")
1557

    
1558
  _WipeDevice(rdev.dev_path, offset, size)
1559

    
1560

    
1561
def BlockdevPauseResumeSync(disks, pause):
1562
  """Pause or resume the sync of the block device.
1563

1564
  @type disks: list of L{objects.Disk}
1565
  @param disks: the disks object we want to pause/resume
1566
  @type pause: bool
1567
  @param pause: Wheater to pause or resume
1568

1569
  """
1570
  success = []
1571
  for disk in disks:
1572
    try:
1573
      rdev = _RecursiveFindBD(disk)
1574
    except errors.BlockDeviceError:
1575
      rdev = None
1576

    
1577
    if not rdev:
1578
      success.append((False, ("Cannot change sync for device %s:"
1579
                              " device not found" % disk.iv_name)))
1580
      continue
1581

    
1582
    result = rdev.PauseResumeSync(pause)
1583

    
1584
    if result:
1585
      success.append((result, None))
1586
    else:
1587
      if pause:
1588
        msg = "Pause"
1589
      else:
1590
        msg = "Resume"
1591
      success.append((result, "%s for device %s failed" % (msg, disk.iv_name)))
1592

    
1593
  return success
1594

    
1595

    
1596
def BlockdevRemove(disk):
1597
  """Remove a block device.
1598

1599
  @note: This is intended to be called recursively.
1600

1601
  @type disk: L{objects.Disk}
1602
  @param disk: the disk object we should remove
1603
  @rtype: boolean
1604
  @return: the success of the operation
1605

1606
  """
1607
  msgs = []
1608
  try:
1609
    rdev = _RecursiveFindBD(disk)
1610
  except errors.BlockDeviceError, err:
1611
    # probably can't attach
1612
    logging.info("Can't attach to device %s in remove", disk)
1613
    rdev = None
1614
  if rdev is not None:
1615
    r_path = rdev.dev_path
1616
    try:
1617
      rdev.Remove()
1618
    except errors.BlockDeviceError, err:
1619
      msgs.append(str(err))
1620
    if not msgs:
1621
      DevCacheManager.RemoveCache(r_path)
1622

    
1623
  if disk.children:
1624
    for child in disk.children:
1625
      try:
1626
        BlockdevRemove(child)
1627
      except RPCFail, err:
1628
        msgs.append(str(err))
1629

    
1630
  if msgs:
1631
    _Fail("; ".join(msgs))
1632

    
1633

    
1634
def _RecursiveAssembleBD(disk, owner, as_primary):
1635
  """Activate a block device for an instance.
1636

1637
  This is run on the primary and secondary nodes for an instance.
1638

1639
  @note: this function is called recursively.
1640

1641
  @type disk: L{objects.Disk}
1642
  @param disk: the disk we try to assemble
1643
  @type owner: str
1644
  @param owner: the name of the instance which owns the disk
1645
  @type as_primary: boolean
1646
  @param as_primary: if we should make the block device
1647
      read/write
1648

1649
  @return: the assembled device or None (in case no device
1650
      was assembled)
1651
  @raise errors.BlockDeviceError: in case there is an error
1652
      during the activation of the children or the device
1653
      itself
1654

1655
  """
1656
  children = []
1657
  if disk.children:
1658
    mcn = disk.ChildrenNeeded()
1659
    if mcn == -1:
1660
      mcn = 0 # max number of Nones allowed
1661
    else:
1662
      mcn = len(disk.children) - mcn # max number of Nones
1663
    for chld_disk in disk.children:
1664
      try:
1665
        cdev = _RecursiveAssembleBD(chld_disk, owner, as_primary)
1666
      except errors.BlockDeviceError, err:
1667
        if children.count(None) >= mcn:
1668
          raise
1669
        cdev = None
1670
        logging.error("Error in child activation (but continuing): %s",
1671
                      str(err))
1672
      children.append(cdev)
1673

    
1674
  if as_primary or disk.AssembleOnSecondary():
1675
    r_dev = bdev.Assemble(disk.dev_type, disk.physical_id, children, disk.size)
1676
    r_dev.SetSyncSpeed(constants.SYNC_SPEED)
1677
    result = r_dev
1678
    if as_primary or disk.OpenOnSecondary():
1679
      r_dev.Open()
1680
    DevCacheManager.UpdateCache(r_dev.dev_path, owner,
1681
                                as_primary, disk.iv_name)
1682

    
1683
  else:
1684
    result = True
1685
  return result
1686

    
1687

    
1688
def BlockdevAssemble(disk, owner, as_primary, idx):
1689
  """Activate a block device for an instance.
1690

1691
  This is a wrapper over _RecursiveAssembleBD.
1692

1693
  @rtype: str or boolean
1694
  @return: a C{/dev/...} path for primary nodes, and
1695
      C{True} for secondary nodes
1696

1697
  """
1698
  try:
1699
    result = _RecursiveAssembleBD(disk, owner, as_primary)
1700
    if isinstance(result, bdev.BlockDev):
1701
      # pylint: disable=E1103
1702
      result = result.dev_path
1703
      if as_primary:
1704
        _SymlinkBlockDev(owner, result, idx)
1705
  except errors.BlockDeviceError, err:
1706
    _Fail("Error while assembling disk: %s", err, exc=True)
1707
  except OSError, err:
1708
    _Fail("Error while symlinking disk: %s", err, exc=True)
1709

    
1710
  return result
1711

    
1712

    
1713
def BlockdevShutdown(disk):
1714
  """Shut down a block device.
1715

1716
  First, if the device is assembled (Attach() is successful), then
1717
  the device is shutdown. Then the children of the device are
1718
  shutdown.
1719

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

1724
  @type disk: L{objects.Disk}
1725
  @param disk: the description of the disk we should
1726
      shutdown
1727
  @rtype: None
1728

1729
  """
1730
  msgs = []
1731
  r_dev = _RecursiveFindBD(disk)
1732
  if r_dev is not None:
1733
    r_path = r_dev.dev_path
1734
    try:
1735
      r_dev.Shutdown()
1736
      DevCacheManager.RemoveCache(r_path)
1737
    except errors.BlockDeviceError, err:
1738
      msgs.append(str(err))
1739

    
1740
  if disk.children:
1741
    for child in disk.children:
1742
      try:
1743
        BlockdevShutdown(child)
1744
      except RPCFail, err:
1745
        msgs.append(str(err))
1746

    
1747
  if msgs:
1748
    _Fail("; ".join(msgs))
1749

    
1750

    
1751
def BlockdevAddchildren(parent_cdev, new_cdevs):
1752
  """Extend a mirrored block device.
1753

1754
  @type parent_cdev: L{objects.Disk}
1755
  @param parent_cdev: the disk to which we should add children
1756
  @type new_cdevs: list of L{objects.Disk}
1757
  @param new_cdevs: the list of children which we should add
1758
  @rtype: None
1759

1760
  """
1761
  parent_bdev = _RecursiveFindBD(parent_cdev)
1762
  if parent_bdev is None:
1763
    _Fail("Can't find parent device '%s' in add children", parent_cdev)
1764
  new_bdevs = [_RecursiveFindBD(disk) for disk in new_cdevs]
1765
  if new_bdevs.count(None) > 0:
1766
    _Fail("Can't find new device(s) to add: %s:%s", new_bdevs, new_cdevs)
1767
  parent_bdev.AddChildren(new_bdevs)
1768

    
1769

    
1770
def BlockdevRemovechildren(parent_cdev, new_cdevs):
1771
  """Shrink a mirrored block device.
1772

1773
  @type parent_cdev: L{objects.Disk}
1774
  @param parent_cdev: the disk from which we should remove children
1775
  @type new_cdevs: list of L{objects.Disk}
1776
  @param new_cdevs: the list of children which we should remove
1777
  @rtype: None
1778

1779
  """
1780
  parent_bdev = _RecursiveFindBD(parent_cdev)
1781
  if parent_bdev is None:
1782
    _Fail("Can't find parent device '%s' in remove children", parent_cdev)
1783
  devs = []
1784
  for disk in new_cdevs:
1785
    rpath = disk.StaticDevPath()
1786
    if rpath is None:
1787
      bd = _RecursiveFindBD(disk)
1788
      if bd is None:
1789
        _Fail("Can't find device %s while removing children", disk)
1790
      else:
1791
        devs.append(bd.dev_path)
1792
    else:
1793
      if not utils.IsNormAbsPath(rpath):
1794
        _Fail("Strange path returned from StaticDevPath: '%s'", rpath)
1795
      devs.append(rpath)
1796
  parent_bdev.RemoveChildren(devs)
1797

    
1798

    
1799
def BlockdevGetmirrorstatus(disks):
1800
  """Get the mirroring status of a list of devices.
1801

1802
  @type disks: list of L{objects.Disk}
1803
  @param disks: the list of disks which we should query
1804
  @rtype: disk
1805
  @return: List of L{objects.BlockDevStatus}, one for each disk
1806
  @raise errors.BlockDeviceError: if any of the disks cannot be
1807
      found
1808

1809
  """
1810
  stats = []
1811
  for dsk in disks:
1812
    rbd = _RecursiveFindBD(dsk)
1813
    if rbd is None:
1814
      _Fail("Can't find device %s", dsk)
1815

    
1816
    stats.append(rbd.CombinedSyncStatus())
1817

    
1818
  return stats
1819

    
1820

    
1821
def BlockdevGetmirrorstatusMulti(disks):
1822
  """Get the mirroring status of a list of devices.
1823

1824
  @type disks: list of L{objects.Disk}
1825
  @param disks: the list of disks which we should query
1826
  @rtype: disk
1827
  @return: List of tuples, (bool, status), one for each disk; bool denotes
1828
    success/failure, status is L{objects.BlockDevStatus} on success, string
1829
    otherwise
1830

1831
  """
1832
  result = []
1833
  for disk in disks:
1834
    try:
1835
      rbd = _RecursiveFindBD(disk)
1836
      if rbd is None:
1837
        result.append((False, "Can't find device %s" % disk))
1838
        continue
1839

    
1840
      status = rbd.CombinedSyncStatus()
1841
    except errors.BlockDeviceError, err:
1842
      logging.exception("Error while getting disk status")
1843
      result.append((False, str(err)))
1844
    else:
1845
      result.append((True, status))
1846

    
1847
  assert len(disks) == len(result)
1848

    
1849
  return result
1850

    
1851

    
1852
def _RecursiveFindBD(disk):
1853
  """Check if a device is activated.
1854

1855
  If so, return information about the real device.
1856

1857
  @type disk: L{objects.Disk}
1858
  @param disk: the disk object we need to find
1859

1860
  @return: None if the device can't be found,
1861
      otherwise the device instance
1862

1863
  """
1864
  children = []
1865
  if disk.children:
1866
    for chdisk in disk.children:
1867
      children.append(_RecursiveFindBD(chdisk))
1868

    
1869
  return bdev.FindDevice(disk.dev_type, disk.physical_id, children, disk.size)
1870

    
1871

    
1872
def _OpenRealBD(disk):
1873
  """Opens the underlying block device of a disk.
1874

1875
  @type disk: L{objects.Disk}
1876
  @param disk: the disk object we want to open
1877

1878
  """
1879
  real_disk = _RecursiveFindBD(disk)
1880
  if real_disk is None:
1881
    _Fail("Block device '%s' is not set up", disk)
1882

    
1883
  real_disk.Open()
1884

    
1885
  return real_disk
1886

    
1887

    
1888
def BlockdevFind(disk):
1889
  """Check if a device is activated.
1890

1891
  If it is, return information about the real device.
1892

1893
  @type disk: L{objects.Disk}
1894
  @param disk: the disk to find
1895
  @rtype: None or objects.BlockDevStatus
1896
  @return: None if the disk cannot be found, otherwise a the current
1897
           information
1898

1899
  """
1900
  try:
1901
    rbd = _RecursiveFindBD(disk)
1902
  except errors.BlockDeviceError, err:
1903
    _Fail("Failed to find device: %s", err, exc=True)
1904

    
1905
  if rbd is None:
1906
    return None
1907

    
1908
  return rbd.GetSyncStatus()
1909

    
1910

    
1911
def BlockdevGetsize(disks):
1912
  """Computes the size of the given disks.
1913

1914
  If a disk is not found, returns None instead.
1915

1916
  @type disks: list of L{objects.Disk}
1917
  @param disks: the list of disk to compute the size for
1918
  @rtype: list
1919
  @return: list with elements None if the disk cannot be found,
1920
      otherwise the size
1921

1922
  """
1923
  result = []
1924
  for cf in disks:
1925
    try:
1926
      rbd = _RecursiveFindBD(cf)
1927
    except errors.BlockDeviceError:
1928
      result.append(None)
1929
      continue
1930
    if rbd is None:
1931
      result.append(None)
1932
    else:
1933
      result.append(rbd.GetActualSize())
1934
  return result
1935

    
1936

    
1937
def BlockdevExport(disk, dest_node, dest_path, cluster_name):
1938
  """Export a block device to a remote node.
1939

1940
  @type disk: L{objects.Disk}
1941
  @param disk: the description of the disk to export
1942
  @type dest_node: str
1943
  @param dest_node: the destination node to export to
1944
  @type dest_path: str
1945
  @param dest_path: the destination path on the target node
1946
  @type cluster_name: str
1947
  @param cluster_name: the cluster name, needed for SSH hostalias
1948
  @rtype: None
1949

1950
  """
1951
  real_disk = _OpenRealBD(disk)
1952

    
1953
  # the block size on the read dd is 1MiB to match our units
1954
  expcmd = utils.BuildShellCmd("set -e; set -o pipefail; "
1955
                               "dd if=%s bs=1048576 count=%s",
1956
                               real_disk.dev_path, str(disk.size))
1957

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

    
1967
  remotecmd = _GetSshRunner(cluster_name).BuildCmd(dest_node,
1968
                                                   constants.GANETI_RUNAS,
1969
                                                   destcmd)
1970

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

    
1974
  result = utils.RunCmd(["bash", "-c", command])
1975

    
1976
  if result.failed:
1977
    _Fail("Disk copy command '%s' returned error: %s"
1978
          " output: %s", command, result.fail_reason, result.output)
1979

    
1980

    
1981
def UploadFile(file_name, data, mode, uid, gid, atime, mtime):
1982
  """Write a file to the filesystem.
1983

1984
  This allows the master to overwrite(!) a file. It will only perform
1985
  the operation if the file belongs to a list of configuration files.
1986

1987
  @type file_name: str
1988
  @param file_name: the target file name
1989
  @type data: str
1990
  @param data: the new contents of the file
1991
  @type mode: int
1992
  @param mode: the mode to give the file (can be None)
1993
  @type uid: string
1994
  @param uid: the owner of the file
1995
  @type gid: string
1996
  @param gid: the group of the file
1997
  @type atime: float
1998
  @param atime: the atime to set on the file (can be None)
1999
  @type mtime: float
2000
  @param mtime: the mtime to set on the file (can be None)
2001
  @rtype: None
2002

2003
  """
2004
  if not os.path.isabs(file_name):
2005
    _Fail("Filename passed to UploadFile is not absolute: '%s'", file_name)
2006

    
2007
  if file_name not in _ALLOWED_UPLOAD_FILES:
2008
    _Fail("Filename passed to UploadFile not in allowed upload targets: '%s'",
2009
          file_name)
2010

    
2011
  raw_data = _Decompress(data)
2012

    
2013
  if not (isinstance(uid, basestring) and isinstance(gid, basestring)):
2014
    _Fail("Invalid username/groupname type")
2015

    
2016
  getents = runtime.GetEnts()
2017
  uid = getents.LookupUser(uid)
2018
  gid = getents.LookupGroup(gid)
2019

    
2020
  utils.SafeWriteFile(file_name, None,
2021
                      data=raw_data, mode=mode, uid=uid, gid=gid,
2022
                      atime=atime, mtime=mtime)
2023

    
2024

    
2025
def RunOob(oob_program, command, node, timeout):
2026
  """Executes oob_program with given command on given node.
2027

2028
  @param oob_program: The path to the executable oob_program
2029
  @param command: The command to invoke on oob_program
2030
  @param node: The node given as an argument to the program
2031
  @param timeout: Timeout after which we kill the oob program
2032

2033
  @return: stdout
2034
  @raise RPCFail: If execution fails for some reason
2035

2036
  """
2037
  result = utils.RunCmd([oob_program, command, node], timeout=timeout)
2038

    
2039
  if result.failed:
2040
    _Fail("'%s' failed with reason '%s'; output: %s", result.cmd,
2041
          result.fail_reason, result.output)
2042

    
2043
  return result.stdout
2044

    
2045

    
2046
def WriteSsconfFiles(values):
2047
  """Update all ssconf files.
2048

2049
  Wrapper around the SimpleStore.WriteFiles.
2050

2051
  """
2052
  ssconf.SimpleStore().WriteFiles(values)
2053

    
2054

    
2055
def _ErrnoOrStr(err):
2056
  """Format an EnvironmentError exception.
2057

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

2062
  @type err: L{EnvironmentError}
2063
  @param err: the exception to format
2064

2065
  """
2066
  if hasattr(err, "errno"):
2067
    detail = errno.errorcode[err.errno]
2068
  else:
2069
    detail = str(err)
2070
  return detail
2071

    
2072

    
2073
def _OSOndiskAPIVersion(os_dir):
2074
  """Compute and return the API version of a given OS.
2075

2076
  This function will try to read the API version of the OS residing in
2077
  the 'os_dir' directory.
2078

2079
  @type os_dir: str
2080
  @param os_dir: the directory in which we should look for the OS
2081
  @rtype: tuple
2082
  @return: tuple (status, data) with status denoting the validity and
2083
      data holding either the vaid versions or an error message
2084

2085
  """
2086
  api_file = utils.PathJoin(os_dir, constants.OS_API_FILE)
2087

    
2088
  try:
2089
    st = os.stat(api_file)
2090
  except EnvironmentError, err:
2091
    return False, ("Required file '%s' not found under path %s: %s" %
2092
                   (constants.OS_API_FILE, os_dir, _ErrnoOrStr(err)))
2093

    
2094
  if not stat.S_ISREG(stat.S_IFMT(st.st_mode)):
2095
    return False, ("File '%s' in %s is not a regular file" %
2096
                   (constants.OS_API_FILE, os_dir))
2097

    
2098
  try:
2099
    api_versions = utils.ReadFile(api_file).splitlines()
2100
  except EnvironmentError, err:
2101
    return False, ("Error while reading the API version file at %s: %s" %
2102
                   (api_file, _ErrnoOrStr(err)))
2103

    
2104
  try:
2105
    api_versions = [int(version.strip()) for version in api_versions]
2106
  except (TypeError, ValueError), err:
2107
    return False, ("API version(s) can't be converted to integer: %s" %
2108
                   str(err))
2109

    
2110
  return True, api_versions
2111

    
2112

    
2113
def DiagnoseOS(top_dirs=None):
2114
  """Compute the validity for all OSes.
2115

2116
  @type top_dirs: list
2117
  @param top_dirs: the list of directories in which to
2118
      search (if not given defaults to
2119
      L{constants.OS_SEARCH_PATH})
2120
  @rtype: list of L{objects.OS}
2121
  @return: a list of tuples (name, path, status, diagnose, variants,
2122
      parameters, api_version) for all (potential) OSes under all
2123
      search paths, where:
2124
          - name is the (potential) OS name
2125
          - path is the full path to the OS
2126
          - status True/False is the validity of the OS
2127
          - diagnose is the error message for an invalid OS, otherwise empty
2128
          - variants is a list of supported OS variants, if any
2129
          - parameters is a list of (name, help) parameters, if any
2130
          - api_version is a list of support OS API versions
2131

2132
  """
2133
  if top_dirs is None:
2134
    top_dirs = constants.OS_SEARCH_PATH
2135

    
2136
  result = []
2137
  for dir_name in top_dirs:
2138
    if os.path.isdir(dir_name):
2139
      try:
2140
        f_names = utils.ListVisibleFiles(dir_name)
2141
      except EnvironmentError, err:
2142
        logging.exception("Can't list the OS directory %s: %s", dir_name, err)
2143
        break
2144
      for name in f_names:
2145
        os_path = utils.PathJoin(dir_name, name)
2146
        status, os_inst = _TryOSFromDisk(name, base_dir=dir_name)
2147
        if status:
2148
          diagnose = ""
2149
          variants = os_inst.supported_variants
2150
          parameters = os_inst.supported_parameters
2151
          api_versions = os_inst.api_versions
2152
        else:
2153
          diagnose = os_inst
2154
          variants = parameters = api_versions = []
2155
        result.append((name, os_path, status, diagnose, variants,
2156
                       parameters, api_versions))
2157

    
2158
  return result
2159

    
2160

    
2161
def _TryOSFromDisk(name, base_dir=None):
2162
  """Create an OS instance from disk.
2163

2164
  This function will return an OS instance if the given name is a
2165
  valid OS name.
2166

2167
  @type base_dir: string
2168
  @keyword base_dir: Base directory containing OS installations.
2169
                     Defaults to a search in all the OS_SEARCH_PATH dirs.
2170
  @rtype: tuple
2171
  @return: success and either the OS instance if we find a valid one,
2172
      or error message
2173

2174
  """
2175
  if base_dir is None:
2176
    os_dir = utils.FindFile(name, constants.OS_SEARCH_PATH, os.path.isdir)
2177
  else:
2178
    os_dir = utils.FindFile(name, [base_dir], os.path.isdir)
2179

    
2180
  if os_dir is None:
2181
    return False, "Directory for OS %s not found in search path" % name
2182

    
2183
  status, api_versions = _OSOndiskAPIVersion(os_dir)
2184
  if not status:
2185
    # push the error up
2186
    return status, api_versions
2187

    
2188
  if not constants.OS_API_VERSIONS.intersection(api_versions):
2189
    return False, ("API version mismatch for path '%s': found %s, want %s." %
2190
                   (os_dir, api_versions, constants.OS_API_VERSIONS))
2191

    
2192
  # OS Files dictionary, we will populate it with the absolute path
2193
  # names; if the value is True, then it is a required file, otherwise
2194
  # an optional one
2195
  os_files = dict.fromkeys(constants.OS_SCRIPTS, True)
2196

    
2197
  if max(api_versions) >= constants.OS_API_V15:
2198
    os_files[constants.OS_VARIANTS_FILE] = False
2199

    
2200
  if max(api_versions) >= constants.OS_API_V20:
2201
    os_files[constants.OS_PARAMETERS_FILE] = True
2202
  else:
2203
    del os_files[constants.OS_SCRIPT_VERIFY]
2204

    
2205
  for (filename, required) in os_files.items():
2206
    os_files[filename] = utils.PathJoin(os_dir, filename)
2207

    
2208
    try:
2209
      st = os.stat(os_files[filename])
2210
    except EnvironmentError, err:
2211
      if err.errno == errno.ENOENT and not required:
2212
        del os_files[filename]
2213
        continue
2214
      return False, ("File '%s' under path '%s' is missing (%s)" %
2215
                     (filename, os_dir, _ErrnoOrStr(err)))
2216

    
2217
    if not stat.S_ISREG(stat.S_IFMT(st.st_mode)):
2218
      return False, ("File '%s' under path '%s' is not a regular file" %
2219
                     (filename, os_dir))
2220

    
2221
    if filename in constants.OS_SCRIPTS:
2222
      if stat.S_IMODE(st.st_mode) & stat.S_IXUSR != stat.S_IXUSR:
2223
        return False, ("File '%s' under path '%s' is not executable" %
2224
                       (filename, os_dir))
2225

    
2226
  variants = []
2227
  if constants.OS_VARIANTS_FILE in os_files:
2228
    variants_file = os_files[constants.OS_VARIANTS_FILE]
2229
    try:
2230
      variants = utils.ReadFile(variants_file).splitlines()
2231
    except EnvironmentError, err:
2232
      # we accept missing files, but not other errors
2233
      if err.errno != errno.ENOENT:
2234
        return False, ("Error while reading the OS variants file at %s: %s" %
2235
                       (variants_file, _ErrnoOrStr(err)))
2236

    
2237
  parameters = []
2238
  if constants.OS_PARAMETERS_FILE in os_files:
2239
    parameters_file = os_files[constants.OS_PARAMETERS_FILE]
2240
    try:
2241
      parameters = utils.ReadFile(parameters_file).splitlines()
2242
    except EnvironmentError, err:
2243
      return False, ("Error while reading the OS parameters file at %s: %s" %
2244
                     (parameters_file, _ErrnoOrStr(err)))
2245
    parameters = [v.split(None, 1) for v in parameters]
2246

    
2247
  os_obj = objects.OS(name=name, path=os_dir,
2248
                      create_script=os_files[constants.OS_SCRIPT_CREATE],
2249
                      export_script=os_files[constants.OS_SCRIPT_EXPORT],
2250
                      import_script=os_files[constants.OS_SCRIPT_IMPORT],
2251
                      rename_script=os_files[constants.OS_SCRIPT_RENAME],
2252
                      verify_script=os_files.get(constants.OS_SCRIPT_VERIFY,
2253
                                                 None),
2254
                      supported_variants=variants,
2255
                      supported_parameters=parameters,
2256
                      api_versions=api_versions)
2257
  return True, os_obj
2258

    
2259

    
2260
def OSFromDisk(name, base_dir=None):
2261
  """Create an OS instance from disk.
2262

2263
  This function will return an OS instance if the given name is a
2264
  valid OS name. Otherwise, it will raise an appropriate
2265
  L{RPCFail} exception, detailing why this is not a valid OS.
2266

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

2270
  @type base_dir: string
2271
  @keyword base_dir: Base directory containing OS installations.
2272
                     Defaults to a search in all the OS_SEARCH_PATH dirs.
2273
  @rtype: L{objects.OS}
2274
  @return: the OS instance if we find a valid one
2275
  @raise RPCFail: if we don't find a valid OS
2276

2277
  """
2278
  name_only = objects.OS.GetName(name)
2279
  status, payload = _TryOSFromDisk(name_only, base_dir)
2280

    
2281
  if not status:
2282
    _Fail(payload)
2283

    
2284
  return payload
2285

    
2286

    
2287
def OSCoreEnv(os_name, inst_os, os_params, debug=0):
2288
  """Calculate the basic environment for an os script.
2289

2290
  @type os_name: str
2291
  @param os_name: full operating system name (including variant)
2292
  @type inst_os: L{objects.OS}
2293
  @param inst_os: operating system for which the environment is being built
2294
  @type os_params: dict
2295
  @param os_params: the OS parameters
2296
  @type debug: integer
2297
  @param debug: debug level (0 or 1, for OS Api 10)
2298
  @rtype: dict
2299
  @return: dict of environment variables
2300
  @raise errors.BlockDeviceError: if the block device
2301
      cannot be found
2302

2303
  """
2304
  result = {}
2305
  api_version = \
2306
    max(constants.OS_API_VERSIONS.intersection(inst_os.api_versions))
2307
  result["OS_API_VERSION"] = "%d" % api_version
2308
  result["OS_NAME"] = inst_os.name
2309
  result["DEBUG_LEVEL"] = "%d" % debug
2310

    
2311
  # OS variants
2312
  if api_version >= constants.OS_API_V15 and inst_os.supported_variants:
2313
    variant = objects.OS.GetVariant(os_name)
2314
    if not variant:
2315
      variant = inst_os.supported_variants[0]
2316
  else:
2317
    variant = ""
2318
  result["OS_VARIANT"] = variant
2319

    
2320
  # OS params
2321
  for pname, pvalue in os_params.items():
2322
    result["OSP_%s" % pname.upper()] = pvalue
2323

    
2324
  return result
2325

    
2326

    
2327
def OSEnvironment(instance, inst_os, debug=0):
2328
  """Calculate the environment for an os script.
2329

2330
  @type instance: L{objects.Instance}
2331
  @param instance: target instance for the os script run
2332
  @type inst_os: L{objects.OS}
2333
  @param inst_os: operating system for which the environment is being built
2334
  @type debug: integer
2335
  @param debug: debug level (0 or 1, for OS Api 10)
2336
  @rtype: dict
2337
  @return: dict of environment variables
2338
  @raise errors.BlockDeviceError: if the block device
2339
      cannot be found
2340

2341
  """
2342
  result = OSCoreEnv(instance.os, inst_os, instance.osparams, debug=debug)
2343

    
2344
  for attr in ["name", "os", "uuid", "ctime", "mtime", "primary_node"]:
2345
    result["INSTANCE_%s" % attr.upper()] = str(getattr(instance, attr))
2346

    
2347
  result["HYPERVISOR"] = instance.hypervisor
2348
  result["DISK_COUNT"] = "%d" % len(instance.disks)
2349
  result["NIC_COUNT"] = "%d" % len(instance.nics)
2350
  result["INSTANCE_SECONDARY_NODES"] = \
2351
      ("%s" % " ".join(instance.secondary_nodes))
2352

    
2353
  # Disks
2354
  for idx, disk in enumerate(instance.disks):
2355
    real_disk = _OpenRealBD(disk)
2356
    result["DISK_%d_PATH" % idx] = real_disk.dev_path
2357
    result["DISK_%d_ACCESS" % idx] = disk.mode
2358
    if constants.HV_DISK_TYPE in instance.hvparams:
2359
      result["DISK_%d_FRONTEND_TYPE" % idx] = \
2360
        instance.hvparams[constants.HV_DISK_TYPE]
2361
    if disk.dev_type in constants.LDS_BLOCK:
2362
      result["DISK_%d_BACKEND_TYPE" % idx] = "block"
2363
    elif disk.dev_type == constants.LD_FILE:
2364
      result["DISK_%d_BACKEND_TYPE" % idx] = \
2365
        "file:%s" % disk.physical_id[0]
2366

    
2367
  # NICs
2368
  for idx, nic in enumerate(instance.nics):
2369
    result["NIC_%d_MAC" % idx] = nic.mac
2370
    if nic.ip:
2371
      result["NIC_%d_IP" % idx] = nic.ip
2372
    result["NIC_%d_MODE" % idx] = nic.nicparams[constants.NIC_MODE]
2373
    if nic.nicparams[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED:
2374
      result["NIC_%d_BRIDGE" % idx] = nic.nicparams[constants.NIC_LINK]
2375
    if nic.nicparams[constants.NIC_LINK]:
2376
      result["NIC_%d_LINK" % idx] = nic.nicparams[constants.NIC_LINK]
2377
    if constants.HV_NIC_TYPE in instance.hvparams:
2378
      result["NIC_%d_FRONTEND_TYPE" % idx] = \
2379
        instance.hvparams[constants.HV_NIC_TYPE]
2380

    
2381
  # HV/BE params
2382
  for source, kind in [(instance.beparams, "BE"), (instance.hvparams, "HV")]:
2383
    for key, value in source.items():
2384
      result["INSTANCE_%s_%s" % (kind, key)] = str(value)
2385

    
2386
  return result
2387

    
2388

    
2389
def BlockdevGrow(disk, amount, dryrun):
2390
  """Grow a stack of block devices.
2391

2392
  This function is called recursively, with the childrens being the
2393
  first ones to resize.
2394

2395
  @type disk: L{objects.Disk}
2396
  @param disk: the disk to be grown
2397
  @type amount: integer
2398
  @param amount: the amount (in mebibytes) to grow with
2399
  @type dryrun: boolean
2400
  @param dryrun: whether to execute the operation in simulation mode
2401
      only, without actually increasing the size
2402
  @rtype: (status, result)
2403
  @return: a tuple with the status of the operation (True/False), and
2404
      the errors message if status is False
2405

2406
  """
2407
  r_dev = _RecursiveFindBD(disk)
2408
  if r_dev is None:
2409
    _Fail("Cannot find block device %s", disk)
2410

    
2411
  try:
2412
    r_dev.Grow(amount, dryrun)
2413
  except errors.BlockDeviceError, err:
2414
    _Fail("Failed to grow block device: %s", err, exc=True)
2415

    
2416

    
2417
def BlockdevSnapshot(disk):
2418
  """Create a snapshot copy of a block device.
2419

2420
  This function is called recursively, and the snapshot is actually created
2421
  just for the leaf lvm backend device.
2422

2423
  @type disk: L{objects.Disk}
2424
  @param disk: the disk to be snapshotted
2425
  @rtype: string
2426
  @return: snapshot disk ID as (vg, lv)
2427

2428
  """
2429
  if disk.dev_type == constants.LD_DRBD8:
2430
    if not disk.children:
2431
      _Fail("DRBD device '%s' without backing storage cannot be snapshotted",
2432
            disk.unique_id)
2433
    return BlockdevSnapshot(disk.children[0])
2434
  elif disk.dev_type == constants.LD_LV:
2435
    r_dev = _RecursiveFindBD(disk)
2436
    if r_dev is not None:
2437
      # FIXME: choose a saner value for the snapshot size
2438
      # let's stay on the safe side and ask for the full size, for now
2439
      return r_dev.Snapshot(disk.size)
2440
    else:
2441
      _Fail("Cannot find block device %s", disk)
2442
  else:
2443
    _Fail("Cannot snapshot non-lvm block device '%s' of type '%s'",
2444
          disk.unique_id, disk.dev_type)
2445

    
2446

    
2447
def FinalizeExport(instance, snap_disks):
2448
  """Write out the export configuration information.
2449

2450
  @type instance: L{objects.Instance}
2451
  @param instance: the instance which we export, used for
2452
      saving configuration
2453
  @type snap_disks: list of L{objects.Disk}
2454
  @param snap_disks: list of snapshot block devices, which
2455
      will be used to get the actual name of the dump file
2456

2457
  @rtype: None
2458

2459
  """
2460
  destdir = utils.PathJoin(constants.EXPORT_DIR, instance.name + ".new")
2461
  finaldestdir = utils.PathJoin(constants.EXPORT_DIR, instance.name)
2462

    
2463
  config = objects.SerializableConfigParser()
2464

    
2465
  config.add_section(constants.INISECT_EXP)
2466
  config.set(constants.INISECT_EXP, "version", "0")
2467
  config.set(constants.INISECT_EXP, "timestamp", "%d" % int(time.time()))
2468
  config.set(constants.INISECT_EXP, "source", instance.primary_node)
2469
  config.set(constants.INISECT_EXP, "os", instance.os)
2470
  config.set(constants.INISECT_EXP, "compression", "none")
2471

    
2472
  config.add_section(constants.INISECT_INS)
2473
  config.set(constants.INISECT_INS, "name", instance.name)
2474
  config.set(constants.INISECT_INS, "memory", "%d" %
2475
             instance.beparams[constants.BE_MEMORY])
2476
  config.set(constants.INISECT_INS, "vcpus", "%d" %
2477
             instance.beparams[constants.BE_VCPUS])
2478
  config.set(constants.INISECT_INS, "disk_template", instance.disk_template)
2479
  config.set(constants.INISECT_INS, "hypervisor", instance.hypervisor)
2480
  config.set(constants.INISECT_INS, "tags", " ".join(instance.GetTags()))
2481

    
2482
  nic_total = 0
2483
  for nic_count, nic in enumerate(instance.nics):
2484
    nic_total += 1
2485
    config.set(constants.INISECT_INS, "nic%d_mac" %
2486
               nic_count, "%s" % nic.mac)
2487
    config.set(constants.INISECT_INS, "nic%d_ip" % nic_count, "%s" % nic.ip)
2488
    for param in constants.NICS_PARAMETER_TYPES:
2489
      config.set(constants.INISECT_INS, "nic%d_%s" % (nic_count, param),
2490
                 "%s" % nic.nicparams.get(param, None))
2491
  # TODO: redundant: on load can read nics until it doesn't exist
2492
  config.set(constants.INISECT_INS, "nic_count", "%d" % nic_total)
2493

    
2494
  disk_total = 0
2495
  for disk_count, disk in enumerate(snap_disks):
2496
    if disk:
2497
      disk_total += 1
2498
      config.set(constants.INISECT_INS, "disk%d_ivname" % disk_count,
2499
                 ("%s" % disk.iv_name))
2500
      config.set(constants.INISECT_INS, "disk%d_dump" % disk_count,
2501
                 ("%s" % disk.physical_id[1]))
2502
      config.set(constants.INISECT_INS, "disk%d_size" % disk_count,
2503
                 ("%d" % disk.size))
2504

    
2505
  config.set(constants.INISECT_INS, "disk_count", "%d" % disk_total)
2506

    
2507
  # New-style hypervisor/backend parameters
2508

    
2509
  config.add_section(constants.INISECT_HYP)
2510
  for name, value in instance.hvparams.items():
2511
    if name not in constants.HVC_GLOBALS:
2512
      config.set(constants.INISECT_HYP, name, str(value))
2513

    
2514
  config.add_section(constants.INISECT_BEP)
2515
  for name, value in instance.beparams.items():
2516
    config.set(constants.INISECT_BEP, name, str(value))
2517

    
2518
  config.add_section(constants.INISECT_OSP)
2519
  for name, value in instance.osparams.items():
2520
    config.set(constants.INISECT_OSP, name, str(value))
2521

    
2522
  utils.WriteFile(utils.PathJoin(destdir, constants.EXPORT_CONF_FILE),
2523
                  data=config.Dumps())
2524
  shutil.rmtree(finaldestdir, ignore_errors=True)
2525
  shutil.move(destdir, finaldestdir)
2526

    
2527

    
2528
def ExportInfo(dest):
2529
  """Get export configuration information.
2530

2531
  @type dest: str
2532
  @param dest: directory containing the export
2533

2534
  @rtype: L{objects.SerializableConfigParser}
2535
  @return: a serializable config file containing the
2536
      export info
2537

2538
  """
2539
  cff = utils.PathJoin(dest, constants.EXPORT_CONF_FILE)
2540

    
2541
  config = objects.SerializableConfigParser()
2542
  config.read(cff)
2543

    
2544
  if (not config.has_section(constants.INISECT_EXP) or
2545
      not config.has_section(constants.INISECT_INS)):
2546
    _Fail("Export info file doesn't have the required fields")
2547

    
2548
  return config.Dumps()
2549

    
2550

    
2551
def ListExports():
2552
  """Return a list of exports currently available on this machine.
2553

2554
  @rtype: list
2555
  @return: list of the exports
2556

2557
  """
2558
  if os.path.isdir(constants.EXPORT_DIR):
2559
    return sorted(utils.ListVisibleFiles(constants.EXPORT_DIR))
2560
  else:
2561
    _Fail("No exports directory")
2562

    
2563

    
2564
def RemoveExport(export):
2565
  """Remove an existing export from the node.
2566

2567
  @type export: str
2568
  @param export: the name of the export to remove
2569
  @rtype: None
2570

2571
  """
2572
  target = utils.PathJoin(constants.EXPORT_DIR, export)
2573

    
2574
  try:
2575
    shutil.rmtree(target)
2576
  except EnvironmentError, err:
2577
    _Fail("Error while removing the export: %s", err, exc=True)
2578

    
2579

    
2580
def BlockdevRename(devlist):
2581
  """Rename a list of block devices.
2582

2583
  @type devlist: list of tuples
2584
  @param devlist: list of tuples of the form  (disk,
2585
      new_logical_id, new_physical_id); disk is an
2586
      L{objects.Disk} object describing the current disk,
2587
      and new logical_id/physical_id is the name we
2588
      rename it to
2589
  @rtype: boolean
2590
  @return: True if all renames succeeded, False otherwise
2591

2592
  """
2593
  msgs = []
2594
  result = True
2595
  for disk, unique_id in devlist:
2596
    dev = _RecursiveFindBD(disk)
2597
    if dev is None:
2598
      msgs.append("Can't find device %s in rename" % str(disk))
2599
      result = False
2600
      continue
2601
    try:
2602
      old_rpath = dev.dev_path
2603
      dev.Rename(unique_id)
2604
      new_rpath = dev.dev_path
2605
      if old_rpath != new_rpath:
2606
        DevCacheManager.RemoveCache(old_rpath)
2607
        # FIXME: we should add the new cache information here, like:
2608
        # DevCacheManager.UpdateCache(new_rpath, owner, ...)
2609
        # but we don't have the owner here - maybe parse from existing
2610
        # cache? for now, we only lose lvm data when we rename, which
2611
        # is less critical than DRBD or MD
2612
    except errors.BlockDeviceError, err:
2613
      msgs.append("Can't rename device '%s' to '%s': %s" %
2614
                  (dev, unique_id, err))
2615
      logging.exception("Can't rename device '%s' to '%s'", dev, unique_id)
2616
      result = False
2617
  if not result:
2618
    _Fail("; ".join(msgs))
2619

    
2620

    
2621
def _TransformFileStorageDir(fs_dir):
2622
  """Checks whether given file_storage_dir is valid.
2623

2624
  Checks wheter the given fs_dir is within the cluster-wide default
2625
  file_storage_dir or the shared_file_storage_dir, which are stored in
2626
  SimpleStore. Only paths under those directories are allowed.
2627

2628
  @type fs_dir: str
2629
  @param fs_dir: the path to check
2630

2631
  @return: the normalized path if valid, None otherwise
2632

2633
  """
2634
  if not constants.ENABLE_FILE_STORAGE:
2635
    _Fail("File storage disabled at configure time")
2636
  cfg = _GetConfig()
2637
  fs_dir = os.path.normpath(fs_dir)
2638
  base_fstore = cfg.GetFileStorageDir()
2639
  base_shared = cfg.GetSharedFileStorageDir()
2640
  if not (utils.IsBelowDir(base_fstore, fs_dir) or
2641
          utils.IsBelowDir(base_shared, fs_dir)):
2642
    _Fail("File storage directory '%s' is not under base file"
2643
          " storage directory '%s' or shared storage directory '%s'",
2644
          fs_dir, base_fstore, base_shared)
2645
  return fs_dir
2646

    
2647

    
2648
def CreateFileStorageDir(file_storage_dir):
2649
  """Create file storage directory.
2650

2651
  @type file_storage_dir: str
2652
  @param file_storage_dir: directory to create
2653

2654
  @rtype: tuple
2655
  @return: tuple with first element a boolean indicating wheter dir
2656
      creation was successful or not
2657

2658
  """
2659
  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
2660
  if os.path.exists(file_storage_dir):
2661
    if not os.path.isdir(file_storage_dir):
2662
      _Fail("Specified storage dir '%s' is not a directory",
2663
            file_storage_dir)
2664
  else:
2665
    try:
2666
      os.makedirs(file_storage_dir, 0750)
2667
    except OSError, err:
2668
      _Fail("Cannot create file storage directory '%s': %s",
2669
            file_storage_dir, err, exc=True)
2670

    
2671

    
2672
def RemoveFileStorageDir(file_storage_dir):
2673
  """Remove file storage directory.
2674

2675
  Remove it only if it's empty. If not log an error and return.
2676

2677
  @type file_storage_dir: str
2678
  @param file_storage_dir: the directory we should cleanup
2679
  @rtype: tuple (success,)
2680
  @return: tuple of one element, C{success}, denoting
2681
      whether the operation was successful
2682

2683
  """
2684
  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
2685
  if os.path.exists(file_storage_dir):
2686
    if not os.path.isdir(file_storage_dir):
2687
      _Fail("Specified Storage directory '%s' is not a directory",
2688
            file_storage_dir)
2689
    # deletes dir only if empty, otherwise we want to fail the rpc call
2690
    try:
2691
      os.rmdir(file_storage_dir)
2692
    except OSError, err:
2693
      _Fail("Cannot remove file storage directory '%s': %s",
2694
            file_storage_dir, err)
2695

    
2696

    
2697
def RenameFileStorageDir(old_file_storage_dir, new_file_storage_dir):
2698
  """Rename the file storage directory.
2699

2700
  @type old_file_storage_dir: str
2701
  @param old_file_storage_dir: the current path
2702
  @type new_file_storage_dir: str
2703
  @param new_file_storage_dir: the name we should rename to
2704
  @rtype: tuple (success,)
2705
  @return: tuple of one element, C{success}, denoting
2706
      whether the operation was successful
2707

2708
  """
2709
  old_file_storage_dir = _TransformFileStorageDir(old_file_storage_dir)
2710
  new_file_storage_dir = _TransformFileStorageDir(new_file_storage_dir)
2711
  if not os.path.exists(new_file_storage_dir):
2712
    if os.path.isdir(old_file_storage_dir):
2713
      try:
2714
        os.rename(old_file_storage_dir, new_file_storage_dir)
2715
      except OSError, err:
2716
        _Fail("Cannot rename '%s' to '%s': %s",
2717
              old_file_storage_dir, new_file_storage_dir, err)
2718
    else:
2719
      _Fail("Specified storage dir '%s' is not a directory",
2720
            old_file_storage_dir)
2721
  else:
2722
    if os.path.exists(old_file_storage_dir):
2723
      _Fail("Cannot rename '%s' to '%s': both locations exist",
2724
            old_file_storage_dir, new_file_storage_dir)
2725

    
2726

    
2727
def _EnsureJobQueueFile(file_name):
2728
  """Checks whether the given filename is in the queue directory.
2729

2730
  @type file_name: str
2731
  @param file_name: the file name we should check
2732
  @rtype: None
2733
  @raises RPCFail: if the file is not valid
2734

2735
  """
2736
  queue_dir = os.path.normpath(constants.QUEUE_DIR)
2737
  result = (os.path.commonprefix([queue_dir, file_name]) == queue_dir)
2738

    
2739
  if not result:
2740
    _Fail("Passed job queue file '%s' does not belong to"
2741
          " the queue directory '%s'", file_name, queue_dir)
2742

    
2743

    
2744
def JobQueueUpdate(file_name, content):
2745
  """Updates a file in the queue directory.
2746

2747
  This is just a wrapper over L{utils.io.WriteFile}, with proper
2748
  checking.
2749

2750
  @type file_name: str
2751
  @param file_name: the job file name
2752
  @type content: str
2753
  @param content: the new job contents
2754
  @rtype: boolean
2755
  @return: the success of the operation
2756

2757
  """
2758
  _EnsureJobQueueFile(file_name)
2759
  getents = runtime.GetEnts()
2760

    
2761
  # Write and replace the file atomically
2762
  utils.WriteFile(file_name, data=_Decompress(content), uid=getents.masterd_uid,
2763
                  gid=getents.masterd_gid)
2764

    
2765

    
2766
def JobQueueRename(old, new):
2767
  """Renames a job queue file.
2768

2769
  This is just a wrapper over os.rename with proper checking.
2770

2771
  @type old: str
2772
  @param old: the old (actual) file name
2773
  @type new: str
2774
  @param new: the desired file name
2775
  @rtype: tuple
2776
  @return: the success of the operation and payload
2777

2778
  """
2779
  _EnsureJobQueueFile(old)
2780
  _EnsureJobQueueFile(new)
2781

    
2782
  getents = runtime.GetEnts()
2783

    
2784
  utils.RenameFile(old, new, mkdir=True, mkdir_mode=0700,
2785
                   dir_uid=getents.masterd_uid, dir_gid=getents.masterd_gid)
2786

    
2787

    
2788
def BlockdevClose(instance_name, disks):
2789
  """Closes the given block devices.
2790

2791
  This means they will be switched to secondary mode (in case of
2792
  DRBD).
2793

2794
  @param instance_name: if the argument is not empty, the symlinks
2795
      of this instance will be removed
2796
  @type disks: list of L{objects.Disk}
2797
  @param disks: the list of disks to be closed
2798
  @rtype: tuple (success, message)
2799
  @return: a tuple of success and message, where success
2800
      indicates the succes of the operation, and message
2801
      which will contain the error details in case we
2802
      failed
2803

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

    
2812
  msg = []
2813
  for rd in bdevs:
2814
    try:
2815
      rd.Close()
2816
    except errors.BlockDeviceError, err:
2817
      msg.append(str(err))
2818
  if msg:
2819
    _Fail("Can't make devices secondary: %s", ",".join(msg))
2820
  else:
2821
    if instance_name:
2822
      _RemoveBlockDevLinks(instance_name, disks)
2823

    
2824

    
2825
def ValidateHVParams(hvname, hvparams):
2826
  """Validates the given hypervisor parameters.
2827

2828
  @type hvname: string
2829
  @param hvname: the hypervisor name
2830
  @type hvparams: dict
2831
  @param hvparams: the hypervisor parameters to be validated
2832
  @rtype: None
2833

2834
  """
2835
  try:
2836
    hv_type = hypervisor.GetHypervisor(hvname)
2837
    hv_type.ValidateParameters(hvparams)
2838
  except errors.HypervisorError, err:
2839
    _Fail(str(err), log=False)
2840

    
2841

    
2842
def _CheckOSPList(os_obj, parameters):
2843
  """Check whether a list of parameters is supported by the OS.
2844

2845
  @type os_obj: L{objects.OS}
2846
  @param os_obj: OS object to check
2847
  @type parameters: list
2848
  @param parameters: the list of parameters to check
2849

2850
  """
2851
  supported = [v[0] for v in os_obj.supported_parameters]
2852
  delta = frozenset(parameters).difference(supported)
2853
  if delta:
2854
    _Fail("The following parameters are not supported"
2855
          " by the OS %s: %s" % (os_obj.name, utils.CommaJoin(delta)))
2856

    
2857

    
2858
def ValidateOS(required, osname, checks, osparams):
2859
  """Validate the given OS' parameters.
2860

2861
  @type required: boolean
2862
  @param required: whether absence of the OS should translate into
2863
      failure or not
2864
  @type osname: string
2865
  @param osname: the OS to be validated
2866
  @type checks: list
2867
  @param checks: list of the checks to run (currently only 'parameters')
2868
  @type osparams: dict
2869
  @param osparams: dictionary with OS parameters
2870
  @rtype: boolean
2871
  @return: True if the validation passed, or False if the OS was not
2872
      found and L{required} was false
2873

2874
  """
2875
  if not constants.OS_VALIDATE_CALLS.issuperset(checks):
2876
    _Fail("Unknown checks required for OS %s: %s", osname,
2877
          set(checks).difference(constants.OS_VALIDATE_CALLS))
2878

    
2879
  name_only = objects.OS.GetName(osname)
2880
  status, tbv = _TryOSFromDisk(name_only, None)
2881

    
2882
  if not status:
2883
    if required:
2884
      _Fail(tbv)
2885
    else:
2886
      return False
2887

    
2888
  if max(tbv.api_versions) < constants.OS_API_V20:
2889
    return True
2890

    
2891
  if constants.OS_VALIDATE_PARAMETERS in checks:
2892
    _CheckOSPList(tbv, osparams.keys())
2893

    
2894
  validate_env = OSCoreEnv(osname, tbv, osparams)
2895
  result = utils.RunCmd([tbv.verify_script] + checks, env=validate_env,
2896
                        cwd=tbv.path, reset_env=True)
2897
  if result.failed:
2898
    logging.error("os validate command '%s' returned error: %s output: %s",
2899
                  result.cmd, result.fail_reason, result.output)
2900
    _Fail("OS validation script failed (%s), output: %s",
2901
          result.fail_reason, result.output, log=False)
2902

    
2903
  return True
2904

    
2905

    
2906
def DemoteFromMC():
2907
  """Demotes the current node from master candidate role.
2908

2909
  """
2910
  # try to ensure we're not the master by mistake
2911
  master, myself = ssconf.GetMasterAndMyself()
2912
  if master == myself:
2913
    _Fail("ssconf status shows I'm the master node, will not demote")
2914

    
2915
  result = utils.RunCmd([constants.DAEMON_UTIL, "check", constants.MASTERD])
2916
  if not result.failed:
2917
    _Fail("The master daemon is running, will not demote")
2918

    
2919
  try:
2920
    if os.path.isfile(constants.CLUSTER_CONF_FILE):
2921
      utils.CreateBackup(constants.CLUSTER_CONF_FILE)
2922
  except EnvironmentError, err:
2923
    if err.errno != errno.ENOENT:
2924
      _Fail("Error while backing up cluster file: %s", err, exc=True)
2925

    
2926
  utils.RemoveFile(constants.CLUSTER_CONF_FILE)
2927

    
2928

    
2929
def _GetX509Filenames(cryptodir, name):
2930
  """Returns the full paths for the private key and certificate.
2931

2932
  """
2933
  return (utils.PathJoin(cryptodir, name),
2934
          utils.PathJoin(cryptodir, name, _X509_KEY_FILE),
2935
          utils.PathJoin(cryptodir, name, _X509_CERT_FILE))
2936

    
2937

    
2938
def CreateX509Certificate(validity, cryptodir=constants.CRYPTO_KEYS_DIR):
2939
  """Creates a new X509 certificate for SSL/TLS.
2940

2941
  @type validity: int
2942
  @param validity: Validity in seconds
2943
  @rtype: tuple; (string, string)
2944
  @return: Certificate name and public part
2945

2946
  """
2947
  (key_pem, cert_pem) = \
2948
    utils.GenerateSelfSignedX509Cert(netutils.Hostname.GetSysName(),
2949
                                     min(validity, _MAX_SSL_CERT_VALIDITY))
2950

    
2951
  cert_dir = tempfile.mkdtemp(dir=cryptodir,
2952
                              prefix="x509-%s-" % utils.TimestampForFilename())
2953
  try:
2954
    name = os.path.basename(cert_dir)
2955
    assert len(name) > 5
2956

    
2957
    (_, key_file, cert_file) = _GetX509Filenames(cryptodir, name)
2958

    
2959
    utils.WriteFile(key_file, mode=0400, data=key_pem)
2960
    utils.WriteFile(cert_file, mode=0400, data=cert_pem)
2961

    
2962
    # Never return private key as it shouldn't leave the node
2963
    return (name, cert_pem)
2964
  except Exception:
2965
    shutil.rmtree(cert_dir, ignore_errors=True)
2966
    raise
2967

    
2968

    
2969
def RemoveX509Certificate(name, cryptodir=constants.CRYPTO_KEYS_DIR):
2970
  """Removes a X509 certificate.
2971

2972
  @type name: string
2973
  @param name: Certificate name
2974

2975
  """
2976
  (cert_dir, key_file, cert_file) = _GetX509Filenames(cryptodir, name)
2977

    
2978
  utils.RemoveFile(key_file)
2979
  utils.RemoveFile(cert_file)
2980

    
2981
  try:
2982
    os.rmdir(cert_dir)
2983
  except EnvironmentError, err:
2984
    _Fail("Cannot remove certificate directory '%s': %s",
2985
          cert_dir, err)
2986

    
2987

    
2988
def _GetImportExportIoCommand(instance, mode, ieio, ieargs):
2989
  """Returns the command for the requested input/output.
2990

2991
  @type instance: L{objects.Instance}
2992
  @param instance: The instance object
2993
  @param mode: Import/export mode
2994
  @param ieio: Input/output type
2995
  @param ieargs: Input/output arguments
2996

2997
  """
2998
  assert mode in (constants.IEM_IMPORT, constants.IEM_EXPORT)
2999

    
3000
  env = None
3001
  prefix = None
3002
  suffix = None
3003
  exp_size = None
3004

    
3005
  if ieio == constants.IEIO_FILE:
3006
    (filename, ) = ieargs
3007

    
3008
    if not utils.IsNormAbsPath(filename):
3009
      _Fail("Path '%s' is not normalized or absolute", filename)
3010

    
3011
    real_filename = os.path.realpath(filename)
3012
    directory = os.path.dirname(real_filename)
3013

    
3014
    if not utils.IsBelowDir(constants.EXPORT_DIR, real_filename):
3015
      _Fail("File '%s' is not under exports directory '%s': %s",
3016
            filename, constants.EXPORT_DIR, real_filename)
3017

    
3018
    # Create directory
3019
    utils.Makedirs(directory, mode=0750)
3020

    
3021
    quoted_filename = utils.ShellQuote(filename)
3022

    
3023
    if mode == constants.IEM_IMPORT:
3024
      suffix = "> %s" % quoted_filename
3025
    elif mode == constants.IEM_EXPORT:
3026
      suffix = "< %s" % quoted_filename
3027

    
3028
      # Retrieve file size
3029
      try:
3030
        st = os.stat(filename)
3031
      except EnvironmentError, err:
3032
        logging.error("Can't stat(2) %s: %s", filename, err)
3033
      else:
3034
        exp_size = utils.BytesToMebibyte(st.st_size)
3035

    
3036
  elif ieio == constants.IEIO_RAW_DISK:
3037
    (disk, ) = ieargs
3038

    
3039
    real_disk = _OpenRealBD(disk)
3040

    
3041
    if mode == constants.IEM_IMPORT:
3042
      # we set here a smaller block size as, due to transport buffering, more
3043
      # than 64-128k will mostly ignored; we use nocreat to fail if the device
3044
      # is not already there or we pass a wrong path; we use notrunc to no
3045
      # attempt truncate on an LV device; we use oflag=dsync to not buffer too
3046
      # much memory; this means that at best, we flush every 64k, which will
3047
      # not be very fast
3048
      suffix = utils.BuildShellCmd(("| dd of=%s conv=nocreat,notrunc"
3049
                                    " bs=%s oflag=dsync"),
3050
                                    real_disk.dev_path,
3051
                                    str(64 * 1024))
3052

    
3053
    elif mode == constants.IEM_EXPORT:
3054
      # the block size on the read dd is 1MiB to match our units
3055
      prefix = utils.BuildShellCmd("dd if=%s bs=%s count=%s |",
3056
                                   real_disk.dev_path,
3057
                                   str(1024 * 1024), # 1 MB
3058
                                   str(disk.size))
3059
      exp_size = disk.size
3060

    
3061
  elif ieio == constants.IEIO_SCRIPT:
3062
    (disk, disk_index, ) = ieargs
3063

    
3064
    assert isinstance(disk_index, (int, long))
3065

    
3066
    real_disk = _OpenRealBD(disk)
3067

    
3068
    inst_os = OSFromDisk(instance.os)
3069
    env = OSEnvironment(instance, inst_os)
3070

    
3071
    if mode == constants.IEM_IMPORT:
3072
      env["IMPORT_DEVICE"] = env["DISK_%d_PATH" % disk_index]
3073
      env["IMPORT_INDEX"] = str(disk_index)
3074
      script = inst_os.import_script
3075

    
3076
    elif mode == constants.IEM_EXPORT:
3077
      env["EXPORT_DEVICE"] = real_disk.dev_path
3078
      env["EXPORT_INDEX"] = str(disk_index)
3079
      script = inst_os.export_script
3080

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

    
3084
    if mode == constants.IEM_IMPORT:
3085
      suffix = "| %s" % script_cmd
3086

    
3087
    elif mode == constants.IEM_EXPORT:
3088
      prefix = "%s |" % script_cmd
3089

    
3090
    # Let script predict size
3091
    exp_size = constants.IE_CUSTOM_SIZE
3092

    
3093
  else:
3094
    _Fail("Invalid %s I/O mode %r", mode, ieio)
3095

    
3096
  return (env, prefix, suffix, exp_size)
3097

    
3098

    
3099
def _CreateImportExportStatusDir(prefix):
3100
  """Creates status directory for import/export.
3101

3102
  """
3103
  return tempfile.mkdtemp(dir=constants.IMPORT_EXPORT_DIR,
3104
                          prefix=("%s-%s-" %
3105
                                  (prefix, utils.TimestampForFilename())))
3106

    
3107

    
3108
def StartImportExportDaemon(mode, opts, host, port, instance, component,
3109
                            ieio, ieioargs):
3110
  """Starts an import or export daemon.
3111

3112
  @param mode: Import/output mode
3113
  @type opts: L{objects.ImportExportOptions}
3114
  @param opts: Daemon options
3115
  @type host: string
3116
  @param host: Remote host for export (None for import)
3117
  @type port: int
3118
  @param port: Remote port for export (None for import)
3119
  @type instance: L{objects.Instance}
3120
  @param instance: Instance object
3121
  @type component: string
3122
  @param component: which part of the instance is transferred now,
3123
      e.g. 'disk/0'
3124
  @param ieio: Input/output type
3125
  @param ieioargs: Input/output arguments
3126

3127
  """
3128
  if mode == constants.IEM_IMPORT:
3129
    prefix = "import"
3130

    
3131
    if not (host is None and port is None):
3132
      _Fail("Can not specify host or port on import")
3133

    
3134
  elif mode == constants.IEM_EXPORT:
3135
    prefix = "export"
3136

    
3137
    if host is None or port is None:
3138
      _Fail("Host and port must be specified for an export")
3139

    
3140
  else:
3141
    _Fail("Invalid mode %r", mode)
3142

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

    
3146
  (cmd_env, cmd_prefix, cmd_suffix, exp_size) = \
3147
    _GetImportExportIoCommand(instance, mode, ieio, ieioargs)
3148

    
3149
  if opts.key_name is None:
3150
    # Use server.pem
3151
    key_path = constants.NODED_CERT_FILE
3152
    cert_path = constants.NODED_CERT_FILE
3153
    assert opts.ca_pem is None
3154
  else:
3155
    (_, key_path, cert_path) = _GetX509Filenames(constants.CRYPTO_KEYS_DIR,
3156
                                                 opts.key_name)
3157
    assert opts.ca_pem is not None
3158

    
3159
  for i in [key_path, cert_path]:
3160
    if not os.path.exists(i):
3161
      _Fail("File '%s' does not exist" % i)
3162

    
3163
  status_dir = _CreateImportExportStatusDir("%s-%s" % (prefix, component))
3164
  try:
3165
    status_file = utils.PathJoin(status_dir, _IES_STATUS_FILE)
3166
    pid_file = utils.PathJoin(status_dir, _IES_PID_FILE)
3167
    ca_file = utils.PathJoin(status_dir, _IES_CA_FILE)
3168

    
3169
    if opts.ca_pem is None:
3170
      # Use server.pem
3171
      ca = utils.ReadFile(constants.NODED_CERT_FILE)
3172
    else:
3173
      ca = opts.ca_pem
3174

    
3175
    # Write CA file
3176
    utils.WriteFile(ca_file, data=ca, mode=0400)
3177

    
3178
    cmd = [
3179
      constants.IMPORT_EXPORT_DAEMON,
3180
      status_file, mode,
3181
      "--key=%s" % key_path,
3182
      "--cert=%s" % cert_path,
3183
      "--ca=%s" % ca_file,
3184
      ]
3185

    
3186
    if host:
3187
      cmd.append("--host=%s" % host)
3188

    
3189
    if port:
3190
      cmd.append("--port=%s" % port)
3191

    
3192
    if opts.ipv6:
3193
      cmd.append("--ipv6")
3194
    else:
3195
      cmd.append("--ipv4")
3196

    
3197
    if opts.compress:
3198
      cmd.append("--compress=%s" % opts.compress)
3199

    
3200
    if opts.magic:
3201
      cmd.append("--magic=%s" % opts.magic)
3202

    
3203
    if exp_size is not None:
3204
      cmd.append("--expected-size=%s" % exp_size)
3205

    
3206
    if cmd_prefix:
3207
      cmd.append("--cmd-prefix=%s" % cmd_prefix)
3208

    
3209
    if cmd_suffix:
3210
      cmd.append("--cmd-suffix=%s" % cmd_suffix)
3211

    
3212
    if mode == constants.IEM_EXPORT:
3213
      # Retry connection a few times when connecting to remote peer
3214
      cmd.append("--connect-retries=%s" % constants.RIE_CONNECT_RETRIES)
3215
      cmd.append("--connect-timeout=%s" % constants.RIE_CONNECT_ATTEMPT_TIMEOUT)
3216
    elif opts.connect_timeout is not None:
3217
      assert mode == constants.IEM_IMPORT
3218
      # Overall timeout for establishing connection while listening
3219
      cmd.append("--connect-timeout=%s" % opts.connect_timeout)
3220

    
3221
    logfile = _InstanceLogName(prefix, instance.os, instance.name, component)
3222

    
3223
    # TODO: Once _InstanceLogName uses tempfile.mkstemp, StartDaemon has
3224
    # support for receiving a file descriptor for output
3225
    utils.StartDaemon(cmd, env=cmd_env, pidfile=pid_file,
3226
                      output=logfile)
3227

    
3228
    # The import/export name is simply the status directory name
3229
    return os.path.basename(status_dir)
3230

    
3231
  except Exception:
3232
    shutil.rmtree(status_dir, ignore_errors=True)
3233
    raise
3234

    
3235

    
3236
def GetImportExportStatus(names):
3237
  """Returns import/export daemon status.
3238

3239
  @type names: sequence
3240
  @param names: List of names
3241
  @rtype: List of dicts
3242
  @return: Returns a list of the state of each named import/export or None if a
3243
           status couldn't be read
3244

3245
  """
3246
  result = []
3247

    
3248
  for name in names:
3249
    status_file = utils.PathJoin(constants.IMPORT_EXPORT_DIR, name,
3250
                                 _IES_STATUS_FILE)
3251

    
3252
    try:
3253
      data = utils.ReadFile(status_file)
3254
    except EnvironmentError, err:
3255
      if err.errno != errno.ENOENT:
3256
        raise
3257
      data = None
3258

    
3259
    if not data:
3260
      result.append(None)
3261
      continue
3262

    
3263
    result.append(serializer.LoadJson(data))
3264

    
3265
  return result
3266

    
3267

    
3268
def AbortImportExport(name):
3269
  """Sends SIGTERM to a running import/export daemon.
3270

3271
  """
3272
  logging.info("Abort import/export %s", name)
3273

    
3274
  status_dir = utils.PathJoin(constants.IMPORT_EXPORT_DIR, name)
3275
  pid = utils.ReadLockedPidFile(utils.PathJoin(status_dir, _IES_PID_FILE))
3276

    
3277
  if pid:
3278
    logging.info("Import/export %s is running with PID %s, sending SIGTERM",
3279
                 name, pid)
3280
    utils.IgnoreProcessNotFound(os.kill, pid, signal.SIGTERM)
3281

    
3282

    
3283
def CleanupImportExport(name):
3284
  """Cleanup after an import or export.
3285

3286
  If the import/export daemon is still running it's killed. Afterwards the
3287
  whole status directory is removed.
3288

3289
  """
3290
  logging.info("Finalizing import/export %s", name)
3291

    
3292
  status_dir = utils.PathJoin(constants.IMPORT_EXPORT_DIR, name)
3293

    
3294
  pid = utils.ReadLockedPidFile(utils.PathJoin(status_dir, _IES_PID_FILE))
3295

    
3296
  if pid:
3297
    logging.info("Import/export %s is still running with PID %s",
3298
                 name, pid)
3299
    utils.KillProcess(pid, waitpid=False)
3300

    
3301
  shutil.rmtree(status_dir, ignore_errors=True)
3302

    
3303

    
3304
def _FindDisks(nodes_ip, disks):
3305
  """Sets the physical ID on disks and returns the block devices.
3306

3307
  """
3308
  # set the correct physical ID
3309
  my_name = netutils.Hostname.GetSysName()
3310
  for cf in disks:
3311
    cf.SetPhysicalID(my_name, nodes_ip)
3312

    
3313
  bdevs = []
3314

    
3315
  for cf in disks:
3316
    rd = _RecursiveFindBD(cf)
3317
    if rd is None:
3318
      _Fail("Can't find device %s", cf)
3319
    bdevs.append(rd)
3320
  return bdevs
3321

    
3322

    
3323
def DrbdDisconnectNet(nodes_ip, disks):
3324
  """Disconnects the network on a list of drbd devices.
3325

3326
  """
3327
  bdevs = _FindDisks(nodes_ip, disks)
3328

    
3329
  # disconnect disks
3330
  for rd in bdevs:
3331
    try:
3332
      rd.DisconnectNet()
3333
    except errors.BlockDeviceError, err:
3334
      _Fail("Can't change network configuration to standalone mode: %s",
3335
            err, exc=True)
3336

    
3337

    
3338
def DrbdAttachNet(nodes_ip, disks, instance_name, multimaster):
3339
  """Attaches the network on a list of drbd devices.
3340

3341
  """
3342
  bdevs = _FindDisks(nodes_ip, disks)
3343

    
3344
  if multimaster:
3345
    for idx, rd in enumerate(bdevs):
3346
      try:
3347
        _SymlinkBlockDev(instance_name, rd.dev_path, idx)
3348
      except EnvironmentError, err:
3349
        _Fail("Can't create symlink: %s", err)
3350
  # reconnect disks, switch to new master configuration and if
3351
  # needed primary mode
3352
  for rd in bdevs:
3353
    try:
3354
      rd.AttachNet(multimaster)
3355
    except errors.BlockDeviceError, err:
3356
      _Fail("Can't change network configuration: %s", err)
3357

    
3358
  # wait until the disks are connected; we need to retry the re-attach
3359
  # if the device becomes standalone, as this might happen if the one
3360
  # node disconnects and reconnects in a different mode before the
3361
  # other node reconnects; in this case, one or both of the nodes will
3362
  # decide it has wrong configuration and switch to standalone
3363

    
3364
  def _Attach():
3365
    all_connected = True
3366

    
3367
    for rd in bdevs:
3368
      stats = rd.GetProcStatus()
3369

    
3370
      all_connected = (all_connected and
3371
                       (stats.is_connected or stats.is_in_resync))
3372

    
3373
      if stats.is_standalone:
3374
        # peer had different config info and this node became
3375
        # standalone, even though this should not happen with the
3376
        # new staged way of changing disk configs
3377
        try:
3378
          rd.AttachNet(multimaster)
3379
        except errors.BlockDeviceError, err:
3380
          _Fail("Can't change network configuration: %s", err)
3381

    
3382
    if not all_connected:
3383
      raise utils.RetryAgain()
3384

    
3385
  try:
3386
    # Start with a delay of 100 miliseconds and go up to 5 seconds
3387
    utils.Retry(_Attach, (0.1, 1.5, 5.0), 2 * 60)
3388
  except utils.RetryTimeout:
3389
    _Fail("Timeout in disk reconnecting")
3390

    
3391
  if multimaster:
3392
    # change to primary mode
3393
    for rd in bdevs:
3394
      try:
3395
        rd.Open()
3396
      except errors.BlockDeviceError, err:
3397
        _Fail("Can't change to primary mode: %s", err)
3398

    
3399

    
3400
def DrbdWaitSync(nodes_ip, disks):
3401
  """Wait until DRBDs have synchronized.
3402

3403
  """
3404
  def _helper(rd):
3405
    stats = rd.GetProcStatus()
3406
    if not (stats.is_connected or stats.is_in_resync):
3407
      raise utils.RetryAgain()
3408
    return stats
3409

    
3410
  bdevs = _FindDisks(nodes_ip, disks)
3411

    
3412
  min_resync = 100
3413
  alldone = True
3414
  for rd in bdevs:
3415
    try:
3416
      # poll each second for 15 seconds
3417
      stats = utils.Retry(_helper, 1, 15, args=[rd])
3418
    except utils.RetryTimeout:
3419
      stats = rd.GetProcStatus()
3420
      # last check
3421
      if not (stats.is_connected or stats.is_in_resync):
3422
        _Fail("DRBD device %s is not in sync: stats=%s", rd, stats)
3423
    alldone = alldone and (not stats.is_in_resync)
3424
    if stats.sync_percent is not None:
3425
      min_resync = min(min_resync, stats.sync_percent)
3426

    
3427
  return (alldone, min_resync)
3428

    
3429

    
3430
def GetDrbdUsermodeHelper():
3431
  """Returns DRBD usermode helper currently configured.
3432

3433
  """
3434
  try:
3435
    return bdev.BaseDRBD.GetUsermodeHelper()
3436
  except errors.BlockDeviceError, err:
3437
    _Fail(str(err))
3438

    
3439

    
3440
def PowercycleNode(hypervisor_type):
3441
  """Hard-powercycle the node.
3442

3443
  Because we need to return first, and schedule the powercycle in the
3444
  background, we won't be able to report failures nicely.
3445

3446
  """
3447
  hyper = hypervisor.GetHypervisor(hypervisor_type)
3448
  try:
3449
    pid = os.fork()
3450
  except OSError:
3451
    # if we can't fork, we'll pretend that we're in the child process
3452
    pid = 0
3453
  if pid > 0:
3454
    return "Reboot scheduled in 5 seconds"
3455
  # ensure the child is running on ram
3456
  try:
3457
    utils.Mlockall()
3458
  except Exception: # pylint: disable=W0703
3459
    pass
3460
  time.sleep(5)
3461
  hyper.PowercycleNode()
3462

    
3463

    
3464
class HooksRunner(object):
3465
  """Hook runner.
3466

3467
  This class is instantiated on the node side (ganeti-noded) and not
3468
  on the master side.
3469

3470
  """
3471
  def __init__(self, hooks_base_dir=None):
3472
    """Constructor for hooks runner.
3473

3474
    @type hooks_base_dir: str or None
3475
    @param hooks_base_dir: if not None, this overrides the
3476
        L{constants.HOOKS_BASE_DIR} (useful for unittests)
3477

3478
    """
3479
    if hooks_base_dir is None:
3480
      hooks_base_dir = constants.HOOKS_BASE_DIR
3481
    # yeah, _BASE_DIR is not valid for attributes, we use it like a
3482
    # constant
3483
    self._BASE_DIR = hooks_base_dir # pylint: disable=C0103
3484

    
3485
  def RunLocalHooks(self, node_list, hpath, phase, env):
3486
    """Check that the hooks will be run only locally and then run them.
3487

3488
    """
3489
    assert len(node_list) == 1
3490
    node = node_list[0]
3491
    _, myself = ssconf.GetMasterAndMyself()
3492
    assert node == myself
3493

    
3494
    results = self.RunHooks(hpath, phase, env)
3495

    
3496
    # Return values in the form expected by HooksMaster
3497
    return {node: (None, False, results)}
3498

    
3499
  def RunHooks(self, hpath, phase, env):
3500
    """Run the scripts in the hooks directory.
3501

3502
    @type hpath: str
3503
    @param hpath: the path to the hooks directory which
3504
        holds the scripts
3505
    @type phase: str
3506
    @param phase: either L{constants.HOOKS_PHASE_PRE} or
3507
        L{constants.HOOKS_PHASE_POST}
3508
    @type env: dict
3509
    @param env: dictionary with the environment for the hook
3510
    @rtype: list
3511
    @return: list of 3-element tuples:
3512
      - script path
3513
      - script result, either L{constants.HKR_SUCCESS} or
3514
        L{constants.HKR_FAIL}
3515
      - output of the script
3516

3517
    @raise errors.ProgrammerError: for invalid input
3518
        parameters
3519

3520
    """
3521
    if phase == constants.HOOKS_PHASE_PRE:
3522
      suffix = "pre"
3523
    elif phase == constants.HOOKS_PHASE_POST:
3524
      suffix = "post"
3525
    else:
3526
      _Fail("Unknown hooks phase '%s'", phase)
3527

    
3528
    subdir = "%s-%s.d" % (hpath, suffix)
3529
    dir_name = utils.PathJoin(self._BASE_DIR, subdir)
3530

    
3531
    results = []
3532

    
3533
    if not os.path.isdir(dir_name):
3534
      # for non-existing/non-dirs, we simply exit instead of logging a
3535
      # warning at every operation
3536
      return results
3537

    
3538
    runparts_results = utils.RunParts(dir_name, env=env, reset_env=True)
3539

    
3540
    for (relname, relstatus, runresult)  in runparts_results:
3541
      if relstatus == constants.RUNPARTS_SKIP:
3542
        rrval = constants.HKR_SKIP
3543
        output = ""
3544
      elif relstatus == constants.RUNPARTS_ERR:
3545
        rrval = constants.HKR_FAIL
3546
        output = "Hook script execution error: %s" % runresult
3547
      elif relstatus == constants.RUNPARTS_RUN:
3548
        if runresult.failed:
3549
          rrval = constants.HKR_FAIL
3550
        else:
3551
          rrval = constants.HKR_SUCCESS
3552
        output = utils.SafeEncode(runresult.output.strip())
3553
      results.append(("%s/%s" % (subdir, relname), rrval, output))
3554

    
3555
    return results
3556

    
3557

    
3558
class IAllocatorRunner(object):
3559
  """IAllocator runner.
3560

3561
  This class is instantiated on the node side (ganeti-noded) and not on
3562
  the master side.
3563

3564
  """
3565
  @staticmethod
3566
  def Run(name, idata):
3567
    """Run an iallocator script.
3568

3569
    @type name: str
3570
    @param name: the iallocator script name
3571
    @type idata: str
3572
    @param idata: the allocator input data
3573

3574
    @rtype: tuple
3575
    @return: two element tuple of:
3576
       - status
3577
       - either error message or stdout of allocator (for success)
3578

3579
    """
3580
    alloc_script = utils.FindFile(name, constants.IALLOCATOR_SEARCH_PATH,
3581
                                  os.path.isfile)
3582
    if alloc_script is None:
3583
      _Fail("iallocator module '%s' not found in the search path", name)
3584

    
3585
    fd, fin_name = tempfile.mkstemp(prefix="ganeti-iallocator.")
3586
    try:
3587
      os.write(fd, idata)
3588
      os.close(fd)
3589
      result = utils.RunCmd([alloc_script, fin_name])
3590
      if result.failed:
3591
        _Fail("iallocator module '%s' failed: %s, output '%s'",
3592
              name, result.fail_reason, result.output)
3593
    finally:
3594
      os.unlink(fin_name)
3595

    
3596
    return result.stdout
3597

    
3598

    
3599
class DevCacheManager(object):
3600
  """Simple class for managing a cache of block device information.
3601

3602
  """
3603
  _DEV_PREFIX = "/dev/"
3604
  _ROOT_DIR = constants.BDEV_CACHE_DIR
3605

    
3606
  @classmethod
3607
  def _ConvertPath(cls, dev_path):
3608
    """Converts a /dev/name path to the cache file name.
3609

3610
    This replaces slashes with underscores and strips the /dev
3611
    prefix. It then returns the full path to the cache file.
3612

3613
    @type dev_path: str
3614
    @param dev_path: the C{/dev/} path name
3615
    @rtype: str
3616
    @return: the converted path name
3617

3618
    """
3619
    if dev_path.startswith(cls._DEV_PREFIX):
3620
      dev_path = dev_path[len(cls._DEV_PREFIX):]
3621
    dev_path = dev_path.replace("/", "_")
3622
    fpath = utils.PathJoin(cls._ROOT_DIR, "bdev_%s" % dev_path)
3623
    return fpath
3624

    
3625
  @classmethod
3626
  def UpdateCache(cls, dev_path, owner, on_primary, iv_name):
3627
    """Updates the cache information for a given device.
3628

3629
    @type dev_path: str
3630
    @param dev_path: the pathname of the device
3631
    @type owner: str
3632
    @param owner: the owner (instance name) of the device
3633
    @type on_primary: bool
3634
    @param on_primary: whether this is the primary
3635
        node nor not
3636
    @type iv_name: str
3637
    @param iv_name: the instance-visible name of the
3638
        device, as in objects.Disk.iv_name
3639

3640
    @rtype: None
3641

3642
    """
3643
    if dev_path is None:
3644
      logging.error("DevCacheManager.UpdateCache got a None dev_path")
3645
      return
3646
    fpath = cls._ConvertPath(dev_path)
3647
    if on_primary:
3648
      state = "primary"
3649
    else:
3650
      state = "secondary"
3651
    if iv_name is None:
3652
      iv_name = "not_visible"
3653
    fdata = "%s %s %s\n" % (str(owner), state, iv_name)
3654
    try:
3655
      utils.WriteFile(fpath, data=fdata)
3656
    except EnvironmentError, err:
3657
      logging.exception("Can't update bdev cache for %s: %s", dev_path, err)
3658

    
3659
  @classmethod
3660
  def RemoveCache(cls, dev_path):
3661
    """Remove data for a dev_path.
3662

3663
    This is just a wrapper over L{utils.io.RemoveFile} with a converted
3664
    path name and logging.
3665

3666
    @type dev_path: str
3667
    @param dev_path: the pathname of the device
3668

3669
    @rtype: None
3670

3671
    """
3672
    if dev_path is None:
3673
      logging.error("DevCacheManager.RemoveCache got a None dev_path")
3674
      return
3675
    fpath = cls._ConvertPath(dev_path)
3676
    try:
3677
      utils.RemoveFile(fpath)
3678
    except EnvironmentError, err:
3679
      logging.exception("Can't update bdev cache for %s: %s", dev_path, err)