Statistics
| Branch: | Tag: | Revision:

root / lib / backend.py @ 06c2fb4a

History | View | Annotate | Download (128.5 kB)

1
#
2
#
3

    
4
# Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011, 2012, 2013 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 compat
64
from ganeti import pathutils
65
from ganeti import vcluster
66
from ganeti import ht
67
from ganeti import hooksmaster
68

    
69

    
70
_BOOT_ID_PATH = "/proc/sys/kernel/random/boot_id"
71
_ALLOWED_CLEAN_DIRS = compat.UniqueFrozenset([
72
  pathutils.DATA_DIR,
73
  pathutils.JOB_QUEUE_ARCHIVE_DIR,
74
  pathutils.QUEUE_DIR,
75
  pathutils.CRYPTO_KEYS_DIR,
76
  ])
77
_MAX_SSL_CERT_VALIDITY = 7 * 24 * 60 * 60
78
_X509_KEY_FILE = "key"
79
_X509_CERT_FILE = "cert"
80
_IES_STATUS_FILE = "status"
81
_IES_PID_FILE = "pid"
82
_IES_CA_FILE = "ca"
83

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

    
87
# Actions for the master setup script
88
_MASTER_START = "start"
89
_MASTER_STOP = "stop"
90

    
91
#: Maximum file permissions for restricted command directory and executables
92
_RCMD_MAX_MODE = (stat.S_IRWXU |
93
                  stat.S_IRGRP | stat.S_IXGRP |
94
                  stat.S_IROTH | stat.S_IXOTH)
95

    
96
#: Delay before returning an error for restricted commands
97
_RCMD_INVALID_DELAY = 10
98

    
99
#: How long to wait to acquire lock for restricted commands (shorter than
100
#: L{_RCMD_INVALID_DELAY}) to reduce blockage of noded forks when many
101
#: command requests arrive
102
_RCMD_LOCK_TIMEOUT = _RCMD_INVALID_DELAY * 0.8
103

    
104

    
105
class RPCFail(Exception):
106
  """Class denoting RPC failure.
107

108
  Its argument is the error message.
109

110
  """
111

    
112

    
113
def _GetInstReasonFilename(instance_name):
114
  """Path of the file containing the reason of the instance status change.
115

116
  @type instance_name: string
117
  @param instance_name: The name of the instance
118
  @rtype: string
119
  @return: The path of the file
120

121
  """
122
  return utils.PathJoin(pathutils.INSTANCE_REASON_DIR, instance_name)
123

    
124

    
125
def _StoreInstReasonTrail(instance_name, trail):
126
  """Serialize a reason trail related to an instance change of state to file.
127

128
  The exact location of the file depends on the name of the instance and on
129
  the configuration of the Ganeti cluster defined at deploy time.
130

131
  @type instance_name: string
132
  @param instance_name: The name of the instance
133
  @rtype: None
134

135
  """
136
  json = serializer.DumpJson(trail)
137
  filename = _GetInstReasonFilename(instance_name)
138
  utils.WriteFile(filename, data=json)
139

    
140

    
141
def _Fail(msg, *args, **kwargs):
142
  """Log an error and the raise an RPCFail exception.
143

144
  This exception is then handled specially in the ganeti daemon and
145
  turned into a 'failed' return type. As such, this function is a
146
  useful shortcut for logging the error and returning it to the master
147
  daemon.
148

149
  @type msg: string
150
  @param msg: the text of the exception
151
  @raise RPCFail
152

153
  """
154
  if args:
155
    msg = msg % args
156
  if "log" not in kwargs or kwargs["log"]: # if we should log this error
157
    if "exc" in kwargs and kwargs["exc"]:
158
      logging.exception(msg)
159
    else:
160
      logging.error(msg)
161
  raise RPCFail(msg)
162

    
163

    
164
def _GetConfig():
165
  """Simple wrapper to return a SimpleStore.
166

167
  @rtype: L{ssconf.SimpleStore}
168
  @return: a SimpleStore instance
169

170
  """
171
  return ssconf.SimpleStore()
172

    
173

    
174
def _GetSshRunner(cluster_name):
175
  """Simple wrapper to return an SshRunner.
176

177
  @type cluster_name: str
178
  @param cluster_name: the cluster name, which is needed
179
      by the SshRunner constructor
180
  @rtype: L{ssh.SshRunner}
181
  @return: an SshRunner instance
182

183
  """
184
  return ssh.SshRunner(cluster_name)
185

    
186

    
187
def _Decompress(data):
188
  """Unpacks data compressed by the RPC client.
189

190
  @type data: list or tuple
191
  @param data: Data sent by RPC client
192
  @rtype: str
193
  @return: Decompressed data
194

195
  """
196
  assert isinstance(data, (list, tuple))
197
  assert len(data) == 2
198
  (encoding, content) = data
199
  if encoding == constants.RPC_ENCODING_NONE:
200
    return content
201
  elif encoding == constants.RPC_ENCODING_ZLIB_BASE64:
202
    return zlib.decompress(base64.b64decode(content))
203
  else:
204
    raise AssertionError("Unknown data encoding")
205

    
206

    
207
def _CleanDirectory(path, exclude=None):
208
  """Removes all regular files in a directory.
209

210
  @type path: str
211
  @param path: the directory to clean
212
  @type exclude: list
213
  @param exclude: list of files to be excluded, defaults
214
      to the empty list
215

216
  """
217
  if path not in _ALLOWED_CLEAN_DIRS:
218
    _Fail("Path passed to _CleanDirectory not in allowed clean targets: '%s'",
219
          path)
220

    
221
  if not os.path.isdir(path):
222
    return
223
  if exclude is None:
224
    exclude = []
225
  else:
226
    # Normalize excluded paths
227
    exclude = [os.path.normpath(i) for i in exclude]
228

    
229
  for rel_name in utils.ListVisibleFiles(path):
230
    full_name = utils.PathJoin(path, rel_name)
231
    if full_name in exclude:
232
      continue
233
    if os.path.isfile(full_name) and not os.path.islink(full_name):
234
      utils.RemoveFile(full_name)
235

    
236

    
237
def _BuildUploadFileList():
238
  """Build the list of allowed upload files.
239

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

242
  """
243
  allowed_files = set([
244
    pathutils.CLUSTER_CONF_FILE,
245
    pathutils.ETC_HOSTS,
246
    pathutils.SSH_KNOWN_HOSTS_FILE,
247
    pathutils.VNC_PASSWORD_FILE,
248
    pathutils.RAPI_CERT_FILE,
249
    pathutils.SPICE_CERT_FILE,
250
    pathutils.SPICE_CACERT_FILE,
251
    pathutils.RAPI_USERS_FILE,
252
    pathutils.CONFD_HMAC_KEY,
253
    pathutils.CLUSTER_DOMAIN_SECRET_FILE,
254
    ])
255

    
256
  for hv_name in constants.HYPER_TYPES:
257
    hv_class = hypervisor.GetHypervisorClass(hv_name)
258
    allowed_files.update(hv_class.GetAncillaryFiles()[0])
259

    
260
  assert pathutils.FILE_STORAGE_PATHS_FILE not in allowed_files, \
261
    "Allowed file storage paths should never be uploaded via RPC"
262

    
263
  return frozenset(allowed_files)
264

    
265

    
266
_ALLOWED_UPLOAD_FILES = _BuildUploadFileList()
267

    
268

    
269
def JobQueuePurge():
270
  """Removes job queue files and archived jobs.
271

272
  @rtype: tuple
273
  @return: True, None
274

275
  """
276
  _CleanDirectory(pathutils.QUEUE_DIR, exclude=[pathutils.JOB_QUEUE_LOCK_FILE])
277
  _CleanDirectory(pathutils.JOB_QUEUE_ARCHIVE_DIR)
278

    
279

    
280
def GetMasterInfo():
281
  """Returns master information.
282

283
  This is an utility function to compute master information, either
284
  for consumption here or from the node daemon.
285

286
  @rtype: tuple
287
  @return: master_netdev, master_ip, master_name, primary_ip_family,
288
    master_netmask
289
  @raise RPCFail: in case of errors
290

291
  """
292
  try:
293
    cfg = _GetConfig()
294
    master_netdev = cfg.GetMasterNetdev()
295
    master_ip = cfg.GetMasterIP()
296
    master_netmask = cfg.GetMasterNetmask()
297
    master_node = cfg.GetMasterNode()
298
    primary_ip_family = cfg.GetPrimaryIPFamily()
299
  except errors.ConfigurationError, err:
300
    _Fail("Cluster configuration incomplete: %s", err, exc=True)
301
  return (master_netdev, master_ip, master_node, primary_ip_family,
302
          master_netmask)
303

    
304

    
305
def RunLocalHooks(hook_opcode, hooks_path, env_builder_fn):
306
  """Decorator that runs hooks before and after the decorated function.
307

308
  @type hook_opcode: string
309
  @param hook_opcode: opcode of the hook
310
  @type hooks_path: string
311
  @param hooks_path: path of the hooks
312
  @type env_builder_fn: function
313
  @param env_builder_fn: function that returns a dictionary containing the
314
    environment variables for the hooks. Will get all the parameters of the
315
    decorated function.
316
  @raise RPCFail: in case of pre-hook failure
317

318
  """
319
  def decorator(fn):
320
    def wrapper(*args, **kwargs):
321
      _, myself = ssconf.GetMasterAndMyself()
322
      nodes = ([myself], [myself])  # these hooks run locally
323

    
324
      env_fn = compat.partial(env_builder_fn, *args, **kwargs)
325

    
326
      cfg = _GetConfig()
327
      hr = HooksRunner()
328
      hm = hooksmaster.HooksMaster(hook_opcode, hooks_path, nodes,
329
                                   hr.RunLocalHooks, None, env_fn,
330
                                   logging.warning, cfg.GetClusterName(),
331
                                   cfg.GetMasterNode())
332
      hm.RunPhase(constants.HOOKS_PHASE_PRE)
333
      result = fn(*args, **kwargs)
334
      hm.RunPhase(constants.HOOKS_PHASE_POST)
335

    
336
      return result
337
    return wrapper
338
  return decorator
339

    
340

    
341
def _BuildMasterIpEnv(master_params, use_external_mip_script=None):
342
  """Builds environment variables for master IP hooks.
343

344
  @type master_params: L{objects.MasterNetworkParameters}
345
  @param master_params: network parameters of the master
346
  @type use_external_mip_script: boolean
347
  @param use_external_mip_script: whether to use an external master IP
348
    address setup script (unused, but necessary per the implementation of the
349
    _RunLocalHooks decorator)
350

351
  """
352
  # pylint: disable=W0613
353
  ver = netutils.IPAddress.GetVersionFromAddressFamily(master_params.ip_family)
354
  env = {
355
    "MASTER_NETDEV": master_params.netdev,
356
    "MASTER_IP": master_params.ip,
357
    "MASTER_NETMASK": str(master_params.netmask),
358
    "CLUSTER_IP_VERSION": str(ver),
359
  }
360

    
361
  return env
362

    
363

    
364
def _RunMasterSetupScript(master_params, action, use_external_mip_script):
365
  """Execute the master IP address setup script.
366

367
  @type master_params: L{objects.MasterNetworkParameters}
368
  @param master_params: network parameters of the master
369
  @type action: string
370
  @param action: action to pass to the script. Must be one of
371
    L{backend._MASTER_START} or L{backend._MASTER_STOP}
372
  @type use_external_mip_script: boolean
373
  @param use_external_mip_script: whether to use an external master IP
374
    address setup script
375
  @raise backend.RPCFail: if there are errors during the execution of the
376
    script
377

378
  """
379
  env = _BuildMasterIpEnv(master_params)
380

    
381
  if use_external_mip_script:
382
    setup_script = pathutils.EXTERNAL_MASTER_SETUP_SCRIPT
383
  else:
384
    setup_script = pathutils.DEFAULT_MASTER_SETUP_SCRIPT
385

    
386
  result = utils.RunCmd([setup_script, action], env=env, reset_env=True)
387

    
388
  if result.failed:
389
    _Fail("Failed to %s the master IP. Script return value: %s, output: '%s'" %
390
          (action, result.exit_code, result.output), log=True)
391

    
392

    
393
@RunLocalHooks(constants.FAKE_OP_MASTER_TURNUP, "master-ip-turnup",
394
               _BuildMasterIpEnv)
395
def ActivateMasterIp(master_params, use_external_mip_script):
396
  """Activate the IP address of the master daemon.
397

398
  @type master_params: L{objects.MasterNetworkParameters}
399
  @param master_params: network parameters of the master
400
  @type use_external_mip_script: boolean
401
  @param use_external_mip_script: whether to use an external master IP
402
    address setup script
403
  @raise RPCFail: in case of errors during the IP startup
404

405
  """
406
  _RunMasterSetupScript(master_params, _MASTER_START,
407
                        use_external_mip_script)
408

    
409

    
410
def StartMasterDaemons(no_voting):
411
  """Activate local node as master node.
412

413
  The function will start the master daemons (ganeti-masterd and ganeti-rapi).
414

415
  @type no_voting: boolean
416
  @param no_voting: whether to start ganeti-masterd without a node vote
417
      but still non-interactively
418
  @rtype: None
419

420
  """
421

    
422
  if no_voting:
423
    masterd_args = "--no-voting --yes-do-it"
424
  else:
425
    masterd_args = ""
426

    
427
  env = {
428
    "EXTRA_MASTERD_ARGS": masterd_args,
429
    }
430

    
431
  result = utils.RunCmd([pathutils.DAEMON_UTIL, "start-master"], env=env)
432
  if result.failed:
433
    msg = "Can't start Ganeti master: %s" % result.output
434
    logging.error(msg)
435
    _Fail(msg)
436

    
437

    
438
@RunLocalHooks(constants.FAKE_OP_MASTER_TURNDOWN, "master-ip-turndown",
439
               _BuildMasterIpEnv)
440
def DeactivateMasterIp(master_params, use_external_mip_script):
441
  """Deactivate the master IP on this node.
442

443
  @type master_params: L{objects.MasterNetworkParameters}
444
  @param master_params: network parameters of the master
445
  @type use_external_mip_script: boolean
446
  @param use_external_mip_script: whether to use an external master IP
447
    address setup script
448
  @raise RPCFail: in case of errors during the IP turndown
449

450
  """
451
  _RunMasterSetupScript(master_params, _MASTER_STOP,
452
                        use_external_mip_script)
453

    
454

    
455
def StopMasterDaemons():
456
  """Stop the master daemons on this node.
457

458
  Stop the master daemons (ganeti-masterd and ganeti-rapi) on this node.
459

460
  @rtype: None
461

462
  """
463
  # TODO: log and report back to the caller the error failures; we
464
  # need to decide in which case we fail the RPC for this
465

    
466
  result = utils.RunCmd([pathutils.DAEMON_UTIL, "stop-master"])
467
  if result.failed:
468
    logging.error("Could not stop Ganeti master, command %s had exitcode %s"
469
                  " and error %s",
470
                  result.cmd, result.exit_code, result.output)
471

    
472

    
473
def ChangeMasterNetmask(old_netmask, netmask, master_ip, master_netdev):
474
  """Change the netmask of the master IP.
475

476
  @param old_netmask: the old value of the netmask
477
  @param netmask: the new value of the netmask
478
  @param master_ip: the master IP
479
  @param master_netdev: the master network device
480

481
  """
482
  if old_netmask == netmask:
483
    return
484

    
485
  if not netutils.IPAddress.Own(master_ip):
486
    _Fail("The master IP address is not up, not attempting to change its"
487
          " netmask")
488

    
489
  result = utils.RunCmd([constants.IP_COMMAND_PATH, "address", "add",
490
                         "%s/%s" % (master_ip, netmask),
491
                         "dev", master_netdev, "label",
492
                         "%s:0" % master_netdev])
493
  if result.failed:
494
    _Fail("Could not set the new netmask on the master IP address")
495

    
496
  result = utils.RunCmd([constants.IP_COMMAND_PATH, "address", "del",
497
                         "%s/%s" % (master_ip, old_netmask),
498
                         "dev", master_netdev, "label",
499
                         "%s:0" % master_netdev])
500
  if result.failed:
501
    _Fail("Could not bring down the master IP address with the old netmask")
502

    
503

    
504
def EtcHostsModify(mode, host, ip):
505
  """Modify a host entry in /etc/hosts.
506

507
  @param mode: The mode to operate. Either add or remove entry
508
  @param host: The host to operate on
509
  @param ip: The ip associated with the entry
510

511
  """
512
  if mode == constants.ETC_HOSTS_ADD:
513
    if not ip:
514
      RPCFail("Mode 'add' needs 'ip' parameter, but parameter not"
515
              " present")
516
    utils.AddHostToEtcHosts(host, ip)
517
  elif mode == constants.ETC_HOSTS_REMOVE:
518
    if ip:
519
      RPCFail("Mode 'remove' does not allow 'ip' parameter, but"
520
              " parameter is present")
521
    utils.RemoveHostFromEtcHosts(host)
522
  else:
523
    RPCFail("Mode not supported")
524

    
525

    
526
def LeaveCluster(modify_ssh_setup):
527
  """Cleans up and remove the current node.
528

529
  This function cleans up and prepares the current node to be removed
530
  from the cluster.
531

532
  If processing is successful, then it raises an
533
  L{errors.QuitGanetiException} which is used as a special case to
534
  shutdown the node daemon.
535

536
  @param modify_ssh_setup: boolean
537

538
  """
539
  _CleanDirectory(pathutils.DATA_DIR)
540
  _CleanDirectory(pathutils.CRYPTO_KEYS_DIR)
541
  JobQueuePurge()
542

    
543
  if modify_ssh_setup:
544
    try:
545
      priv_key, pub_key, auth_keys = ssh.GetUserFiles(constants.SSH_LOGIN_USER)
546

    
547
      utils.RemoveAuthorizedKey(auth_keys, utils.ReadFile(pub_key))
548

    
549
      utils.RemoveFile(priv_key)
550
      utils.RemoveFile(pub_key)
551
    except errors.OpExecError:
552
      logging.exception("Error while processing ssh files")
553

    
554
  try:
555
    utils.RemoveFile(pathutils.CONFD_HMAC_KEY)
556
    utils.RemoveFile(pathutils.RAPI_CERT_FILE)
557
    utils.RemoveFile(pathutils.SPICE_CERT_FILE)
558
    utils.RemoveFile(pathutils.SPICE_CACERT_FILE)
559
    utils.RemoveFile(pathutils.NODED_CERT_FILE)
560
  except: # pylint: disable=W0702
561
    logging.exception("Error while removing cluster secrets")
562

    
563
  result = utils.RunCmd([pathutils.DAEMON_UTIL, "stop", constants.CONFD])
564
  if result.failed:
565
    logging.error("Command %s failed with exitcode %s and error %s",
566
                  result.cmd, result.exit_code, result.output)
567

    
568
  # Raise a custom exception (handled in ganeti-noded)
569
  raise errors.QuitGanetiException(True, "Shutdown scheduled")
570

    
571

    
572
def _GetVgInfo(name, excl_stor):
573
  """Retrieves information about a LVM volume group.
574

575
  """
576
  # TODO: GetVGInfo supports returning information for multiple VGs at once
577
  vginfo = bdev.LogicalVolume.GetVGInfo([name], excl_stor)
578
  if vginfo:
579
    vg_free = int(round(vginfo[0][0], 0))
580
    vg_size = int(round(vginfo[0][1], 0))
581
  else:
582
    vg_free = None
583
    vg_size = None
584

    
585
  return {
586
    "name": name,
587
    "vg_free": vg_free,
588
    "vg_size": vg_size,
589
    }
590

    
591

    
592
def _GetHvInfo(name):
593
  """Retrieves node information from a hypervisor.
594

595
  The information returned depends on the hypervisor. Common items:
596

597
    - vg_size is the size of the configured volume group in MiB
598
    - vg_free is the free size of the volume group in MiB
599
    - memory_dom0 is the memory allocated for domain0 in MiB
600
    - memory_free is the currently available (free) ram in MiB
601
    - memory_total is the total number of ram in MiB
602
    - hv_version: the hypervisor version, if available
603

604
  """
605
  return hypervisor.GetHypervisor(name).GetNodeInfo()
606

    
607

    
608
def _GetNamedNodeInfo(names, fn):
609
  """Calls C{fn} for all names in C{names} and returns a dictionary.
610

611
  @rtype: None or dict
612

613
  """
614
  if names is None:
615
    return None
616
  else:
617
    return map(fn, names)
618

    
619

    
620
def GetNodeInfo(vg_names, hv_names, excl_stor):
621
  """Gives back a hash with different information about the node.
622

623
  @type vg_names: list of string
624
  @param vg_names: Names of the volume groups to ask for disk space information
625
  @type hv_names: list of string
626
  @param hv_names: Names of the hypervisors to ask for node information
627
  @type excl_stor: boolean
628
  @param excl_stor: Whether exclusive_storage is active
629
  @rtype: tuple; (string, None/dict, None/dict)
630
  @return: Tuple containing boot ID, volume group information and hypervisor
631
    information
632

633
  """
634
  bootid = utils.ReadFile(_BOOT_ID_PATH, size=128).rstrip("\n")
635
  vg_info = _GetNamedNodeInfo(vg_names, (lambda vg: _GetVgInfo(vg, excl_stor)))
636
  hv_info = _GetNamedNodeInfo(hv_names, _GetHvInfo)
637

    
638
  return (bootid, vg_info, hv_info)
639

    
640

    
641
def _CheckExclusivePvs(pvi_list):
642
  """Check that PVs are not shared among LVs
643

644
  @type pvi_list: list of L{objects.LvmPvInfo} objects
645
  @param pvi_list: information about the PVs
646

647
  @rtype: list of tuples (string, list of strings)
648
  @return: offending volumes, as tuples: (pv_name, [lv1_name, lv2_name...])
649

650
  """
651
  res = []
652
  for pvi in pvi_list:
653
    if len(pvi.lv_list) > 1:
654
      res.append((pvi.name, pvi.lv_list))
655
  return res
656

    
657

    
658
def VerifyNode(what, cluster_name):
659
  """Verify the status of the local node.
660

661
  Based on the input L{what} parameter, various checks are done on the
662
  local node.
663

664
  If the I{filelist} key is present, this list of
665
  files is checksummed and the file/checksum pairs are returned.
666

667
  If the I{nodelist} key is present, we check that we have
668
  connectivity via ssh with the target nodes (and check the hostname
669
  report).
670

671
  If the I{node-net-test} key is present, we check that we have
672
  connectivity to the given nodes via both primary IP and, if
673
  applicable, secondary IPs.
674

675
  @type what: C{dict}
676
  @param what: a dictionary of things to check:
677
      - filelist: list of files for which to compute checksums
678
      - nodelist: list of nodes we should check ssh communication with
679
      - node-net-test: list of nodes we should check node daemon port
680
        connectivity with
681
      - hypervisor: list with hypervisors to run the verify for
682
  @rtype: dict
683
  @return: a dictionary with the same keys as the input dict, and
684
      values representing the result of the checks
685

686
  """
687
  result = {}
688
  my_name = netutils.Hostname.GetSysName()
689
  port = netutils.GetDaemonPort(constants.NODED)
690
  vm_capable = my_name not in what.get(constants.NV_VMNODES, [])
691

    
692
  if constants.NV_HYPERVISOR in what and vm_capable:
693
    result[constants.NV_HYPERVISOR] = tmp = {}
694
    for hv_name in what[constants.NV_HYPERVISOR]:
695
      try:
696
        val = hypervisor.GetHypervisor(hv_name).Verify()
697
      except errors.HypervisorError, err:
698
        val = "Error while checking hypervisor: %s" % str(err)
699
      tmp[hv_name] = val
700

    
701
  if constants.NV_HVPARAMS in what and vm_capable:
702
    result[constants.NV_HVPARAMS] = tmp = []
703
    for source, hv_name, hvparms in what[constants.NV_HVPARAMS]:
704
      try:
705
        logging.info("Validating hv %s, %s", hv_name, hvparms)
706
        hypervisor.GetHypervisor(hv_name).ValidateParameters(hvparms)
707
      except errors.HypervisorError, err:
708
        tmp.append((source, hv_name, str(err)))
709

    
710
  if constants.NV_FILELIST in what:
711
    fingerprints = utils.FingerprintFiles(map(vcluster.LocalizeVirtualPath,
712
                                              what[constants.NV_FILELIST]))
713
    result[constants.NV_FILELIST] = \
714
      dict((vcluster.MakeVirtualPath(key), value)
715
           for (key, value) in fingerprints.items())
716

    
717
  if constants.NV_NODELIST in what:
718
    (nodes, bynode) = what[constants.NV_NODELIST]
719

    
720
    # Add nodes from other groups (different for each node)
721
    try:
722
      nodes.extend(bynode[my_name])
723
    except KeyError:
724
      pass
725

    
726
    # Use a random order
727
    random.shuffle(nodes)
728

    
729
    # Try to contact all nodes
730
    val = {}
731
    for node in nodes:
732
      success, message = _GetSshRunner(cluster_name).VerifyNodeHostname(node)
733
      if not success:
734
        val[node] = message
735

    
736
    result[constants.NV_NODELIST] = val
737

    
738
  if constants.NV_NODENETTEST in what:
739
    result[constants.NV_NODENETTEST] = tmp = {}
740
    my_pip = my_sip = None
741
    for name, pip, sip in what[constants.NV_NODENETTEST]:
742
      if name == my_name:
743
        my_pip = pip
744
        my_sip = sip
745
        break
746
    if not my_pip:
747
      tmp[my_name] = ("Can't find my own primary/secondary IP"
748
                      " in the node list")
749
    else:
750
      for name, pip, sip in what[constants.NV_NODENETTEST]:
751
        fail = []
752
        if not netutils.TcpPing(pip, port, source=my_pip):
753
          fail.append("primary")
754
        if sip != pip:
755
          if not netutils.TcpPing(sip, port, source=my_sip):
756
            fail.append("secondary")
757
        if fail:
758
          tmp[name] = ("failure using the %s interface(s)" %
759
                       " and ".join(fail))
760

    
761
  if constants.NV_MASTERIP in what:
762
    # FIXME: add checks on incoming data structures (here and in the
763
    # rest of the function)
764
    master_name, master_ip = what[constants.NV_MASTERIP]
765
    if master_name == my_name:
766
      source = constants.IP4_ADDRESS_LOCALHOST
767
    else:
768
      source = None
769
    result[constants.NV_MASTERIP] = netutils.TcpPing(master_ip, port,
770
                                                     source=source)
771

    
772
  if constants.NV_USERSCRIPTS in what:
773
    result[constants.NV_USERSCRIPTS] = \
774
      [script for script in what[constants.NV_USERSCRIPTS]
775
       if not utils.IsExecutable(script)]
776

    
777
  if constants.NV_OOB_PATHS in what:
778
    result[constants.NV_OOB_PATHS] = tmp = []
779
    for path in what[constants.NV_OOB_PATHS]:
780
      try:
781
        st = os.stat(path)
782
      except OSError, err:
783
        tmp.append("error stating out of band helper: %s" % err)
784
      else:
785
        if stat.S_ISREG(st.st_mode):
786
          if stat.S_IMODE(st.st_mode) & stat.S_IXUSR:
787
            tmp.append(None)
788
          else:
789
            tmp.append("out of band helper %s is not executable" % path)
790
        else:
791
          tmp.append("out of band helper %s is not a file" % path)
792

    
793
  if constants.NV_LVLIST in what and vm_capable:
794
    try:
795
      val = GetVolumeList(utils.ListVolumeGroups().keys())
796
    except RPCFail, err:
797
      val = str(err)
798
    result[constants.NV_LVLIST] = val
799

    
800
  if constants.NV_INSTANCELIST in what and vm_capable:
801
    # GetInstanceList can fail
802
    try:
803
      val = GetInstanceList(what[constants.NV_INSTANCELIST])
804
    except RPCFail, err:
805
      val = str(err)
806
    result[constants.NV_INSTANCELIST] = val
807

    
808
  if constants.NV_VGLIST in what and vm_capable:
809
    result[constants.NV_VGLIST] = utils.ListVolumeGroups()
810

    
811
  if constants.NV_PVLIST in what and vm_capable:
812
    check_exclusive_pvs = constants.NV_EXCLUSIVEPVS in what
813
    val = bdev.LogicalVolume.GetPVInfo(what[constants.NV_PVLIST],
814
                                       filter_allocatable=False,
815
                                       include_lvs=check_exclusive_pvs)
816
    if check_exclusive_pvs:
817
      result[constants.NV_EXCLUSIVEPVS] = _CheckExclusivePvs(val)
818
      for pvi in val:
819
        # Avoid sending useless data on the wire
820
        pvi.lv_list = []
821
    result[constants.NV_PVLIST] = map(objects.LvmPvInfo.ToDict, val)
822

    
823
  if constants.NV_VERSION in what:
824
    result[constants.NV_VERSION] = (constants.PROTOCOL_VERSION,
825
                                    constants.RELEASE_VERSION)
826

    
827
  if constants.NV_HVINFO in what and vm_capable:
828
    hyper = hypervisor.GetHypervisor(what[constants.NV_HVINFO])
829
    result[constants.NV_HVINFO] = hyper.GetNodeInfo()
830

    
831
  if constants.NV_DRBDLIST in what and vm_capable:
832
    try:
833
      used_minors = bdev.DRBD8.GetUsedDevs().keys()
834
    except errors.BlockDeviceError, err:
835
      logging.warning("Can't get used minors list", exc_info=True)
836
      used_minors = str(err)
837
    result[constants.NV_DRBDLIST] = used_minors
838

    
839
  if constants.NV_DRBDHELPER in what and vm_capable:
840
    status = True
841
    try:
842
      payload = bdev.BaseDRBD.GetUsermodeHelper()
843
    except errors.BlockDeviceError, err:
844
      logging.error("Can't get DRBD usermode helper: %s", str(err))
845
      status = False
846
      payload = str(err)
847
    result[constants.NV_DRBDHELPER] = (status, payload)
848

    
849
  if constants.NV_NODESETUP in what:
850
    result[constants.NV_NODESETUP] = tmpr = []
851
    if not os.path.isdir("/sys/block") or not os.path.isdir("/sys/class/net"):
852
      tmpr.append("The sysfs filesytem doesn't seem to be mounted"
853
                  " under /sys, missing required directories /sys/block"
854
                  " and /sys/class/net")
855
    if (not os.path.isdir("/proc/sys") or
856
        not os.path.isfile("/proc/sysrq-trigger")):
857
      tmpr.append("The procfs filesystem doesn't seem to be mounted"
858
                  " under /proc, missing required directory /proc/sys and"
859
                  " the file /proc/sysrq-trigger")
860

    
861
  if constants.NV_TIME in what:
862
    result[constants.NV_TIME] = utils.SplitTime(time.time())
863

    
864
  if constants.NV_OSLIST in what and vm_capable:
865
    result[constants.NV_OSLIST] = DiagnoseOS()
866

    
867
  if constants.NV_BRIDGES in what and vm_capable:
868
    result[constants.NV_BRIDGES] = [bridge
869
                                    for bridge in what[constants.NV_BRIDGES]
870
                                    if not utils.BridgeExists(bridge)]
871

    
872
  if what.get(constants.NV_FILE_STORAGE_PATHS) == my_name:
873
    result[constants.NV_FILE_STORAGE_PATHS] = \
874
      bdev.ComputeWrongFileStoragePaths()
875

    
876
  return result
877

    
878

    
879
def GetBlockDevSizes(devices):
880
  """Return the size of the given block devices
881

882
  @type devices: list
883
  @param devices: list of block device nodes to query
884
  @rtype: dict
885
  @return:
886
    dictionary of all block devices under /dev (key). The value is their
887
    size in MiB.
888

889
    {'/dev/disk/by-uuid/123456-12321231-312312-312': 124}
890

891
  """
892
  DEV_PREFIX = "/dev/"
893
  blockdevs = {}
894

    
895
  for devpath in devices:
896
    if not utils.IsBelowDir(DEV_PREFIX, devpath):
897
      continue
898

    
899
    try:
900
      st = os.stat(devpath)
901
    except EnvironmentError, err:
902
      logging.warning("Error stat()'ing device %s: %s", devpath, str(err))
903
      continue
904

    
905
    if stat.S_ISBLK(st.st_mode):
906
      result = utils.RunCmd(["blockdev", "--getsize64", devpath])
907
      if result.failed:
908
        # We don't want to fail, just do not list this device as available
909
        logging.warning("Cannot get size for block device %s", devpath)
910
        continue
911

    
912
      size = int(result.stdout) / (1024 * 1024)
913
      blockdevs[devpath] = size
914
  return blockdevs
915

    
916

    
917
def GetVolumeList(vg_names):
918
  """Compute list of logical volumes and their size.
919

920
  @type vg_names: list
921
  @param vg_names: the volume groups whose LVs we should list, or
922
      empty for all volume groups
923
  @rtype: dict
924
  @return:
925
      dictionary of all partions (key) with value being a tuple of
926
      their size (in MiB), inactive and online status::
927

928
        {'xenvg/test1': ('20.06', True, True)}
929

930
      in case of errors, a string is returned with the error
931
      details.
932

933
  """
934
  lvs = {}
935
  sep = "|"
936
  if not vg_names:
937
    vg_names = []
938
  result = utils.RunCmd(["lvs", "--noheadings", "--units=m", "--nosuffix",
939
                         "--separator=%s" % sep,
940
                         "-ovg_name,lv_name,lv_size,lv_attr"] + vg_names)
941
  if result.failed:
942
    _Fail("Failed to list logical volumes, lvs output: %s", result.output)
943

    
944
  for line in result.stdout.splitlines():
945
    line = line.strip()
946
    match = _LVSLINE_REGEX.match(line)
947
    if not match:
948
      logging.error("Invalid line returned from lvs output: '%s'", line)
949
      continue
950
    vg_name, name, size, attr = match.groups()
951
    inactive = attr[4] == "-"
952
    online = attr[5] == "o"
953
    virtual = attr[0] == "v"
954
    if virtual:
955
      # we don't want to report such volumes as existing, since they
956
      # don't really hold data
957
      continue
958
    lvs[vg_name + "/" + name] = (size, inactive, online)
959

    
960
  return lvs
961

    
962

    
963
def ListVolumeGroups():
964
  """List the volume groups and their size.
965

966
  @rtype: dict
967
  @return: dictionary with keys volume name and values the
968
      size of the volume
969

970
  """
971
  return utils.ListVolumeGroups()
972

    
973

    
974
def NodeVolumes():
975
  """List all volumes on this node.
976

977
  @rtype: list
978
  @return:
979
    A list of dictionaries, each having four keys:
980
      - name: the logical volume name,
981
      - size: the size of the logical volume
982
      - dev: the physical device on which the LV lives
983
      - vg: the volume group to which it belongs
984

985
    In case of errors, we return an empty list and log the
986
    error.
987

988
    Note that since a logical volume can live on multiple physical
989
    volumes, the resulting list might include a logical volume
990
    multiple times.
991

992
  """
993
  result = utils.RunCmd(["lvs", "--noheadings", "--units=m", "--nosuffix",
994
                         "--separator=|",
995
                         "--options=lv_name,lv_size,devices,vg_name"])
996
  if result.failed:
997
    _Fail("Failed to list logical volumes, lvs output: %s",
998
          result.output)
999

    
1000
  def parse_dev(dev):
1001
    return dev.split("(")[0]
1002

    
1003
  def handle_dev(dev):
1004
    return [parse_dev(x) for x in dev.split(",")]
1005

    
1006
  def map_line(line):
1007
    line = [v.strip() for v in line]
1008
    return [{"name": line[0], "size": line[1],
1009
             "dev": dev, "vg": line[3]} for dev in handle_dev(line[2])]
1010

    
1011
  all_devs = []
1012
  for line in result.stdout.splitlines():
1013
    if line.count("|") >= 3:
1014
      all_devs.extend(map_line(line.split("|")))
1015
    else:
1016
      logging.warning("Strange line in the output from lvs: '%s'", line)
1017
  return all_devs
1018

    
1019

    
1020
def BridgesExist(bridges_list):
1021
  """Check if a list of bridges exist on the current node.
1022

1023
  @rtype: boolean
1024
  @return: C{True} if all of them exist, C{False} otherwise
1025

1026
  """
1027
  missing = []
1028
  for bridge in bridges_list:
1029
    if not utils.BridgeExists(bridge):
1030
      missing.append(bridge)
1031

    
1032
  if missing:
1033
    _Fail("Missing bridges %s", utils.CommaJoin(missing))
1034

    
1035

    
1036
def GetInstanceList(hypervisor_list):
1037
  """Provides a list of instances.
1038

1039
  @type hypervisor_list: list
1040
  @param hypervisor_list: the list of hypervisors to query information
1041

1042
  @rtype: list
1043
  @return: a list of all running instances on the current node
1044
    - instance1.example.com
1045
    - instance2.example.com
1046

1047
  """
1048
  results = []
1049
  for hname in hypervisor_list:
1050
    try:
1051
      names = hypervisor.GetHypervisor(hname).ListInstances()
1052
      results.extend(names)
1053
    except errors.HypervisorError, err:
1054
      _Fail("Error enumerating instances (hypervisor %s): %s",
1055
            hname, err, exc=True)
1056

    
1057
  return results
1058

    
1059

    
1060
def GetInstanceInfo(instance, hname):
1061
  """Gives back the information about an instance as a dictionary.
1062

1063
  @type instance: string
1064
  @param instance: the instance name
1065
  @type hname: string
1066
  @param hname: the hypervisor type of the instance
1067

1068
  @rtype: dict
1069
  @return: dictionary with the following keys:
1070
      - memory: memory size of instance (int)
1071
      - state: xen state of instance (string)
1072
      - time: cpu time of instance (float)
1073
      - vcpus: the number of vcpus (int)
1074

1075
  """
1076
  output = {}
1077

    
1078
  iinfo = hypervisor.GetHypervisor(hname).GetInstanceInfo(instance)
1079
  if iinfo is not None:
1080
    output["memory"] = iinfo[2]
1081
    output["vcpus"] = iinfo[3]
1082
    output["state"] = iinfo[4]
1083
    output["time"] = iinfo[5]
1084

    
1085
  return output
1086

    
1087

    
1088
def GetInstanceMigratable(instance):
1089
  """Gives whether an instance can be migrated.
1090

1091
  @type instance: L{objects.Instance}
1092
  @param instance: object representing the instance to be checked.
1093

1094
  @rtype: tuple
1095
  @return: tuple of (result, description) where:
1096
      - result: whether the instance can be migrated or not
1097
      - description: a description of the issue, if relevant
1098

1099
  """
1100
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1101
  iname = instance.name
1102
  if iname not in hyper.ListInstances():
1103
    _Fail("Instance %s is not running", iname)
1104

    
1105
  for idx in range(len(instance.disks)):
1106
    link_name = _GetBlockDevSymlinkPath(iname, idx)
1107
    if not os.path.islink(link_name):
1108
      logging.warning("Instance %s is missing symlink %s for disk %d",
1109
                      iname, link_name, idx)
1110

    
1111

    
1112
def GetAllInstancesInfo(hypervisor_list):
1113
  """Gather data about all instances.
1114

1115
  This is the equivalent of L{GetInstanceInfo}, except that it
1116
  computes data for all instances at once, thus being faster if one
1117
  needs data about more than one instance.
1118

1119
  @type hypervisor_list: list
1120
  @param hypervisor_list: list of hypervisors to query for instance data
1121

1122
  @rtype: dict
1123
  @return: dictionary of instance: data, with data having the following keys:
1124
      - memory: memory size of instance (int)
1125
      - state: xen state of instance (string)
1126
      - time: cpu time of instance (float)
1127
      - vcpus: the number of vcpus
1128

1129
  """
1130
  output = {}
1131

    
1132
  for hname in hypervisor_list:
1133
    iinfo = hypervisor.GetHypervisor(hname).GetAllInstancesInfo()
1134
    if iinfo:
1135
      for name, _, memory, vcpus, state, times in iinfo:
1136
        value = {
1137
          "memory": memory,
1138
          "vcpus": vcpus,
1139
          "state": state,
1140
          "time": times,
1141
          }
1142
        if name in output:
1143
          # we only check static parameters, like memory and vcpus,
1144
          # and not state and time which can change between the
1145
          # invocations of the different hypervisors
1146
          for key in "memory", "vcpus":
1147
            if value[key] != output[name][key]:
1148
              _Fail("Instance %s is running twice"
1149
                    " with different parameters", name)
1150
        output[name] = value
1151

    
1152
  return output
1153

    
1154

    
1155
def _InstanceLogName(kind, os_name, instance, component):
1156
  """Compute the OS log filename for a given instance and operation.
1157

1158
  The instance name and os name are passed in as strings since not all
1159
  operations have these as part of an instance object.
1160

1161
  @type kind: string
1162
  @param kind: the operation type (e.g. add, import, etc.)
1163
  @type os_name: string
1164
  @param os_name: the os name
1165
  @type instance: string
1166
  @param instance: the name of the instance being imported/added/etc.
1167
  @type component: string or None
1168
  @param component: the name of the component of the instance being
1169
      transferred
1170

1171
  """
1172
  # TODO: Use tempfile.mkstemp to create unique filename
1173
  if component:
1174
    assert "/" not in component
1175
    c_msg = "-%s" % component
1176
  else:
1177
    c_msg = ""
1178
  base = ("%s-%s-%s%s-%s.log" %
1179
          (kind, os_name, instance, c_msg, utils.TimestampForFilename()))
1180
  return utils.PathJoin(pathutils.LOG_OS_DIR, base)
1181

    
1182

    
1183
def InstanceOsAdd(instance, reinstall, debug):
1184
  """Add an OS to an instance.
1185

1186
  @type instance: L{objects.Instance}
1187
  @param instance: Instance whose OS is to be installed
1188
  @type reinstall: boolean
1189
  @param reinstall: whether this is an instance reinstall
1190
  @type debug: integer
1191
  @param debug: debug level, passed to the OS scripts
1192
  @rtype: None
1193

1194
  """
1195
  inst_os = OSFromDisk(instance.os)
1196

    
1197
  create_env = OSEnvironment(instance, inst_os, debug)
1198
  if reinstall:
1199
    create_env["INSTANCE_REINSTALL"] = "1"
1200

    
1201
  logfile = _InstanceLogName("add", instance.os, instance.name, None)
1202

    
1203
  result = utils.RunCmd([inst_os.create_script], env=create_env,
1204
                        cwd=inst_os.path, output=logfile, reset_env=True)
1205
  if result.failed:
1206
    logging.error("os create command '%s' returned error: %s, logfile: %s,"
1207
                  " output: %s", result.cmd, result.fail_reason, logfile,
1208
                  result.output)
1209
    lines = [utils.SafeEncode(val)
1210
             for val in utils.TailFile(logfile, lines=20)]
1211
    _Fail("OS create script failed (%s), last lines in the"
1212
          " log file:\n%s", result.fail_reason, "\n".join(lines), log=False)
1213

    
1214

    
1215
def RunRenameInstance(instance, old_name, debug):
1216
  """Run the OS rename script for an instance.
1217

1218
  @type instance: L{objects.Instance}
1219
  @param instance: Instance whose OS is to be installed
1220
  @type old_name: string
1221
  @param old_name: previous instance name
1222
  @type debug: integer
1223
  @param debug: debug level, passed to the OS scripts
1224
  @rtype: boolean
1225
  @return: the success of the operation
1226

1227
  """
1228
  inst_os = OSFromDisk(instance.os)
1229

    
1230
  rename_env = OSEnvironment(instance, inst_os, debug)
1231
  rename_env["OLD_INSTANCE_NAME"] = old_name
1232

    
1233
  logfile = _InstanceLogName("rename", instance.os,
1234
                             "%s-%s" % (old_name, instance.name), None)
1235

    
1236
  result = utils.RunCmd([inst_os.rename_script], env=rename_env,
1237
                        cwd=inst_os.path, output=logfile, reset_env=True)
1238

    
1239
  if result.failed:
1240
    logging.error("os create command '%s' returned error: %s output: %s",
1241
                  result.cmd, result.fail_reason, result.output)
1242
    lines = [utils.SafeEncode(val)
1243
             for val in utils.TailFile(logfile, lines=20)]
1244
    _Fail("OS rename script failed (%s), last lines in the"
1245
          " log file:\n%s", result.fail_reason, "\n".join(lines), log=False)
1246

    
1247

    
1248
def _GetBlockDevSymlinkPath(instance_name, idx, _dir=None):
1249
  """Returns symlink path for block device.
1250

1251
  """
1252
  if _dir is None:
1253
    _dir = pathutils.DISK_LINKS_DIR
1254

    
1255
  return utils.PathJoin(_dir,
1256
                        ("%s%s%s" %
1257
                         (instance_name, constants.DISK_SEPARATOR, idx)))
1258

    
1259

    
1260
def _SymlinkBlockDev(instance_name, device_path, idx):
1261
  """Set up symlinks to a instance's block device.
1262

1263
  This is an auxiliary function run when an instance is start (on the primary
1264
  node) or when an instance is migrated (on the target node).
1265

1266

1267
  @param instance_name: the name of the target instance
1268
  @param device_path: path of the physical block device, on the node
1269
  @param idx: the disk index
1270
  @return: absolute path to the disk's symlink
1271

1272
  """
1273
  link_name = _GetBlockDevSymlinkPath(instance_name, idx)
1274
  try:
1275
    os.symlink(device_path, link_name)
1276
  except OSError, err:
1277
    if err.errno == errno.EEXIST:
1278
      if (not os.path.islink(link_name) or
1279
          os.readlink(link_name) != device_path):
1280
        os.remove(link_name)
1281
        os.symlink(device_path, link_name)
1282
    else:
1283
      raise
1284

    
1285
  return link_name
1286

    
1287

    
1288
def _RemoveBlockDevLinks(instance_name, disks):
1289
  """Remove the block device symlinks belonging to the given instance.
1290

1291
  """
1292
  for idx, _ in enumerate(disks):
1293
    link_name = _GetBlockDevSymlinkPath(instance_name, idx)
1294
    if os.path.islink(link_name):
1295
      try:
1296
        os.remove(link_name)
1297
      except OSError:
1298
        logging.exception("Can't remove symlink '%s'", link_name)
1299

    
1300

    
1301
def _GatherAndLinkBlockDevs(instance):
1302
  """Set up an instance's block device(s).
1303

1304
  This is run on the primary node at instance startup. The block
1305
  devices must be already assembled.
1306

1307
  @type instance: L{objects.Instance}
1308
  @param instance: the instance whose disks we shoul assemble
1309
  @rtype: list
1310
  @return: list of (disk_object, device_path)
1311

1312
  """
1313
  block_devices = []
1314
  for idx, disk in enumerate(instance.disks):
1315
    device = _RecursiveFindBD(disk)
1316
    if device is None:
1317
      raise errors.BlockDeviceError("Block device '%s' is not set up." %
1318
                                    str(disk))
1319
    device.Open()
1320
    try:
1321
      link_name = _SymlinkBlockDev(instance.name, device.dev_path, idx)
1322
    except OSError, e:
1323
      raise errors.BlockDeviceError("Cannot create block device symlink: %s" %
1324
                                    e.strerror)
1325

    
1326
    block_devices.append((disk, link_name))
1327

    
1328
  return block_devices
1329

    
1330

    
1331
def StartInstance(instance, startup_paused, reason, store_reason=True):
1332
  """Start an instance.
1333

1334
  @type instance: L{objects.Instance}
1335
  @param instance: the instance object
1336
  @type startup_paused: bool
1337
  @param instance: pause instance at startup?
1338
  @type reason: list of reasons
1339
  @param reason: the reason trail for this startup
1340
  @type store_reason: boolean
1341
  @param store_reason: whether to store the shutdown reason trail on file
1342
  @rtype: None
1343

1344
  """
1345
  running_instances = GetInstanceList([instance.hypervisor])
1346

    
1347
  if instance.name in running_instances:
1348
    logging.info("Instance %s already running, not starting", instance.name)
1349
    return
1350

    
1351
  try:
1352
    block_devices = _GatherAndLinkBlockDevs(instance)
1353
    hyper = hypervisor.GetHypervisor(instance.hypervisor)
1354
    hyper.StartInstance(instance, block_devices, startup_paused)
1355
    if store_reason:
1356
      _StoreInstReasonTrail(instance.name, reason)
1357
  except errors.BlockDeviceError, err:
1358
    _Fail("Block device error: %s", err, exc=True)
1359
  except errors.HypervisorError, err:
1360
    _RemoveBlockDevLinks(instance.name, instance.disks)
1361
    _Fail("Hypervisor error: %s", err, exc=True)
1362

    
1363

    
1364
def InstanceShutdown(instance, timeout, reason, store_reason=True):
1365
  """Shut an instance down.
1366

1367
  @note: this functions uses polling with a hardcoded timeout.
1368

1369
  @type instance: L{objects.Instance}
1370
  @param instance: the instance object
1371
  @type timeout: integer
1372
  @param timeout: maximum timeout for soft shutdown
1373
  @type reason: list of reasons
1374
  @param reason: the reason trail for this shutdown
1375
  @type store_reason: boolean
1376
  @param store_reason: whether to store the shutdown reason trail on file
1377
  @rtype: None
1378

1379
  """
1380
  hv_name = instance.hypervisor
1381
  hyper = hypervisor.GetHypervisor(hv_name)
1382
  iname = instance.name
1383

    
1384
  if instance.name not in hyper.ListInstances():
1385
    logging.info("Instance %s not running, doing nothing", iname)
1386
    return
1387

    
1388
  class _TryShutdown:
1389
    def __init__(self):
1390
      self.tried_once = False
1391

    
1392
    def __call__(self):
1393
      if iname not in hyper.ListInstances():
1394
        return
1395

    
1396
      try:
1397
        hyper.StopInstance(instance, retry=self.tried_once)
1398
        if store_reason:
1399
          _StoreInstReasonTrail(instance.name, reason)
1400
      except errors.HypervisorError, err:
1401
        if iname not in hyper.ListInstances():
1402
          # if the instance is no longer existing, consider this a
1403
          # success and go to cleanup
1404
          return
1405

    
1406
        _Fail("Failed to stop instance %s: %s", iname, err)
1407

    
1408
      self.tried_once = True
1409

    
1410
      raise utils.RetryAgain()
1411

    
1412
  try:
1413
    utils.Retry(_TryShutdown(), 5, timeout)
1414
  except utils.RetryTimeout:
1415
    # the shutdown did not succeed
1416
    logging.error("Shutdown of '%s' unsuccessful, forcing", iname)
1417

    
1418
    try:
1419
      hyper.StopInstance(instance, force=True)
1420
    except errors.HypervisorError, err:
1421
      if iname in hyper.ListInstances():
1422
        # only raise an error if the instance still exists, otherwise
1423
        # the error could simply be "instance ... unknown"!
1424
        _Fail("Failed to force stop instance %s: %s", iname, err)
1425

    
1426
    time.sleep(1)
1427

    
1428
    if iname in hyper.ListInstances():
1429
      _Fail("Could not shutdown instance %s even by destroy", iname)
1430

    
1431
  try:
1432
    hyper.CleanupInstance(instance.name)
1433
  except errors.HypervisorError, err:
1434
    logging.warning("Failed to execute post-shutdown cleanup step: %s", err)
1435

    
1436
  _RemoveBlockDevLinks(iname, instance.disks)
1437

    
1438

    
1439
def InstanceReboot(instance, reboot_type, shutdown_timeout, reason):
1440
  """Reboot an instance.
1441

1442
  @type instance: L{objects.Instance}
1443
  @param instance: the instance object to reboot
1444
  @type reboot_type: str
1445
  @param reboot_type: the type of reboot, one the following
1446
    constants:
1447
      - L{constants.INSTANCE_REBOOT_SOFT}: only reboot the
1448
        instance OS, do not recreate the VM
1449
      - L{constants.INSTANCE_REBOOT_HARD}: tear down and
1450
        restart the VM (at the hypervisor level)
1451
      - the other reboot type (L{constants.INSTANCE_REBOOT_FULL}) is
1452
        not accepted here, since that mode is handled differently, in
1453
        cmdlib, and translates into full stop and start of the
1454
        instance (instead of a call_instance_reboot RPC)
1455
  @type shutdown_timeout: integer
1456
  @param shutdown_timeout: maximum timeout for soft shutdown
1457
  @type reason: list of reasons
1458
  @param reason: the reason trail for this reboot
1459
  @rtype: None
1460

1461
  """
1462
  running_instances = GetInstanceList([instance.hypervisor])
1463

    
1464
  if instance.name not in running_instances:
1465
    _Fail("Cannot reboot instance %s that is not running", instance.name)
1466

    
1467
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1468
  if reboot_type == constants.INSTANCE_REBOOT_SOFT:
1469
    try:
1470
      hyper.RebootInstance(instance)
1471
    except errors.HypervisorError, err:
1472
      _Fail("Failed to soft reboot instance %s: %s", instance.name, err)
1473
  elif reboot_type == constants.INSTANCE_REBOOT_HARD:
1474
    try:
1475
      InstanceShutdown(instance, shutdown_timeout, reason, store_reason=False)
1476
      result = StartInstance(instance, False, reason, store_reason=False)
1477
      _StoreInstReasonTrail(instance.name, reason)
1478
      return result
1479
    except errors.HypervisorError, err:
1480
      _Fail("Failed to hard reboot instance %s: %s", instance.name, err)
1481
  else:
1482
    _Fail("Invalid reboot_type received: %s", reboot_type)
1483

    
1484

    
1485
def InstanceBalloonMemory(instance, memory):
1486
  """Resize an instance's memory.
1487

1488
  @type instance: L{objects.Instance}
1489
  @param instance: the instance object
1490
  @type memory: int
1491
  @param memory: new memory amount in MB
1492
  @rtype: None
1493

1494
  """
1495
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1496
  running = hyper.ListInstances()
1497
  if instance.name not in running:
1498
    logging.info("Instance %s is not running, cannot balloon", instance.name)
1499
    return
1500
  try:
1501
    hyper.BalloonInstanceMemory(instance, memory)
1502
  except errors.HypervisorError, err:
1503
    _Fail("Failed to balloon instance memory: %s", err, exc=True)
1504

    
1505

    
1506
def MigrationInfo(instance):
1507
  """Gather information about an instance to be migrated.
1508

1509
  @type instance: L{objects.Instance}
1510
  @param instance: the instance definition
1511

1512
  """
1513
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1514
  try:
1515
    info = hyper.MigrationInfo(instance)
1516
  except errors.HypervisorError, err:
1517
    _Fail("Failed to fetch migration information: %s", err, exc=True)
1518
  return info
1519

    
1520

    
1521
def AcceptInstance(instance, info, target):
1522
  """Prepare the node to accept an instance.
1523

1524
  @type instance: L{objects.Instance}
1525
  @param instance: the instance definition
1526
  @type info: string/data (opaque)
1527
  @param info: migration information, from the source node
1528
  @type target: string
1529
  @param target: target host (usually ip), on this node
1530

1531
  """
1532
  # TODO: why is this required only for DTS_EXT_MIRROR?
1533
  if instance.disk_template in constants.DTS_EXT_MIRROR:
1534
    # Create the symlinks, as the disks are not active
1535
    # in any way
1536
    try:
1537
      _GatherAndLinkBlockDevs(instance)
1538
    except errors.BlockDeviceError, err:
1539
      _Fail("Block device error: %s", err, exc=True)
1540

    
1541
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1542
  try:
1543
    hyper.AcceptInstance(instance, info, target)
1544
  except errors.HypervisorError, err:
1545
    if instance.disk_template in constants.DTS_EXT_MIRROR:
1546
      _RemoveBlockDevLinks(instance.name, instance.disks)
1547
    _Fail("Failed to accept instance: %s", err, exc=True)
1548

    
1549

    
1550
def FinalizeMigrationDst(instance, info, success):
1551
  """Finalize any preparation to accept an instance.
1552

1553
  @type instance: L{objects.Instance}
1554
  @param instance: the instance definition
1555
  @type info: string/data (opaque)
1556
  @param info: migration information, from the source node
1557
  @type success: boolean
1558
  @param success: whether the migration was a success or a failure
1559

1560
  """
1561
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1562
  try:
1563
    hyper.FinalizeMigrationDst(instance, info, success)
1564
  except errors.HypervisorError, err:
1565
    _Fail("Failed to finalize migration on the target node: %s", err, exc=True)
1566

    
1567

    
1568
def MigrateInstance(instance, target, live):
1569
  """Migrates an instance to another node.
1570

1571
  @type instance: L{objects.Instance}
1572
  @param instance: the instance definition
1573
  @type target: string
1574
  @param target: the target node name
1575
  @type live: boolean
1576
  @param live: whether the migration should be done live or not (the
1577
      interpretation of this parameter is left to the hypervisor)
1578
  @raise RPCFail: if migration fails for some reason
1579

1580
  """
1581
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1582

    
1583
  try:
1584
    hyper.MigrateInstance(instance, target, live)
1585
  except errors.HypervisorError, err:
1586
    _Fail("Failed to migrate instance: %s", err, exc=True)
1587

    
1588

    
1589
def FinalizeMigrationSource(instance, success, live):
1590
  """Finalize the instance migration on the source node.
1591

1592
  @type instance: L{objects.Instance}
1593
  @param instance: the instance definition of the migrated instance
1594
  @type success: bool
1595
  @param success: whether the migration succeeded or not
1596
  @type live: bool
1597
  @param live: whether the user requested a live migration or not
1598
  @raise RPCFail: If the execution fails for some reason
1599

1600
  """
1601
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1602

    
1603
  try:
1604
    hyper.FinalizeMigrationSource(instance, success, live)
1605
  except Exception, err:  # pylint: disable=W0703
1606
    _Fail("Failed to finalize the migration on the source node: %s", err,
1607
          exc=True)
1608

    
1609

    
1610
def GetMigrationStatus(instance):
1611
  """Get the migration status
1612

1613
  @type instance: L{objects.Instance}
1614
  @param instance: the instance that is being migrated
1615
  @rtype: L{objects.MigrationStatus}
1616
  @return: the status of the current migration (one of
1617
           L{constants.HV_MIGRATION_VALID_STATUSES}), plus any additional
1618
           progress info that can be retrieved from the hypervisor
1619
  @raise RPCFail: If the migration status cannot be retrieved
1620

1621
  """
1622
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1623
  try:
1624
    return hyper.GetMigrationStatus(instance)
1625
  except Exception, err:  # pylint: disable=W0703
1626
    _Fail("Failed to get migration status: %s", err, exc=True)
1627

    
1628

    
1629
def HotplugDevice(instance, action, dev_type, device, extra, seq):
1630
  """Hotplug a device
1631

1632
  Hotplug is currently supported only for KVM Hypervisor.
1633
  @type instance: L{objects.Instance}
1634
  @param instance: the instance to which we hotplug a device
1635
  @type action: string
1636
  @param action: the hotplug action to perform
1637
  @type dev_type: string
1638
  @param dev_type: the device type to hotplug
1639
  @type device: either L{objects.NIC} or L{objects.Disk}
1640
  @param device: the device object to hotplug
1641
  @type extra: string
1642
  @param extra: extra info used by hotplug code (e.g. disk link)
1643
  @type seq: int
1644
  @param seq: the index of the device from master perspective
1645
  @raise RPCFail: in case instance does not have KVM hypervisor
1646

1647
  """
1648
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1649
  try:
1650
    hyper.VerifyHotplugSupport(instance, action, dev_type)
1651
  except errors.HotplugError, err:
1652
    _Fail("Hotplug is not supported: %s", err)
1653

    
1654
  if action == constants.HOTPLUG_ACTION_ADD:
1655
    fn = hyper.HotAddDevice
1656
  elif action == constants.HOTPLUG_ACTION_REMOVE:
1657
    fn = hyper.HotDelDevice
1658
  elif action == constants.HOTPLUG_ACTION_MODIFY:
1659
    fn = hyper.HotModDevice
1660
  else:
1661
    assert action in constants.HOTPLUG_ALL_ACTIONS
1662

    
1663
  return fn(instance, dev_type, device, extra, seq)
1664

    
1665

    
1666
def HotplugSupported(instance):
1667
  """Checks if hotplug is generally supported.
1668

1669
  """
1670
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1671
  try:
1672
    hyper.HotplugSupported(instance)
1673
  except errors.HotplugError, err:
1674
    _Fail("Hotplug is not supported: %s", err)
1675

    
1676

    
1677
def BlockdevCreate(disk, size, owner, on_primary, info, excl_stor):
1678
  """Creates a block device for an instance.
1679

1680
  @type disk: L{objects.Disk}
1681
  @param disk: the object describing the disk we should create
1682
  @type size: int
1683
  @param size: the size of the physical underlying device, in MiB
1684
  @type owner: str
1685
  @param owner: the name of the instance for which disk is created,
1686
      used for device cache data
1687
  @type on_primary: boolean
1688
  @param on_primary:  indicates if it is the primary node or not
1689
  @type info: string
1690
  @param info: string that will be sent to the physical device
1691
      creation, used for example to set (LVM) tags on LVs
1692
  @type excl_stor: boolean
1693
  @param excl_stor: Whether exclusive_storage is active
1694

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

1699
  """
1700
  # TODO: remove the obsolete "size" argument
1701
  # pylint: disable=W0613
1702
  clist = []
1703
  if disk.children:
1704
    for child in disk.children:
1705
      try:
1706
        crdev = _RecursiveAssembleBD(child, owner, on_primary)
1707
      except errors.BlockDeviceError, err:
1708
        _Fail("Can't assemble device %s: %s", child, err)
1709
      if on_primary or disk.AssembleOnSecondary():
1710
        # we need the children open in case the device itself has to
1711
        # be assembled
1712
        try:
1713
          # pylint: disable=E1103
1714
          crdev.Open()
1715
        except errors.BlockDeviceError, err:
1716
          _Fail("Can't make child '%s' read-write: %s", child, err)
1717
      clist.append(crdev)
1718

    
1719
  try:
1720
    device = bdev.Create(disk, clist, excl_stor)
1721
  except errors.BlockDeviceError, err:
1722
    _Fail("Can't create block device: %s", err)
1723

    
1724
  if on_primary or disk.AssembleOnSecondary():
1725
    try:
1726
      device.Assemble()
1727
    except errors.BlockDeviceError, err:
1728
      _Fail("Can't assemble device after creation, unusual event: %s", err)
1729
    if on_primary or disk.OpenOnSecondary():
1730
      try:
1731
        device.Open(force=True)
1732
      except errors.BlockDeviceError, err:
1733
        _Fail("Can't make device r/w after creation, unusual event: %s", err)
1734
    DevCacheManager.UpdateCache(device.dev_path, owner,
1735
                                on_primary, disk.iv_name)
1736

    
1737
  device.SetInfo(info)
1738

    
1739
  return device.unique_id
1740

    
1741

    
1742
def _WipeDevice(path, offset, size):
1743
  """This function actually wipes the device.
1744

1745
  @param path: The path to the device to wipe
1746
  @param offset: The offset in MiB in the file
1747
  @param size: The size in MiB to write
1748

1749
  """
1750
  # Internal sizes are always in Mebibytes; if the following "dd" command
1751
  # should use a different block size the offset and size given to this
1752
  # function must be adjusted accordingly before being passed to "dd".
1753
  block_size = 1024 * 1024
1754

    
1755
  cmd = [constants.DD_CMD, "if=/dev/zero", "seek=%d" % offset,
1756
         "bs=%s" % block_size, "oflag=direct", "of=%s" % path,
1757
         "count=%d" % size]
1758
  result = utils.RunCmd(cmd)
1759

    
1760
  if result.failed:
1761
    _Fail("Wipe command '%s' exited with error: %s; output: %s", result.cmd,
1762
          result.fail_reason, result.output)
1763

    
1764

    
1765
def BlockdevWipe(disk, offset, size):
1766
  """Wipes a block device.
1767

1768
  @type disk: L{objects.Disk}
1769
  @param disk: the disk object we want to wipe
1770
  @type offset: int
1771
  @param offset: The offset in MiB in the file
1772
  @type size: int
1773
  @param size: The size in MiB to write
1774

1775
  """
1776
  try:
1777
    rdev = _RecursiveFindBD(disk)
1778
  except errors.BlockDeviceError:
1779
    rdev = None
1780

    
1781
  if not rdev:
1782
    _Fail("Cannot execute wipe for device %s: device not found", disk.iv_name)
1783

    
1784
  # Do cross verify some of the parameters
1785
  if offset < 0:
1786
    _Fail("Negative offset")
1787
  if size < 0:
1788
    _Fail("Negative size")
1789
  if offset > rdev.size:
1790
    _Fail("Offset is bigger than device size")
1791
  if (offset + size) > rdev.size:
1792
    _Fail("The provided offset and size to wipe is bigger than device size")
1793

    
1794
  _WipeDevice(rdev.dev_path, offset, size)
1795

    
1796

    
1797
def BlockdevPauseResumeSync(disks, pause):
1798
  """Pause or resume the sync of the block device.
1799

1800
  @type disks: list of L{objects.Disk}
1801
  @param disks: the disks object we want to pause/resume
1802
  @type pause: bool
1803
  @param pause: Wheater to pause or resume
1804

1805
  """
1806
  success = []
1807
  for disk in disks:
1808
    try:
1809
      rdev = _RecursiveFindBD(disk)
1810
    except errors.BlockDeviceError:
1811
      rdev = None
1812

    
1813
    if not rdev:
1814
      success.append((False, ("Cannot change sync for device %s:"
1815
                              " device not found" % disk.iv_name)))
1816
      continue
1817

    
1818
    result = rdev.PauseResumeSync(pause)
1819

    
1820
    if result:
1821
      success.append((result, None))
1822
    else:
1823
      if pause:
1824
        msg = "Pause"
1825
      else:
1826
        msg = "Resume"
1827
      success.append((result, "%s for device %s failed" % (msg, disk.iv_name)))
1828

    
1829
  return success
1830

    
1831

    
1832
def BlockdevRemove(disk):
1833
  """Remove a block device.
1834

1835
  @note: This is intended to be called recursively.
1836

1837
  @type disk: L{objects.Disk}
1838
  @param disk: the disk object we should remove
1839
  @rtype: boolean
1840
  @return: the success of the operation
1841

1842
  """
1843
  msgs = []
1844
  try:
1845
    rdev = _RecursiveFindBD(disk)
1846
  except errors.BlockDeviceError, err:
1847
    # probably can't attach
1848
    logging.info("Can't attach to device %s in remove", disk)
1849
    rdev = None
1850
  if rdev is not None:
1851
    r_path = rdev.dev_path
1852

    
1853
    def _TryRemove():
1854
      try:
1855
        rdev.Remove()
1856
        return []
1857
      except errors.BlockDeviceError, err:
1858
        return [str(err)]
1859

    
1860
    msgs.extend(utils.SimpleRetry([], _TryRemove,
1861
                                  constants.DISK_REMOVE_RETRY_INTERVAL,
1862
                                  constants.DISK_REMOVE_RETRY_TIMEOUT))
1863

    
1864
    if not msgs:
1865
      DevCacheManager.RemoveCache(r_path)
1866

    
1867
  if disk.children:
1868
    for child in disk.children:
1869
      try:
1870
        BlockdevRemove(child)
1871
      except RPCFail, err:
1872
        msgs.append(str(err))
1873

    
1874
  if msgs:
1875
    _Fail("; ".join(msgs))
1876

    
1877

    
1878
def _RecursiveAssembleBD(disk, owner, as_primary):
1879
  """Activate a block device for an instance.
1880

1881
  This is run on the primary and secondary nodes for an instance.
1882

1883
  @note: this function is called recursively.
1884

1885
  @type disk: L{objects.Disk}
1886
  @param disk: the disk we try to assemble
1887
  @type owner: str
1888
  @param owner: the name of the instance which owns the disk
1889
  @type as_primary: boolean
1890
  @param as_primary: if we should make the block device
1891
      read/write
1892

1893
  @return: the assembled device or None (in case no device
1894
      was assembled)
1895
  @raise errors.BlockDeviceError: in case there is an error
1896
      during the activation of the children or the device
1897
      itself
1898

1899
  """
1900
  children = []
1901
  if disk.children:
1902
    mcn = disk.ChildrenNeeded()
1903
    if mcn == -1:
1904
      mcn = 0 # max number of Nones allowed
1905
    else:
1906
      mcn = len(disk.children) - mcn # max number of Nones
1907
    for chld_disk in disk.children:
1908
      try:
1909
        cdev = _RecursiveAssembleBD(chld_disk, owner, as_primary)
1910
      except errors.BlockDeviceError, err:
1911
        if children.count(None) >= mcn:
1912
          raise
1913
        cdev = None
1914
        logging.error("Error in child activation (but continuing): %s",
1915
                      str(err))
1916
      children.append(cdev)
1917

    
1918
  if as_primary or disk.AssembleOnSecondary():
1919
    r_dev = bdev.Assemble(disk, children)
1920
    result = r_dev
1921
    if as_primary or disk.OpenOnSecondary():
1922
      r_dev.Open()
1923
    DevCacheManager.UpdateCache(r_dev.dev_path, owner,
1924
                                as_primary, disk.iv_name)
1925

    
1926
  else:
1927
    result = True
1928
  return result
1929

    
1930

    
1931
def BlockdevAssemble(disk, owner, as_primary, idx):
1932
  """Activate a block device for an instance.
1933

1934
  This is a wrapper over _RecursiveAssembleBD.
1935

1936
  @rtype: str or boolean
1937
  @return: a tuple with the C{/dev/...} path and the created symlink
1938
      for primary nodes, and (C{True}, C{True}) for secondary nodes
1939

1940
  """
1941
  try:
1942
    result = _RecursiveAssembleBD(disk, owner, as_primary)
1943
    if isinstance(result, bdev.BlockDev):
1944
      # pylint: disable=E1103
1945
      dev_path = result.dev_path
1946
      link_name = None
1947
      if as_primary:
1948
        link_name = _SymlinkBlockDev(owner, dev_path, idx)
1949
    elif result:
1950
      return result, result
1951
    else:
1952
      _Fail("Unexpected result from _RecursiveAssembleBD")
1953
  except errors.BlockDeviceError, err:
1954
    _Fail("Error while assembling disk: %s", err, exc=True)
1955
  except OSError, err:
1956
    _Fail("Error while symlinking disk: %s", err, exc=True)
1957

    
1958
  return dev_path, link_name
1959

    
1960

    
1961
def BlockdevShutdown(disk):
1962
  """Shut down a block device.
1963

1964
  First, if the device is assembled (Attach() is successful), then
1965
  the device is shutdown. Then the children of the device are
1966
  shutdown.
1967

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

1972
  @type disk: L{objects.Disk}
1973
  @param disk: the description of the disk we should
1974
      shutdown
1975
  @rtype: None
1976

1977
  """
1978
  msgs = []
1979
  r_dev = _RecursiveFindBD(disk)
1980
  if r_dev is not None:
1981
    r_path = r_dev.dev_path
1982
    try:
1983
      r_dev.Shutdown()
1984
      DevCacheManager.RemoveCache(r_path)
1985
    except errors.BlockDeviceError, err:
1986
      msgs.append(str(err))
1987

    
1988
  if disk.children:
1989
    for child in disk.children:
1990
      try:
1991
        BlockdevShutdown(child)
1992
      except RPCFail, err:
1993
        msgs.append(str(err))
1994

    
1995
  if msgs:
1996
    _Fail("; ".join(msgs))
1997

    
1998

    
1999
def BlockdevAddchildren(parent_cdev, new_cdevs):
2000
  """Extend a mirrored block device.
2001

2002
  @type parent_cdev: L{objects.Disk}
2003
  @param parent_cdev: the disk to which we should add children
2004
  @type new_cdevs: list of L{objects.Disk}
2005
  @param new_cdevs: the list of children which we should add
2006
  @rtype: None
2007

2008
  """
2009
  parent_bdev = _RecursiveFindBD(parent_cdev)
2010
  if parent_bdev is None:
2011
    _Fail("Can't find parent device '%s' in add children", parent_cdev)
2012
  new_bdevs = [_RecursiveFindBD(disk) for disk in new_cdevs]
2013
  if new_bdevs.count(None) > 0:
2014
    _Fail("Can't find new device(s) to add: %s:%s", new_bdevs, new_cdevs)
2015
  parent_bdev.AddChildren(new_bdevs)
2016

    
2017

    
2018
def BlockdevRemovechildren(parent_cdev, new_cdevs):
2019
  """Shrink a mirrored block device.
2020

2021
  @type parent_cdev: L{objects.Disk}
2022
  @param parent_cdev: the disk from which we should remove children
2023
  @type new_cdevs: list of L{objects.Disk}
2024
  @param new_cdevs: the list of children which we should remove
2025
  @rtype: None
2026

2027
  """
2028
  parent_bdev = _RecursiveFindBD(parent_cdev)
2029
  if parent_bdev is None:
2030
    _Fail("Can't find parent device '%s' in remove children", parent_cdev)
2031
  devs = []
2032
  for disk in new_cdevs:
2033
    rpath = disk.StaticDevPath()
2034
    if rpath is None:
2035
      bd = _RecursiveFindBD(disk)
2036
      if bd is None:
2037
        _Fail("Can't find device %s while removing children", disk)
2038
      else:
2039
        devs.append(bd.dev_path)
2040
    else:
2041
      if not utils.IsNormAbsPath(rpath):
2042
        _Fail("Strange path returned from StaticDevPath: '%s'", rpath)
2043
      devs.append(rpath)
2044
  parent_bdev.RemoveChildren(devs)
2045

    
2046

    
2047
def BlockdevGetmirrorstatus(disks):
2048
  """Get the mirroring status of a list of devices.
2049

2050
  @type disks: list of L{objects.Disk}
2051
  @param disks: the list of disks which we should query
2052
  @rtype: disk
2053
  @return: List of L{objects.BlockDevStatus}, one for each disk
2054
  @raise errors.BlockDeviceError: if any of the disks cannot be
2055
      found
2056

2057
  """
2058
  stats = []
2059
  for dsk in disks:
2060
    rbd = _RecursiveFindBD(dsk)
2061
    if rbd is None:
2062
      _Fail("Can't find device %s", dsk)
2063

    
2064
    stats.append(rbd.CombinedSyncStatus())
2065

    
2066
  return stats
2067

    
2068

    
2069
def BlockdevGetmirrorstatusMulti(disks):
2070
  """Get the mirroring status of a list of devices.
2071

2072
  @type disks: list of L{objects.Disk}
2073
  @param disks: the list of disks which we should query
2074
  @rtype: disk
2075
  @return: List of tuples, (bool, status), one for each disk; bool denotes
2076
    success/failure, status is L{objects.BlockDevStatus} on success, string
2077
    otherwise
2078

2079
  """
2080
  result = []
2081
  for disk in disks:
2082
    try:
2083
      rbd = _RecursiveFindBD(disk)
2084
      if rbd is None:
2085
        result.append((False, "Can't find device %s" % disk))
2086
        continue
2087

    
2088
      status = rbd.CombinedSyncStatus()
2089
    except errors.BlockDeviceError, err:
2090
      logging.exception("Error while getting disk status")
2091
      result.append((False, str(err)))
2092
    else:
2093
      result.append((True, status))
2094

    
2095
  assert len(disks) == len(result)
2096

    
2097
  return result
2098

    
2099

    
2100
def _RecursiveFindBD(disk):
2101
  """Check if a device is activated.
2102

2103
  If so, return information about the real device.
2104

2105
  @type disk: L{objects.Disk}
2106
  @param disk: the disk object we need to find
2107

2108
  @return: None if the device can't be found,
2109
      otherwise the device instance
2110

2111
  """
2112
  children = []
2113
  if disk.children:
2114
    for chdisk in disk.children:
2115
      children.append(_RecursiveFindBD(chdisk))
2116

    
2117
  return bdev.FindDevice(disk, children)
2118

    
2119

    
2120
def _OpenRealBD(disk):
2121
  """Opens the underlying block device of a disk.
2122

2123
  @type disk: L{objects.Disk}
2124
  @param disk: the disk object we want to open
2125

2126
  """
2127
  real_disk = _RecursiveFindBD(disk)
2128
  if real_disk is None:
2129
    _Fail("Block device '%s' is not set up", disk)
2130

    
2131
  real_disk.Open()
2132

    
2133
  return real_disk
2134

    
2135

    
2136
def BlockdevFind(disk):
2137
  """Check if a device is activated.
2138

2139
  If it is, return information about the real device.
2140

2141
  @type disk: L{objects.Disk}
2142
  @param disk: the disk to find
2143
  @rtype: None or objects.BlockDevStatus
2144
  @return: None if the disk cannot be found, otherwise a the current
2145
           information
2146

2147
  """
2148
  try:
2149
    rbd = _RecursiveFindBD(disk)
2150
  except errors.BlockDeviceError, err:
2151
    _Fail("Failed to find device: %s", err, exc=True)
2152

    
2153
  if rbd is None:
2154
    return None
2155

    
2156
  return rbd.GetSyncStatus()
2157

    
2158

    
2159
def BlockdevGetsize(disks):
2160
  """Computes the size of the given disks.
2161

2162
  If a disk is not found, returns None instead.
2163

2164
  @type disks: list of L{objects.Disk}
2165
  @param disks: the list of disk to compute the size for
2166
  @rtype: list
2167
  @return: list with elements None if the disk cannot be found,
2168
      otherwise the size
2169

2170
  """
2171
  result = []
2172
  for cf in disks:
2173
    try:
2174
      rbd = _RecursiveFindBD(cf)
2175
    except errors.BlockDeviceError:
2176
      result.append(None)
2177
      continue
2178
    if rbd is None:
2179
      result.append(None)
2180
    else:
2181
      result.append(rbd.GetActualSize())
2182
  return result
2183

    
2184

    
2185
def BlockdevExport(disk, dest_node, dest_path, cluster_name):
2186
  """Export a block device to a remote node.
2187

2188
  @type disk: L{objects.Disk}
2189
  @param disk: the description of the disk to export
2190
  @type dest_node: str
2191
  @param dest_node: the destination node to export to
2192
  @type dest_path: str
2193
  @param dest_path: the destination path on the target node
2194
  @type cluster_name: str
2195
  @param cluster_name: the cluster name, needed for SSH hostalias
2196
  @rtype: None
2197

2198
  """
2199
  real_disk = _OpenRealBD(disk)
2200

    
2201
  # the block size on the read dd is 1MiB to match our units
2202
  expcmd = utils.BuildShellCmd("set -e; set -o pipefail; "
2203
                               "dd if=%s bs=1048576 count=%s",
2204
                               real_disk.dev_path, str(disk.size))
2205

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

    
2215
  remotecmd = _GetSshRunner(cluster_name).BuildCmd(dest_node,
2216
                                                   constants.SSH_LOGIN_USER,
2217
                                                   destcmd)
2218

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

    
2222
  result = utils.RunCmd(["bash", "-c", command])
2223

    
2224
  if result.failed:
2225
    _Fail("Disk copy command '%s' returned error: %s"
2226
          " output: %s", command, result.fail_reason, result.output)
2227

    
2228

    
2229
def UploadFile(file_name, data, mode, uid, gid, atime, mtime):
2230
  """Write a file to the filesystem.
2231

2232
  This allows the master to overwrite(!) a file. It will only perform
2233
  the operation if the file belongs to a list of configuration files.
2234

2235
  @type file_name: str
2236
  @param file_name: the target file name
2237
  @type data: str
2238
  @param data: the new contents of the file
2239
  @type mode: int
2240
  @param mode: the mode to give the file (can be None)
2241
  @type uid: string
2242
  @param uid: the owner of the file
2243
  @type gid: string
2244
  @param gid: the group of the file
2245
  @type atime: float
2246
  @param atime: the atime to set on the file (can be None)
2247
  @type mtime: float
2248
  @param mtime: the mtime to set on the file (can be None)
2249
  @rtype: None
2250

2251
  """
2252
  file_name = vcluster.LocalizeVirtualPath(file_name)
2253

    
2254
  if not os.path.isabs(file_name):
2255
    _Fail("Filename passed to UploadFile is not absolute: '%s'", file_name)
2256

    
2257
  if file_name not in _ALLOWED_UPLOAD_FILES:
2258
    _Fail("Filename passed to UploadFile not in allowed upload targets: '%s'",
2259
          file_name)
2260

    
2261
  raw_data = _Decompress(data)
2262

    
2263
  if not (isinstance(uid, basestring) and isinstance(gid, basestring)):
2264
    _Fail("Invalid username/groupname type")
2265

    
2266
  getents = runtime.GetEnts()
2267
  uid = getents.LookupUser(uid)
2268
  gid = getents.LookupGroup(gid)
2269

    
2270
  utils.SafeWriteFile(file_name, None,
2271
                      data=raw_data, mode=mode, uid=uid, gid=gid,
2272
                      atime=atime, mtime=mtime)
2273

    
2274

    
2275
def RunOob(oob_program, command, node, timeout):
2276
  """Executes oob_program with given command on given node.
2277

2278
  @param oob_program: The path to the executable oob_program
2279
  @param command: The command to invoke on oob_program
2280
  @param node: The node given as an argument to the program
2281
  @param timeout: Timeout after which we kill the oob program
2282

2283
  @return: stdout
2284
  @raise RPCFail: If execution fails for some reason
2285

2286
  """
2287
  result = utils.RunCmd([oob_program, command, node], timeout=timeout)
2288

    
2289
  if result.failed:
2290
    _Fail("'%s' failed with reason '%s'; output: %s", result.cmd,
2291
          result.fail_reason, result.output)
2292

    
2293
  return result.stdout
2294

    
2295

    
2296
def _OSOndiskAPIVersion(os_dir):
2297
  """Compute and return the API version of a given OS.
2298

2299
  This function will try to read the API version of the OS residing in
2300
  the 'os_dir' directory.
2301

2302
  @type os_dir: str
2303
  @param os_dir: the directory in which we should look for the OS
2304
  @rtype: tuple
2305
  @return: tuple (status, data) with status denoting the validity and
2306
      data holding either the vaid versions or an error message
2307

2308
  """
2309
  api_file = utils.PathJoin(os_dir, constants.OS_API_FILE)
2310

    
2311
  try:
2312
    st = os.stat(api_file)
2313
  except EnvironmentError, err:
2314
    return False, ("Required file '%s' not found under path %s: %s" %
2315
                   (constants.OS_API_FILE, os_dir, utils.ErrnoOrStr(err)))
2316

    
2317
  if not stat.S_ISREG(stat.S_IFMT(st.st_mode)):
2318
    return False, ("File '%s' in %s is not a regular file" %
2319
                   (constants.OS_API_FILE, os_dir))
2320

    
2321
  try:
2322
    api_versions = utils.ReadFile(api_file).splitlines()
2323
  except EnvironmentError, err:
2324
    return False, ("Error while reading the API version file at %s: %s" %
2325
                   (api_file, utils.ErrnoOrStr(err)))
2326

    
2327
  try:
2328
    api_versions = [int(version.strip()) for version in api_versions]
2329
  except (TypeError, ValueError), err:
2330
    return False, ("API version(s) can't be converted to integer: %s" %
2331
                   str(err))
2332

    
2333
  return True, api_versions
2334

    
2335

    
2336
def DiagnoseOS(top_dirs=None):
2337
  """Compute the validity for all OSes.
2338

2339
  @type top_dirs: list
2340
  @param top_dirs: the list of directories in which to
2341
      search (if not given defaults to
2342
      L{pathutils.OS_SEARCH_PATH})
2343
  @rtype: list of L{objects.OS}
2344
  @return: a list of tuples (name, path, status, diagnose, variants,
2345
      parameters, api_version) for all (potential) OSes under all
2346
      search paths, where:
2347
          - name is the (potential) OS name
2348
          - path is the full path to the OS
2349
          - status True/False is the validity of the OS
2350
          - diagnose is the error message for an invalid OS, otherwise empty
2351
          - variants is a list of supported OS variants, if any
2352
          - parameters is a list of (name, help) parameters, if any
2353
          - api_version is a list of support OS API versions
2354

2355
  """
2356
  if top_dirs is None:
2357
    top_dirs = pathutils.OS_SEARCH_PATH
2358

    
2359
  result = []
2360
  for dir_name in top_dirs:
2361
    if os.path.isdir(dir_name):
2362
      try:
2363
        f_names = utils.ListVisibleFiles(dir_name)
2364
      except EnvironmentError, err:
2365
        logging.exception("Can't list the OS directory %s: %s", dir_name, err)
2366
        break
2367
      for name in f_names:
2368
        os_path = utils.PathJoin(dir_name, name)
2369
        status, os_inst = _TryOSFromDisk(name, base_dir=dir_name)
2370
        if status:
2371
          diagnose = ""
2372
          variants = os_inst.supported_variants
2373
          parameters = os_inst.supported_parameters
2374
          api_versions = os_inst.api_versions
2375
        else:
2376
          diagnose = os_inst
2377
          variants = parameters = api_versions = []
2378
        result.append((name, os_path, status, diagnose, variants,
2379
                       parameters, api_versions))
2380

    
2381
  return result
2382

    
2383

    
2384
def _TryOSFromDisk(name, base_dir=None):
2385
  """Create an OS instance from disk.
2386

2387
  This function will return an OS instance if the given name is a
2388
  valid OS name.
2389

2390
  @type base_dir: string
2391
  @keyword base_dir: Base directory containing OS installations.
2392
                     Defaults to a search in all the OS_SEARCH_PATH dirs.
2393
  @rtype: tuple
2394
  @return: success and either the OS instance if we find a valid one,
2395
      or error message
2396

2397
  """
2398
  if base_dir is None:
2399
    os_dir = utils.FindFile(name, pathutils.OS_SEARCH_PATH, os.path.isdir)
2400
  else:
2401
    os_dir = utils.FindFile(name, [base_dir], os.path.isdir)
2402

    
2403
  if os_dir is None:
2404
    return False, "Directory for OS %s not found in search path" % name
2405

    
2406
  status, api_versions = _OSOndiskAPIVersion(os_dir)
2407
  if not status:
2408
    # push the error up
2409
    return status, api_versions
2410

    
2411
  if not constants.OS_API_VERSIONS.intersection(api_versions):
2412
    return False, ("API version mismatch for path '%s': found %s, want %s." %
2413
                   (os_dir, api_versions, constants.OS_API_VERSIONS))
2414

    
2415
  # OS Files dictionary, we will populate it with the absolute path
2416
  # names; if the value is True, then it is a required file, otherwise
2417
  # an optional one
2418
  os_files = dict.fromkeys(constants.OS_SCRIPTS, True)
2419

    
2420
  if max(api_versions) >= constants.OS_API_V15:
2421
    os_files[constants.OS_VARIANTS_FILE] = False
2422

    
2423
  if max(api_versions) >= constants.OS_API_V20:
2424
    os_files[constants.OS_PARAMETERS_FILE] = True
2425
  else:
2426
    del os_files[constants.OS_SCRIPT_VERIFY]
2427

    
2428
  for (filename, required) in os_files.items():
2429
    os_files[filename] = utils.PathJoin(os_dir, filename)
2430

    
2431
    try:
2432
      st = os.stat(os_files[filename])
2433
    except EnvironmentError, err:
2434
      if err.errno == errno.ENOENT and not required:
2435
        del os_files[filename]
2436
        continue
2437
      return False, ("File '%s' under path '%s' is missing (%s)" %
2438
                     (filename, os_dir, utils.ErrnoOrStr(err)))
2439

    
2440
    if not stat.S_ISREG(stat.S_IFMT(st.st_mode)):
2441
      return False, ("File '%s' under path '%s' is not a regular file" %
2442
                     (filename, os_dir))
2443

    
2444
    if filename in constants.OS_SCRIPTS:
2445
      if stat.S_IMODE(st.st_mode) & stat.S_IXUSR != stat.S_IXUSR:
2446
        return False, ("File '%s' under path '%s' is not executable" %
2447
                       (filename, os_dir))
2448

    
2449
  variants = []
2450
  if constants.OS_VARIANTS_FILE in os_files:
2451
    variants_file = os_files[constants.OS_VARIANTS_FILE]
2452
    try:
2453
      variants = \
2454
        utils.FilterEmptyLinesAndComments(utils.ReadFile(variants_file))
2455
    except EnvironmentError, err:
2456
      # we accept missing files, but not other errors
2457
      if err.errno != errno.ENOENT:
2458
        return False, ("Error while reading the OS variants file at %s: %s" %
2459
                       (variants_file, utils.ErrnoOrStr(err)))
2460

    
2461
  parameters = []
2462
  if constants.OS_PARAMETERS_FILE in os_files:
2463
    parameters_file = os_files[constants.OS_PARAMETERS_FILE]
2464
    try:
2465
      parameters = utils.ReadFile(parameters_file).splitlines()
2466
    except EnvironmentError, err:
2467
      return False, ("Error while reading the OS parameters file at %s: %s" %
2468
                     (parameters_file, utils.ErrnoOrStr(err)))
2469
    parameters = [v.split(None, 1) for v in parameters]
2470

    
2471
  os_obj = objects.OS(name=name, path=os_dir,
2472
                      create_script=os_files[constants.OS_SCRIPT_CREATE],
2473
                      export_script=os_files[constants.OS_SCRIPT_EXPORT],
2474
                      import_script=os_files[constants.OS_SCRIPT_IMPORT],
2475
                      rename_script=os_files[constants.OS_SCRIPT_RENAME],
2476
                      verify_script=os_files.get(constants.OS_SCRIPT_VERIFY,
2477
                                                 None),
2478
                      supported_variants=variants,
2479
                      supported_parameters=parameters,
2480
                      api_versions=api_versions)
2481
  return True, os_obj
2482

    
2483

    
2484
def OSFromDisk(name, base_dir=None):
2485
  """Create an OS instance from disk.
2486

2487
  This function will return an OS instance if the given name is a
2488
  valid OS name. Otherwise, it will raise an appropriate
2489
  L{RPCFail} exception, detailing why this is not a valid OS.
2490

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

2494
  @type base_dir: string
2495
  @keyword base_dir: Base directory containing OS installations.
2496
                     Defaults to a search in all the OS_SEARCH_PATH dirs.
2497
  @rtype: L{objects.OS}
2498
  @return: the OS instance if we find a valid one
2499
  @raise RPCFail: if we don't find a valid OS
2500

2501
  """
2502
  name_only = objects.OS.GetName(name)
2503
  status, payload = _TryOSFromDisk(name_only, base_dir)
2504

    
2505
  if not status:
2506
    _Fail(payload)
2507

    
2508
  return payload
2509

    
2510

    
2511
def OSCoreEnv(os_name, inst_os, os_params, debug=0):
2512
  """Calculate the basic environment for an os script.
2513

2514
  @type os_name: str
2515
  @param os_name: full operating system name (including variant)
2516
  @type inst_os: L{objects.OS}
2517
  @param inst_os: operating system for which the environment is being built
2518
  @type os_params: dict
2519
  @param os_params: the OS parameters
2520
  @type debug: integer
2521
  @param debug: debug level (0 or 1, for OS Api 10)
2522
  @rtype: dict
2523
  @return: dict of environment variables
2524
  @raise errors.BlockDeviceError: if the block device
2525
      cannot be found
2526

2527
  """
2528
  result = {}
2529
  api_version = \
2530
    max(constants.OS_API_VERSIONS.intersection(inst_os.api_versions))
2531
  result["OS_API_VERSION"] = "%d" % api_version
2532
  result["OS_NAME"] = inst_os.name
2533
  result["DEBUG_LEVEL"] = "%d" % debug
2534

    
2535
  # OS variants
2536
  if api_version >= constants.OS_API_V15 and inst_os.supported_variants:
2537
    variant = objects.OS.GetVariant(os_name)
2538
    if not variant:
2539
      variant = inst_os.supported_variants[0]
2540
  else:
2541
    variant = ""
2542
  result["OS_VARIANT"] = variant
2543

    
2544
  # OS params
2545
  for pname, pvalue in os_params.items():
2546
    result["OSP_%s" % pname.upper()] = pvalue
2547

    
2548
  # Set a default path otherwise programs called by OS scripts (or
2549
  # even hooks called from OS scripts) might break, and we don't want
2550
  # to have each script require setting a PATH variable
2551
  result["PATH"] = constants.HOOKS_PATH
2552

    
2553
  return result
2554

    
2555

    
2556
def OSEnvironment(instance, inst_os, debug=0):
2557
  """Calculate the environment for an os script.
2558

2559
  @type instance: L{objects.Instance}
2560
  @param instance: target instance for the os script run
2561
  @type inst_os: L{objects.OS}
2562
  @param inst_os: operating system for which the environment is being built
2563
  @type debug: integer
2564
  @param debug: debug level (0 or 1, for OS Api 10)
2565
  @rtype: dict
2566
  @return: dict of environment variables
2567
  @raise errors.BlockDeviceError: if the block device
2568
      cannot be found
2569

2570
  """
2571
  result = OSCoreEnv(instance.os, inst_os, instance.osparams, debug=debug)
2572

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

    
2576
  result["HYPERVISOR"] = instance.hypervisor
2577
  result["DISK_COUNT"] = "%d" % len(instance.disks)
2578
  result["NIC_COUNT"] = "%d" % len(instance.nics)
2579
  result["INSTANCE_SECONDARY_NODES"] = \
2580
      ("%s" % " ".join(instance.secondary_nodes))
2581

    
2582
  # Disks
2583
  for idx, disk in enumerate(instance.disks):
2584
    real_disk = _OpenRealBD(disk)
2585
    result["DISK_%d_PATH" % idx] = real_disk.dev_path
2586
    result["DISK_%d_ACCESS" % idx] = disk.mode
2587
    result["DISK_%d_UUID" % idx] = disk.uuid
2588
    if disk.name:
2589
      result["DISK_%d_NAME" % idx] = disk.name
2590
    if constants.HV_DISK_TYPE in instance.hvparams:
2591
      result["DISK_%d_FRONTEND_TYPE" % idx] = \
2592
        instance.hvparams[constants.HV_DISK_TYPE]
2593
    if disk.dev_type in constants.LDS_BLOCK:
2594
      result["DISK_%d_BACKEND_TYPE" % idx] = "block"
2595
    elif disk.dev_type == constants.LD_FILE:
2596
      result["DISK_%d_BACKEND_TYPE" % idx] = \
2597
        "file:%s" % disk.physical_id[0]
2598

    
2599
  # NICs
2600
  for idx, nic in enumerate(instance.nics):
2601
    result["NIC_%d_MAC" % idx] = nic.mac
2602
    result["NIC_%d_UUID" % idx] = nic.uuid
2603
    if nic.name:
2604
      result["NIC_%d_NAME" % idx] = nic.name
2605
    if nic.ip:
2606
      result["NIC_%d_IP" % idx] = nic.ip
2607
    result["NIC_%d_MODE" % idx] = nic.nicparams[constants.NIC_MODE]
2608
    if nic.nicparams[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED:
2609
      result["NIC_%d_BRIDGE" % idx] = nic.nicparams[constants.NIC_LINK]
2610
    if nic.nicparams[constants.NIC_LINK]:
2611
      result["NIC_%d_LINK" % idx] = nic.nicparams[constants.NIC_LINK]
2612
    if nic.netinfo:
2613
      nobj = objects.Network.FromDict(nic.netinfo)
2614
      result.update(nobj.HooksDict("NIC_%d_" % idx))
2615
    if constants.HV_NIC_TYPE in instance.hvparams:
2616
      result["NIC_%d_FRONTEND_TYPE" % idx] = \
2617
        instance.hvparams[constants.HV_NIC_TYPE]
2618

    
2619
  # HV/BE params
2620
  for source, kind in [(instance.beparams, "BE"), (instance.hvparams, "HV")]:
2621
    for key, value in source.items():
2622
      result["INSTANCE_%s_%s" % (kind, key)] = str(value)
2623

    
2624
  return result
2625

    
2626

    
2627
def DiagnoseExtStorage(top_dirs=None):
2628
  """Compute the validity for all ExtStorage Providers.
2629

2630
  @type top_dirs: list
2631
  @param top_dirs: the list of directories in which to
2632
      search (if not given defaults to
2633
      L{pathutils.ES_SEARCH_PATH})
2634
  @rtype: list of L{objects.ExtStorage}
2635
  @return: a list of tuples (name, path, status, diagnose, parameters)
2636
      for all (potential) ExtStorage Providers under all
2637
      search paths, where:
2638
          - name is the (potential) ExtStorage Provider
2639
          - path is the full path to the ExtStorage Provider
2640
          - status True/False is the validity of the ExtStorage Provider
2641
          - diagnose is the error message for an invalid ExtStorage Provider,
2642
            otherwise empty
2643
          - parameters is a list of (name, help) parameters, if any
2644

2645
  """
2646
  if top_dirs is None:
2647
    top_dirs = pathutils.ES_SEARCH_PATH
2648

    
2649
  result = []
2650
  for dir_name in top_dirs:
2651
    if os.path.isdir(dir_name):
2652
      try:
2653
        f_names = utils.ListVisibleFiles(dir_name)
2654
      except EnvironmentError, err:
2655
        logging.exception("Can't list the ExtStorage directory %s: %s",
2656
                          dir_name, err)
2657
        break
2658
      for name in f_names:
2659
        es_path = utils.PathJoin(dir_name, name)
2660
        status, es_inst = bdev.ExtStorageFromDisk(name, base_dir=dir_name)
2661
        if status:
2662
          diagnose = ""
2663
          parameters = es_inst.supported_parameters
2664
        else:
2665
          diagnose = es_inst
2666
          parameters = []
2667
        result.append((name, es_path, status, diagnose, parameters))
2668

    
2669
  return result
2670

    
2671

    
2672
def BlockdevGrow(disk, amount, dryrun, backingstore):
2673
  """Grow a stack of block devices.
2674

2675
  This function is called recursively, with the childrens being the
2676
  first ones to resize.
2677

2678
  @type disk: L{objects.Disk}
2679
  @param disk: the disk to be grown
2680
  @type amount: integer
2681
  @param amount: the amount (in mebibytes) to grow with
2682
  @type dryrun: boolean
2683
  @param dryrun: whether to execute the operation in simulation mode
2684
      only, without actually increasing the size
2685
  @param backingstore: whether to execute the operation on backing storage
2686
      only, or on "logical" storage only; e.g. DRBD is logical storage,
2687
      whereas LVM, file, RBD are backing storage
2688
  @rtype: (status, result)
2689
  @return: a tuple with the status of the operation (True/False), and
2690
      the errors message if status is False
2691

2692
  """
2693
  r_dev = _RecursiveFindBD(disk)
2694
  if r_dev is None:
2695
    _Fail("Cannot find block device %s", disk)
2696

    
2697
  try:
2698
    r_dev.Grow(amount, dryrun, backingstore)
2699
  except errors.BlockDeviceError, err:
2700
    _Fail("Failed to grow block device: %s", err, exc=True)
2701

    
2702

    
2703
def BlockdevSnapshot(disk, snapshot_name=None):
2704
  """Create a snapshot copy of a block device.
2705

2706
  This function is called recursively, and the snapshot is actually created
2707
  just for the leaf lvm backend device.
2708

2709
  @type disk: L{objects.Disk}
2710
  @param disk: the disk to be snapshotted
2711
  @rtype: string
2712
  @return: snapshot disk ID as (vg, lv)
2713

2714
  """
2715
  if disk.dev_type == constants.LD_DRBD8:
2716
    if not disk.children:
2717
      _Fail("DRBD device '%s' without backing storage cannot be snapshotted",
2718
            disk.unique_id)
2719
    return BlockdevSnapshot(disk.children[0])
2720
  elif disk.dev_type == constants.LD_LV and not snapshot_name:
2721
    r_dev = _RecursiveFindBD(disk)
2722
    if r_dev is not None:
2723
      # FIXME: choose a saner value for the snapshot size
2724
      # let's stay on the safe side and ask for the full size, for now
2725
      return r_dev.Snapshot(disk.size)
2726
    else:
2727
      _Fail("Cannot find block device %s", disk)
2728
  elif disk.dev_type == constants.DT_EXT:
2729
    r_dev = _RecursiveFindBD(disk)
2730
    if r_dev is not None:
2731
      r_dev.Snapshot(snapshot_name)
2732
    else:
2733
      _Fail("Cannot find block device %s", disk)
2734
  else:
2735
    _Fail("Cannot snapshot non-lvm block device '%s' of type '%s'",
2736
          disk.unique_id, disk.dev_type)
2737

    
2738

    
2739
def BlockdevSetInfo(disk, info):
2740
  """Sets 'metadata' information on block devices.
2741

2742
  This function sets 'info' metadata on block devices. Initial
2743
  information is set at device creation; this function should be used
2744
  for example after renames.
2745

2746
  @type disk: L{objects.Disk}
2747
  @param disk: the disk to be grown
2748
  @type info: string
2749
  @param info: new 'info' metadata
2750
  @rtype: (status, result)
2751
  @return: a tuple with the status of the operation (True/False), and
2752
      the errors message if status is False
2753

2754
  """
2755
  r_dev = _RecursiveFindBD(disk)
2756
  if r_dev is None:
2757
    _Fail("Cannot find block device %s", disk)
2758

    
2759
  try:
2760
    r_dev.SetInfo(info)
2761
  except errors.BlockDeviceError, err:
2762
    _Fail("Failed to set information on block device: %s", err, exc=True)
2763

    
2764

    
2765
def FinalizeExport(instance, snap_disks):
2766
  """Write out the export configuration information.
2767

2768
  @type instance: L{objects.Instance}
2769
  @param instance: the instance which we export, used for
2770
      saving configuration
2771
  @type snap_disks: list of L{objects.Disk}
2772
  @param snap_disks: list of snapshot block devices, which
2773
      will be used to get the actual name of the dump file
2774

2775
  @rtype: None
2776

2777
  """
2778
  destdir = utils.PathJoin(pathutils.EXPORT_DIR, instance.name + ".new")
2779
  finaldestdir = utils.PathJoin(pathutils.EXPORT_DIR, instance.name)
2780

    
2781
  config = objects.SerializableConfigParser()
2782

    
2783
  config.add_section(constants.INISECT_EXP)
2784
  config.set(constants.INISECT_EXP, "version", "0")
2785
  config.set(constants.INISECT_EXP, "timestamp", "%d" % int(time.time()))
2786
  config.set(constants.INISECT_EXP, "source", instance.primary_node)
2787
  config.set(constants.INISECT_EXP, "os", instance.os)
2788
  config.set(constants.INISECT_EXP, "compression", "none")
2789

    
2790
  config.add_section(constants.INISECT_INS)
2791
  config.set(constants.INISECT_INS, "name", instance.name)
2792
  config.set(constants.INISECT_INS, "maxmem", "%d" %
2793
             instance.beparams[constants.BE_MAXMEM])
2794
  config.set(constants.INISECT_INS, "minmem", "%d" %
2795
             instance.beparams[constants.BE_MINMEM])
2796
  # "memory" is deprecated, but useful for exporting to old ganeti versions
2797
  config.set(constants.INISECT_INS, "memory", "%d" %
2798
             instance.beparams[constants.BE_MAXMEM])
2799
  config.set(constants.INISECT_INS, "vcpus", "%d" %
2800
             instance.beparams[constants.BE_VCPUS])
2801
  config.set(constants.INISECT_INS, "disk_template", instance.disk_template)
2802
  config.set(constants.INISECT_INS, "hypervisor", instance.hypervisor)
2803
  config.set(constants.INISECT_INS, "tags", " ".join(instance.GetTags()))
2804

    
2805
  nic_total = 0
2806
  for nic_count, nic in enumerate(instance.nics):
2807
    nic_total += 1
2808
    config.set(constants.INISECT_INS, "nic%d_mac" %
2809
               nic_count, "%s" % nic.mac)
2810
    config.set(constants.INISECT_INS, "nic%d_ip" % nic_count, "%s" % nic.ip)
2811
    config.set(constants.INISECT_INS, "nic%d_network" % nic_count,
2812
               "%s" % nic.network)
2813
    for param in constants.NICS_PARAMETER_TYPES:
2814
      config.set(constants.INISECT_INS, "nic%d_%s" % (nic_count, param),
2815
                 "%s" % nic.nicparams.get(param, None))
2816
  # TODO: redundant: on load can read nics until it doesn't exist
2817
  config.set(constants.INISECT_INS, "nic_count", "%d" % nic_total)
2818

    
2819
  disk_total = 0
2820
  for disk_count, disk in enumerate(snap_disks):
2821
    if disk:
2822
      disk_total += 1
2823
      config.set(constants.INISECT_INS, "disk%d_ivname" % disk_count,
2824
                 ("%s" % disk.iv_name))
2825
      config.set(constants.INISECT_INS, "disk%d_dump" % disk_count,
2826
                 ("%s" % disk.physical_id[1]))
2827
      config.set(constants.INISECT_INS, "disk%d_size" % disk_count,
2828
                 ("%d" % disk.size))
2829

    
2830
  config.set(constants.INISECT_INS, "disk_count", "%d" % disk_total)
2831

    
2832
  # New-style hypervisor/backend parameters
2833

    
2834
  config.add_section(constants.INISECT_HYP)
2835
  for name, value in instance.hvparams.items():
2836
    if name not in constants.HVC_GLOBALS:
2837
      config.set(constants.INISECT_HYP, name, str(value))
2838

    
2839
  config.add_section(constants.INISECT_BEP)
2840
  for name, value in instance.beparams.items():
2841
    config.set(constants.INISECT_BEP, name, str(value))
2842

    
2843
  config.add_section(constants.INISECT_OSP)
2844
  for name, value in instance.osparams.items():
2845
    config.set(constants.INISECT_OSP, name, str(value))
2846

    
2847
  utils.WriteFile(utils.PathJoin(destdir, constants.EXPORT_CONF_FILE),
2848
                  data=config.Dumps())
2849
  shutil.rmtree(finaldestdir, ignore_errors=True)
2850
  shutil.move(destdir, finaldestdir)
2851

    
2852

    
2853
def ExportInfo(dest):
2854
  """Get export configuration information.
2855

2856
  @type dest: str
2857
  @param dest: directory containing the export
2858

2859
  @rtype: L{objects.SerializableConfigParser}
2860
  @return: a serializable config file containing the
2861
      export info
2862

2863
  """
2864
  cff = utils.PathJoin(dest, constants.EXPORT_CONF_FILE)
2865

    
2866
  config = objects.SerializableConfigParser()
2867
  config.read(cff)
2868

    
2869
  if (not config.has_section(constants.INISECT_EXP) or
2870
      not config.has_section(constants.INISECT_INS)):
2871
    _Fail("Export info file doesn't have the required fields")
2872

    
2873
  return config.Dumps()
2874

    
2875

    
2876
def ListExports():
2877
  """Return a list of exports currently available on this machine.
2878

2879
  @rtype: list
2880
  @return: list of the exports
2881

2882
  """
2883
  if os.path.isdir(pathutils.EXPORT_DIR):
2884
    return sorted(utils.ListVisibleFiles(pathutils.EXPORT_DIR))
2885
  else:
2886
    _Fail("No exports directory")
2887

    
2888

    
2889
def RemoveExport(export):
2890
  """Remove an existing export from the node.
2891

2892
  @type export: str
2893
  @param export: the name of the export to remove
2894
  @rtype: None
2895

2896
  """
2897
  target = utils.PathJoin(pathutils.EXPORT_DIR, export)
2898

    
2899
  try:
2900
    shutil.rmtree(target)
2901
  except EnvironmentError, err:
2902
    _Fail("Error while removing the export: %s", err, exc=True)
2903

    
2904

    
2905
def BlockdevRename(devlist):
2906
  """Rename a list of block devices.
2907

2908
  @type devlist: list of tuples
2909
  @param devlist: list of tuples of the form  (disk,
2910
      new_logical_id, new_physical_id); disk is an
2911
      L{objects.Disk} object describing the current disk,
2912
      and new logical_id/physical_id is the name we
2913
      rename it to
2914
  @rtype: boolean
2915
  @return: True if all renames succeeded, False otherwise
2916

2917
  """
2918
  msgs = []
2919
  result = True
2920
  for disk, unique_id in devlist:
2921
    dev = _RecursiveFindBD(disk)
2922
    if dev is None:
2923
      msgs.append("Can't find device %s in rename" % str(disk))
2924
      result = False
2925
      continue
2926
    try:
2927
      old_rpath = dev.dev_path
2928
      dev.Rename(unique_id)
2929
      new_rpath = dev.dev_path
2930
      if old_rpath != new_rpath:
2931
        DevCacheManager.RemoveCache(old_rpath)
2932
        # FIXME: we should add the new cache information here, like:
2933
        # DevCacheManager.UpdateCache(new_rpath, owner, ...)
2934
        # but we don't have the owner here - maybe parse from existing
2935
        # cache? for now, we only lose lvm data when we rename, which
2936
        # is less critical than DRBD or MD
2937
    except errors.BlockDeviceError, err:
2938
      msgs.append("Can't rename device '%s' to '%s': %s" %
2939
                  (dev, unique_id, err))
2940
      logging.exception("Can't rename device '%s' to '%s'", dev, unique_id)
2941
      result = False
2942
  if not result:
2943
    _Fail("; ".join(msgs))
2944

    
2945

    
2946
def _TransformFileStorageDir(fs_dir):
2947
  """Checks whether given file_storage_dir is valid.
2948

2949
  Checks wheter the given fs_dir is within the cluster-wide default
2950
  file_storage_dir or the shared_file_storage_dir, which are stored in
2951
  SimpleStore. Only paths under those directories are allowed.
2952

2953
  @type fs_dir: str
2954
  @param fs_dir: the path to check
2955

2956
  @return: the normalized path if valid, None otherwise
2957

2958
  """
2959
  if not (constants.ENABLE_FILE_STORAGE or
2960
          constants.ENABLE_SHARED_FILE_STORAGE):
2961
    _Fail("File storage disabled at configure time")
2962

    
2963
  bdev.CheckFileStoragePath(fs_dir)
2964

    
2965
  return os.path.normpath(fs_dir)
2966

    
2967

    
2968
def CreateFileStorageDir(file_storage_dir):
2969
  """Create file storage directory.
2970

2971
  @type file_storage_dir: str
2972
  @param file_storage_dir: directory to create
2973

2974
  @rtype: tuple
2975
  @return: tuple with first element a boolean indicating wheter dir
2976
      creation was successful or not
2977

2978
  """
2979
  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
2980
  if os.path.exists(file_storage_dir):
2981
    if not os.path.isdir(file_storage_dir):
2982
      _Fail("Specified storage dir '%s' is not a directory",
2983
            file_storage_dir)
2984
  else:
2985
    try:
2986
      os.makedirs(file_storage_dir, 0750)
2987
    except OSError, err:
2988
      _Fail("Cannot create file storage directory '%s': %s",
2989
            file_storage_dir, err, exc=True)
2990

    
2991

    
2992
def RemoveFileStorageDir(file_storage_dir):
2993
  """Remove file storage directory.
2994

2995
  Remove it only if it's empty. If not log an error and return.
2996

2997
  @type file_storage_dir: str
2998
  @param file_storage_dir: the directory we should cleanup
2999
  @rtype: tuple (success,)
3000
  @return: tuple of one element, C{success}, denoting
3001
      whether the operation was successful
3002

3003
  """
3004
  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
3005
  if os.path.exists(file_storage_dir):
3006
    if not os.path.isdir(file_storage_dir):
3007
      _Fail("Specified Storage directory '%s' is not a directory",
3008
            file_storage_dir)
3009
    # deletes dir only if empty, otherwise we want to fail the rpc call
3010
    try:
3011
      os.rmdir(file_storage_dir)
3012
    except OSError, err:
3013
      _Fail("Cannot remove file storage directory '%s': %s",
3014
            file_storage_dir, err)
3015

    
3016

    
3017
def RenameFileStorageDir(old_file_storage_dir, new_file_storage_dir):
3018
  """Rename the file storage directory.
3019

3020
  @type old_file_storage_dir: str
3021
  @param old_file_storage_dir: the current path
3022
  @type new_file_storage_dir: str
3023
  @param new_file_storage_dir: the name we should rename to
3024
  @rtype: tuple (success,)
3025
  @return: tuple of one element, C{success}, denoting
3026
      whether the operation was successful
3027

3028
  """
3029
  old_file_storage_dir = _TransformFileStorageDir(old_file_storage_dir)
3030
  new_file_storage_dir = _TransformFileStorageDir(new_file_storage_dir)
3031
  if not os.path.exists(new_file_storage_dir):
3032
    if os.path.isdir(old_file_storage_dir):
3033
      try:
3034
        os.rename(old_file_storage_dir, new_file_storage_dir)
3035
      except OSError, err:
3036
        _Fail("Cannot rename '%s' to '%s': %s",
3037
              old_file_storage_dir, new_file_storage_dir, err)
3038
    else:
3039
      _Fail("Specified storage dir '%s' is not a directory",
3040
            old_file_storage_dir)
3041
  else:
3042
    if os.path.exists(old_file_storage_dir):
3043
      _Fail("Cannot rename '%s' to '%s': both locations exist",
3044
            old_file_storage_dir, new_file_storage_dir)
3045

    
3046

    
3047
def _EnsureJobQueueFile(file_name):
3048
  """Checks whether the given filename is in the queue directory.
3049

3050
  @type file_name: str
3051
  @param file_name: the file name we should check
3052
  @rtype: None
3053
  @raises RPCFail: if the file is not valid
3054

3055
  """
3056
  if not utils.IsBelowDir(pathutils.QUEUE_DIR, file_name):
3057
    _Fail("Passed job queue file '%s' does not belong to"
3058
          " the queue directory '%s'", file_name, pathutils.QUEUE_DIR)
3059

    
3060

    
3061
def JobQueueUpdate(file_name, content):
3062
  """Updates a file in the queue directory.
3063

3064
  This is just a wrapper over L{utils.io.WriteFile}, with proper
3065
  checking.
3066

3067
  @type file_name: str
3068
  @param file_name: the job file name
3069
  @type content: str
3070
  @param content: the new job contents
3071
  @rtype: boolean
3072
  @return: the success of the operation
3073

3074
  """
3075
  file_name = vcluster.LocalizeVirtualPath(file_name)
3076

    
3077
  _EnsureJobQueueFile(file_name)
3078
  getents = runtime.GetEnts()
3079

    
3080
  # Write and replace the file atomically
3081
  utils.WriteFile(file_name, data=_Decompress(content), uid=getents.masterd_uid,
3082
                  gid=getents.daemons_gid, mode=constants.JOB_QUEUE_FILES_PERMS)
3083

    
3084

    
3085
def JobQueueRename(old, new):
3086
  """Renames a job queue file.
3087

3088
  This is just a wrapper over os.rename with proper checking.
3089

3090
  @type old: str
3091
  @param old: the old (actual) file name
3092
  @type new: str
3093
  @param new: the desired file name
3094
  @rtype: tuple
3095
  @return: the success of the operation and payload
3096

3097
  """
3098
  old = vcluster.LocalizeVirtualPath(old)
3099
  new = vcluster.LocalizeVirtualPath(new)
3100

    
3101
  _EnsureJobQueueFile(old)
3102
  _EnsureJobQueueFile(new)
3103

    
3104
  getents = runtime.GetEnts()
3105

    
3106
  utils.RenameFile(old, new, mkdir=True, mkdir_mode=0750,
3107
                   dir_uid=getents.masterd_uid, dir_gid=getents.daemons_gid)
3108

    
3109

    
3110
def BlockdevClose(instance_name, disks):
3111
  """Closes the given block devices.
3112

3113
  This means they will be switched to secondary mode (in case of
3114
  DRBD).
3115

3116
  @param instance_name: if the argument is not empty, the symlinks
3117
      of this instance will be removed
3118
  @type disks: list of L{objects.Disk}
3119
  @param disks: the list of disks to be closed
3120
  @rtype: tuple (success, message)
3121
  @return: a tuple of success and message, where success
3122
      indicates the succes of the operation, and message
3123
      which will contain the error details in case we
3124
      failed
3125

3126
  """
3127
  bdevs = []
3128
  for cf in disks:
3129
    rd = _RecursiveFindBD(cf)
3130
    if rd is None:
3131
      _Fail("Can't find device %s", cf)
3132
    bdevs.append(rd)
3133

    
3134
  msg = []
3135
  for rd in bdevs:
3136
    try:
3137
      rd.Close()
3138
    except errors.BlockDeviceError, err:
3139
      msg.append(str(err))
3140
  if msg:
3141
    _Fail("Can't make devices secondary: %s", ",".join(msg))
3142
  else:
3143
    if instance_name:
3144
      _RemoveBlockDevLinks(instance_name, disks)
3145

    
3146

    
3147
def ValidateHVParams(hvname, hvparams):
3148
  """Validates the given hypervisor parameters.
3149

3150
  @type hvname: string
3151
  @param hvname: the hypervisor name
3152
  @type hvparams: dict
3153
  @param hvparams: the hypervisor parameters to be validated
3154
  @rtype: None
3155

3156
  """
3157
  try:
3158
    hv_type = hypervisor.GetHypervisor(hvname)
3159
    hv_type.ValidateParameters(hvparams)
3160
  except errors.HypervisorError, err:
3161
    _Fail(str(err), log=False)
3162

    
3163

    
3164
def _CheckOSPList(os_obj, parameters):
3165
  """Check whether a list of parameters is supported by the OS.
3166

3167
  @type os_obj: L{objects.OS}
3168
  @param os_obj: OS object to check
3169
  @type parameters: list
3170
  @param parameters: the list of parameters to check
3171

3172
  """
3173
  supported = [v[0] for v in os_obj.supported_parameters]
3174
  delta = frozenset(parameters).difference(supported)
3175
  if delta:
3176
    _Fail("The following parameters are not supported"
3177
          " by the OS %s: %s" % (os_obj.name, utils.CommaJoin(delta)))
3178

    
3179

    
3180
def ValidateOS(required, osname, checks, osparams):
3181
  """Validate the given OS' parameters.
3182

3183
  @type required: boolean
3184
  @param required: whether absence of the OS should translate into
3185
      failure or not
3186
  @type osname: string
3187
  @param osname: the OS to be validated
3188
  @type checks: list
3189
  @param checks: list of the checks to run (currently only 'parameters')
3190
  @type osparams: dict
3191
  @param osparams: dictionary with OS parameters
3192
  @rtype: boolean
3193
  @return: True if the validation passed, or False if the OS was not
3194
      found and L{required} was false
3195

3196
  """
3197
  if not constants.OS_VALIDATE_CALLS.issuperset(checks):
3198
    _Fail("Unknown checks required for OS %s: %s", osname,
3199
          set(checks).difference(constants.OS_VALIDATE_CALLS))
3200

    
3201
  name_only = objects.OS.GetName(osname)
3202
  status, tbv = _TryOSFromDisk(name_only, None)
3203

    
3204
  if not status:
3205
    if required:
3206
      _Fail(tbv)
3207
    else:
3208
      return False
3209

    
3210
  if max(tbv.api_versions) < constants.OS_API_V20:
3211
    return True
3212

    
3213
  if constants.OS_VALIDATE_PARAMETERS in checks:
3214
    _CheckOSPList(tbv, osparams.keys())
3215

    
3216
  validate_env = OSCoreEnv(osname, tbv, osparams)
3217
  result = utils.RunCmd([tbv.verify_script] + checks, env=validate_env,
3218
                        cwd=tbv.path, reset_env=True)
3219
  if result.failed:
3220
    logging.error("os validate command '%s' returned error: %s output: %s",
3221
                  result.cmd, result.fail_reason, result.output)
3222
    _Fail("OS validation script failed (%s), output: %s",
3223
          result.fail_reason, result.output, log=False)
3224

    
3225
  return True
3226

    
3227

    
3228
def DemoteFromMC():
3229
  """Demotes the current node from master candidate role.
3230

3231
  """
3232
  # try to ensure we're not the master by mistake
3233
  master, myself = ssconf.GetMasterAndMyself()
3234
  if master == myself:
3235
    _Fail("ssconf status shows I'm the master node, will not demote")
3236

    
3237
  result = utils.RunCmd([pathutils.DAEMON_UTIL, "check", constants.MASTERD])
3238
  if not result.failed:
3239
    _Fail("The master daemon is running, will not demote")
3240

    
3241
  try:
3242
    if os.path.isfile(pathutils.CLUSTER_CONF_FILE):
3243
      utils.CreateBackup(pathutils.CLUSTER_CONF_FILE)
3244
  except EnvironmentError, err:
3245
    if err.errno != errno.ENOENT:
3246
      _Fail("Error while backing up cluster file: %s", err, exc=True)
3247

    
3248
  utils.RemoveFile(pathutils.CLUSTER_CONF_FILE)
3249

    
3250

    
3251
def _GetX509Filenames(cryptodir, name):
3252
  """Returns the full paths for the private key and certificate.
3253

3254
  """
3255
  return (utils.PathJoin(cryptodir, name),
3256
          utils.PathJoin(cryptodir, name, _X509_KEY_FILE),
3257
          utils.PathJoin(cryptodir, name, _X509_CERT_FILE))
3258

    
3259

    
3260
def CreateX509Certificate(validity, cryptodir=pathutils.CRYPTO_KEYS_DIR):
3261
  """Creates a new X509 certificate for SSL/TLS.
3262

3263
  @type validity: int
3264
  @param validity: Validity in seconds
3265
  @rtype: tuple; (string, string)
3266
  @return: Certificate name and public part
3267

3268
  """
3269
  (key_pem, cert_pem) = \
3270
    utils.GenerateSelfSignedX509Cert(netutils.Hostname.GetSysName(),
3271
                                     min(validity, _MAX_SSL_CERT_VALIDITY))
3272

    
3273
  cert_dir = tempfile.mkdtemp(dir=cryptodir,
3274
                              prefix="x509-%s-" % utils.TimestampForFilename())
3275
  try:
3276
    name = os.path.basename(cert_dir)
3277
    assert len(name) > 5
3278

    
3279
    (_, key_file, cert_file) = _GetX509Filenames(cryptodir, name)
3280

    
3281
    utils.WriteFile(key_file, mode=0400, data=key_pem)
3282
    utils.WriteFile(cert_file, mode=0400, data=cert_pem)
3283

    
3284
    # Never return private key as it shouldn't leave the node
3285
    return (name, cert_pem)
3286
  except Exception:
3287
    shutil.rmtree(cert_dir, ignore_errors=True)
3288
    raise
3289

    
3290

    
3291
def RemoveX509Certificate(name, cryptodir=pathutils.CRYPTO_KEYS_DIR):
3292
  """Removes a X509 certificate.
3293

3294
  @type name: string
3295
  @param name: Certificate name
3296

3297
  """
3298
  (cert_dir, key_file, cert_file) = _GetX509Filenames(cryptodir, name)
3299

    
3300
  utils.RemoveFile(key_file)
3301
  utils.RemoveFile(cert_file)
3302

    
3303
  try:
3304
    os.rmdir(cert_dir)
3305
  except EnvironmentError, err:
3306
    _Fail("Cannot remove certificate directory '%s': %s",
3307
          cert_dir, err)
3308

    
3309

    
3310
def _GetImportExportIoCommand(instance, mode, ieio, ieargs):
3311
  """Returns the command for the requested input/output.
3312

3313
  @type instance: L{objects.Instance}
3314
  @param instance: The instance object
3315
  @param mode: Import/export mode
3316
  @param ieio: Input/output type
3317
  @param ieargs: Input/output arguments
3318

3319
  """
3320
  assert mode in (constants.IEM_IMPORT, constants.IEM_EXPORT)
3321

    
3322
  env = None
3323
  prefix = None
3324
  suffix = None
3325
  exp_size = None
3326

    
3327
  if ieio == constants.IEIO_FILE:
3328
    (filename, ) = ieargs
3329

    
3330
    if not utils.IsNormAbsPath(filename):
3331
      _Fail("Path '%s' is not normalized or absolute", filename)
3332

    
3333
    real_filename = os.path.realpath(filename)
3334
    directory = os.path.dirname(real_filename)
3335

    
3336
    if not utils.IsBelowDir(pathutils.EXPORT_DIR, real_filename):
3337
      _Fail("File '%s' is not under exports directory '%s': %s",
3338
            filename, pathutils.EXPORT_DIR, real_filename)
3339

    
3340
    # Create directory
3341
    utils.Makedirs(directory, mode=0750)
3342

    
3343
    quoted_filename = utils.ShellQuote(filename)
3344

    
3345
    if mode == constants.IEM_IMPORT:
3346
      suffix = "> %s" % quoted_filename
3347
    elif mode == constants.IEM_EXPORT:
3348
      suffix = "< %s" % quoted_filename
3349

    
3350
      # Retrieve file size
3351
      try:
3352
        st = os.stat(filename)
3353
      except EnvironmentError, err:
3354
        logging.error("Can't stat(2) %s: %s", filename, err)
3355
      else:
3356
        exp_size = utils.BytesToMebibyte(st.st_size)
3357

    
3358
  elif ieio == constants.IEIO_RAW_DISK:
3359
    (disk, ) = ieargs
3360

    
3361
    real_disk = _OpenRealBD(disk)
3362

    
3363
    if mode == constants.IEM_IMPORT:
3364
      # we set here a smaller block size as, due to transport buffering, more
3365
      # than 64-128k will mostly ignored; we use nocreat to fail if the device
3366
      # is not already there or we pass a wrong path; we use notrunc to no
3367
      # attempt truncate on an LV device; we use oflag=dsync to not buffer too
3368
      # much memory; this means that at best, we flush every 64k, which will
3369
      # not be very fast
3370
      suffix = utils.BuildShellCmd(("| dd of=%s conv=nocreat,notrunc"
3371
                                    " bs=%s oflag=dsync"),
3372
                                    real_disk.dev_path,
3373
                                    str(64 * 1024))
3374

    
3375
    elif mode == constants.IEM_EXPORT:
3376
      # the block size on the read dd is 1MiB to match our units
3377
      prefix = utils.BuildShellCmd("dd if=%s bs=%s count=%s |",
3378
                                   real_disk.dev_path,
3379
                                   str(1024 * 1024), # 1 MB
3380
                                   str(disk.size))
3381
      exp_size = disk.size
3382

    
3383
  elif ieio == constants.IEIO_SCRIPT:
3384
    (disk, disk_index, ) = ieargs
3385

    
3386
    assert isinstance(disk_index, (int, long))
3387

    
3388
    real_disk = _OpenRealBD(disk)
3389

    
3390
    inst_os = OSFromDisk(instance.os)
3391
    env = OSEnvironment(instance, inst_os)
3392

    
3393
    if mode == constants.IEM_IMPORT:
3394
      env["IMPORT_DEVICE"] = env["DISK_%d_PATH" % disk_index]
3395
      env["IMPORT_INDEX"] = str(disk_index)
3396
      script = inst_os.import_script
3397

    
3398
    elif mode == constants.IEM_EXPORT:
3399
      env["EXPORT_DEVICE"] = real_disk.dev_path
3400
      env["EXPORT_INDEX"] = str(disk_index)
3401
      script = inst_os.export_script
3402

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

    
3406
    if mode == constants.IEM_IMPORT:
3407
      suffix = "| %s" % script_cmd
3408

    
3409
    elif mode == constants.IEM_EXPORT:
3410
      prefix = "%s |" % script_cmd
3411

    
3412
    # Let script predict size
3413
    exp_size = constants.IE_CUSTOM_SIZE
3414

    
3415
  else:
3416
    _Fail("Invalid %s I/O mode %r", mode, ieio)
3417

    
3418
  return (env, prefix, suffix, exp_size)
3419

    
3420

    
3421
def _CreateImportExportStatusDir(prefix):
3422
  """Creates status directory for import/export.
3423

3424
  """
3425
  return tempfile.mkdtemp(dir=pathutils.IMPORT_EXPORT_DIR,
3426
                          prefix=("%s-%s-" %
3427
                                  (prefix, utils.TimestampForFilename())))
3428

    
3429

    
3430
def StartImportExportDaemon(mode, opts, host, port, instance, component,
3431
                            ieio, ieioargs):
3432
  """Starts an import or export daemon.
3433

3434
  @param mode: Import/output mode
3435
  @type opts: L{objects.ImportExportOptions}
3436
  @param opts: Daemon options
3437
  @type host: string
3438
  @param host: Remote host for export (None for import)
3439
  @type port: int
3440
  @param port: Remote port for export (None for import)
3441
  @type instance: L{objects.Instance}
3442
  @param instance: Instance object
3443
  @type component: string
3444
  @param component: which part of the instance is transferred now,
3445
      e.g. 'disk/0'
3446
  @param ieio: Input/output type
3447
  @param ieioargs: Input/output arguments
3448

3449
  """
3450
  if mode == constants.IEM_IMPORT:
3451
    prefix = "import"
3452

    
3453
    if not (host is None and port is None):
3454
      _Fail("Can not specify host or port on import")
3455

    
3456
  elif mode == constants.IEM_EXPORT:
3457
    prefix = "export"
3458

    
3459
    if host is None or port is None:
3460
      _Fail("Host and port must be specified for an export")
3461

    
3462
  else:
3463
    _Fail("Invalid mode %r", mode)
3464

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

    
3468
  (cmd_env, cmd_prefix, cmd_suffix, exp_size) = \
3469
    _GetImportExportIoCommand(instance, mode, ieio, ieioargs)
3470

    
3471
  if opts.key_name is None:
3472
    # Use server.pem
3473
    key_path = pathutils.NODED_CERT_FILE
3474
    cert_path = pathutils.NODED_CERT_FILE
3475
    assert opts.ca_pem is None
3476
  else:
3477
    (_, key_path, cert_path) = _GetX509Filenames(pathutils.CRYPTO_KEYS_DIR,
3478
                                                 opts.key_name)
3479
    assert opts.ca_pem is not None
3480

    
3481
  for i in [key_path, cert_path]:
3482
    if not os.path.exists(i):
3483
      _Fail("File '%s' does not exist" % i)
3484

    
3485
  status_dir = _CreateImportExportStatusDir("%s-%s" % (prefix, component))
3486
  try:
3487
    status_file = utils.PathJoin(status_dir, _IES_STATUS_FILE)
3488
    pid_file = utils.PathJoin(status_dir, _IES_PID_FILE)
3489
    ca_file = utils.PathJoin(status_dir, _IES_CA_FILE)
3490

    
3491
    if opts.ca_pem is None:
3492
      # Use server.pem
3493
      ca = utils.ReadFile(pathutils.NODED_CERT_FILE)
3494
    else:
3495
      ca = opts.ca_pem
3496

    
3497
    # Write CA file
3498
    utils.WriteFile(ca_file, data=ca, mode=0400)
3499

    
3500
    cmd = [
3501
      pathutils.IMPORT_EXPORT_DAEMON,
3502
      status_file, mode,
3503
      "--key=%s" % key_path,
3504
      "--cert=%s" % cert_path,
3505
      "--ca=%s" % ca_file,
3506
      ]
3507

    
3508
    if host:
3509
      cmd.append("--host=%s" % host)
3510

    
3511
    if port:
3512
      cmd.append("--port=%s" % port)
3513

    
3514
    if opts.ipv6:
3515
      cmd.append("--ipv6")
3516
    else:
3517
      cmd.append("--ipv4")
3518

    
3519
    if opts.compress:
3520
      cmd.append("--compress=%s" % opts.compress)
3521

    
3522
    if opts.magic:
3523
      cmd.append("--magic=%s" % opts.magic)
3524

    
3525
    if exp_size is not None:
3526
      cmd.append("--expected-size=%s" % exp_size)
3527

    
3528
    if cmd_prefix:
3529
      cmd.append("--cmd-prefix=%s" % cmd_prefix)
3530

    
3531
    if cmd_suffix:
3532
      cmd.append("--cmd-suffix=%s" % cmd_suffix)
3533

    
3534
    if mode == constants.IEM_EXPORT:
3535
      # Retry connection a few times when connecting to remote peer
3536
      cmd.append("--connect-retries=%s" % constants.RIE_CONNECT_RETRIES)
3537
      cmd.append("--connect-timeout=%s" % constants.RIE_CONNECT_ATTEMPT_TIMEOUT)
3538
    elif opts.connect_timeout is not None:
3539
      assert mode == constants.IEM_IMPORT
3540
      # Overall timeout for establishing connection while listening
3541
      cmd.append("--connect-timeout=%s" % opts.connect_timeout)
3542

    
3543
    logfile = _InstanceLogName(prefix, instance.os, instance.name, component)
3544

    
3545
    # TODO: Once _InstanceLogName uses tempfile.mkstemp, StartDaemon has
3546
    # support for receiving a file descriptor for output
3547
    utils.StartDaemon(cmd, env=cmd_env, pidfile=pid_file,
3548
                      output=logfile)
3549

    
3550
    # The import/export name is simply the status directory name
3551
    return os.path.basename(status_dir)
3552

    
3553
  except Exception:
3554
    shutil.rmtree(status_dir, ignore_errors=True)
3555
    raise
3556

    
3557

    
3558
def GetImportExportStatus(names):
3559
  """Returns import/export daemon status.
3560

3561
  @type names: sequence
3562
  @param names: List of names
3563
  @rtype: List of dicts
3564
  @return: Returns a list of the state of each named import/export or None if a
3565
           status couldn't be read
3566

3567
  """
3568
  result = []
3569

    
3570
  for name in names:
3571
    status_file = utils.PathJoin(pathutils.IMPORT_EXPORT_DIR, name,
3572
                                 _IES_STATUS_FILE)
3573

    
3574
    try:
3575
      data = utils.ReadFile(status_file)
3576
    except EnvironmentError, err:
3577
      if err.errno != errno.ENOENT:
3578
        raise
3579
      data = None
3580

    
3581
    if not data:
3582
      result.append(None)
3583
      continue
3584

    
3585
    result.append(serializer.LoadJson(data))
3586

    
3587
  return result
3588

    
3589

    
3590
def AbortImportExport(name):
3591
  """Sends SIGTERM to a running import/export daemon.
3592

3593
  """
3594
  logging.info("Abort import/export %s", name)
3595

    
3596
  status_dir = utils.PathJoin(pathutils.IMPORT_EXPORT_DIR, name)
3597
  pid = utils.ReadLockedPidFile(utils.PathJoin(status_dir, _IES_PID_FILE))
3598

    
3599
  if pid:
3600
    logging.info("Import/export %s is running with PID %s, sending SIGTERM",
3601
                 name, pid)
3602
    utils.IgnoreProcessNotFound(os.kill, pid, signal.SIGTERM)
3603

    
3604

    
3605
def CleanupImportExport(name):
3606
  """Cleanup after an import or export.
3607

3608
  If the import/export daemon is still running it's killed. Afterwards the
3609
  whole status directory is removed.
3610

3611
  """
3612
  logging.info("Finalizing import/export %s", name)
3613

    
3614
  status_dir = utils.PathJoin(pathutils.IMPORT_EXPORT_DIR, name)
3615

    
3616
  pid = utils.ReadLockedPidFile(utils.PathJoin(status_dir, _IES_PID_FILE))
3617

    
3618
  if pid:
3619
    logging.info("Import/export %s is still running with PID %s",
3620
                 name, pid)
3621
    utils.KillProcess(pid, waitpid=False)
3622

    
3623
  shutil.rmtree(status_dir, ignore_errors=True)
3624

    
3625

    
3626
def _FindDisks(nodes_ip, disks):
3627
  """Sets the physical ID on disks and returns the block devices.
3628

3629
  """
3630
  # set the correct physical ID
3631
  my_name = netutils.Hostname.GetSysName()
3632
  for cf in disks:
3633
    cf.SetPhysicalID(my_name, nodes_ip)
3634

    
3635
  bdevs = []
3636

    
3637
  for cf in disks:
3638
    rd = _RecursiveFindBD(cf)
3639
    if rd is None:
3640
      _Fail("Can't find device %s", cf)
3641
    bdevs.append(rd)
3642
  return bdevs
3643

    
3644

    
3645
def DrbdDisconnectNet(nodes_ip, disks):
3646
  """Disconnects the network on a list of drbd devices.
3647

3648
  """
3649
  bdevs = _FindDisks(nodes_ip, disks)
3650

    
3651
  # disconnect disks
3652
  for rd in bdevs:
3653
    try:
3654
      rd.DisconnectNet()
3655
    except errors.BlockDeviceError, err:
3656
      _Fail("Can't change network configuration to standalone mode: %s",
3657
            err, exc=True)
3658

    
3659

    
3660
def DrbdAttachNet(nodes_ip, disks, instance_name, multimaster):
3661
  """Attaches the network on a list of drbd devices.
3662

3663
  """
3664
  bdevs = _FindDisks(nodes_ip, disks)
3665

    
3666
  if multimaster:
3667
    for idx, rd in enumerate(bdevs):
3668
      try:
3669
        _SymlinkBlockDev(instance_name, rd.dev_path, idx)
3670
      except EnvironmentError, err:
3671
        _Fail("Can't create symlink: %s", err)
3672
  # reconnect disks, switch to new master configuration and if
3673
  # needed primary mode
3674
  for rd in bdevs:
3675
    try:
3676
      rd.AttachNet(multimaster)
3677
    except errors.BlockDeviceError, err:
3678
      _Fail("Can't change network configuration: %s", err)
3679

    
3680
  # wait until the disks are connected; we need to retry the re-attach
3681
  # if the device becomes standalone, as this might happen if the one
3682
  # node disconnects and reconnects in a different mode before the
3683
  # other node reconnects; in this case, one or both of the nodes will
3684
  # decide it has wrong configuration and switch to standalone
3685

    
3686
  def _Attach():
3687
    all_connected = True
3688

    
3689
    for rd in bdevs:
3690
      stats = rd.GetProcStatus()
3691

    
3692
      if multimaster:
3693
        # In the multimaster case we have to wait explicitly until
3694
        # the resource is Connected and UpToDate/UpToDate, because
3695
        # we promote *both nodes* to primary directly afterwards.
3696
        # Being in resync is not enough, since there is a race during which we
3697
        # may promote a node with an Outdated disk to primary, effectively
3698
        # tearing down the connection.
3699
        all_connected = (all_connected and
3700
                         stats.is_connected and
3701
                         stats.is_disk_uptodate and
3702
                         stats.peer_disk_uptodate)
3703
      else:
3704
        all_connected = (all_connected and
3705
                         (stats.is_connected or stats.is_in_resync))
3706

    
3707
      if stats.is_standalone:
3708
        # peer had different config info and this node became
3709
        # standalone, even though this should not happen with the
3710
        # new staged way of changing disk configs
3711
        try:
3712
          rd.AttachNet(multimaster)
3713
        except errors.BlockDeviceError, err:
3714
          _Fail("Can't change network configuration: %s", err)
3715

    
3716
    if not all_connected:
3717
      raise utils.RetryAgain()
3718

    
3719
  try:
3720
    # Start with a delay of 100 miliseconds and go up to 5 seconds
3721
    utils.Retry(_Attach, (0.1, 1.5, 5.0), 2 * 60)
3722
  except utils.RetryTimeout:
3723
    _Fail("Timeout in disk reconnecting")
3724

    
3725
  if multimaster:
3726
    # change to primary mode
3727
    for rd in bdevs:
3728
      try:
3729
        rd.Open()
3730
      except errors.BlockDeviceError, err:
3731
        _Fail("Can't change to primary mode: %s", err)
3732

    
3733

    
3734
def DrbdWaitSync(nodes_ip, disks):
3735
  """Wait until DRBDs have synchronized.
3736

3737
  """
3738
  def _helper(rd):
3739
    stats = rd.GetProcStatus()
3740
    if not (stats.is_connected or stats.is_in_resync):
3741
      raise utils.RetryAgain()
3742
    return stats
3743

    
3744
  bdevs = _FindDisks(nodes_ip, disks)
3745

    
3746
  min_resync = 100
3747
  alldone = True
3748
  for rd in bdevs:
3749
    try:
3750
      # poll each second for 15 seconds
3751
      stats = utils.Retry(_helper, 1, 15, args=[rd])
3752
    except utils.RetryTimeout:
3753
      stats = rd.GetProcStatus()
3754
      # last check
3755
      if not (stats.is_connected or stats.is_in_resync):
3756
        _Fail("DRBD device %s is not in sync: stats=%s", rd, stats)
3757
    alldone = alldone and (not stats.is_in_resync)
3758
    if stats.sync_percent is not None:
3759
      min_resync = min(min_resync, stats.sync_percent)
3760

    
3761
  return (alldone, min_resync)
3762

    
3763

    
3764
def GetDrbdUsermodeHelper():
3765
  """Returns DRBD usermode helper currently configured.
3766

3767
  """
3768
  try:
3769
    return bdev.BaseDRBD.GetUsermodeHelper()
3770
  except errors.BlockDeviceError, err:
3771
    _Fail(str(err))
3772

    
3773

    
3774
def PowercycleNode(hypervisor_type):
3775
  """Hard-powercycle the node.
3776

3777
  Because we need to return first, and schedule the powercycle in the
3778
  background, we won't be able to report failures nicely.
3779

3780
  """
3781
  hyper = hypervisor.GetHypervisor(hypervisor_type)
3782
  try:
3783
    pid = os.fork()
3784
  except OSError:
3785
    # if we can't fork, we'll pretend that we're in the child process
3786
    pid = 0
3787
  if pid > 0:
3788
    return "Reboot scheduled in 5 seconds"
3789
  # ensure the child is running on ram
3790
  try:
3791
    utils.Mlockall()
3792
  except Exception: # pylint: disable=W0703
3793
    pass
3794
  time.sleep(5)
3795
  hyper.PowercycleNode()
3796

    
3797

    
3798
def _VerifyRestrictedCmdName(cmd):
3799
  """Verifies a restricted command name.
3800

3801
  @type cmd: string
3802
  @param cmd: Command name
3803
  @rtype: tuple; (boolean, string or None)
3804
  @return: The tuple's first element is the status; if C{False}, the second
3805
    element is an error message string, otherwise it's C{None}
3806

3807
  """
3808
  if not cmd.strip():
3809
    return (False, "Missing command name")
3810

    
3811
  if os.path.basename(cmd) != cmd:
3812
    return (False, "Invalid command name")
3813

    
3814
  if not constants.EXT_PLUGIN_MASK.match(cmd):
3815
    return (False, "Command name contains forbidden characters")
3816

    
3817
  return (True, None)
3818

    
3819

    
3820
def _CommonRestrictedCmdCheck(path, owner):
3821
  """Common checks for restricted command file system directories and files.
3822

3823
  @type path: string
3824
  @param path: Path to check
3825
  @param owner: C{None} or tuple containing UID and GID
3826
  @rtype: tuple; (boolean, string or C{os.stat} result)
3827
  @return: The tuple's first element is the status; if C{False}, the second
3828
    element is an error message string, otherwise it's the result of C{os.stat}
3829

3830
  """
3831
  if owner is None:
3832
    # Default to root as owner
3833
    owner = (0, 0)
3834

    
3835
  try:
3836
    st = os.stat(path)
3837
  except EnvironmentError, err:
3838
    return (False, "Can't stat(2) '%s': %s" % (path, err))
3839

    
3840
  if stat.S_IMODE(st.st_mode) & (~_RCMD_MAX_MODE):
3841
    return (False, "Permissions on '%s' are too permissive" % path)
3842

    
3843
  if (st.st_uid, st.st_gid) != owner:
3844
    (owner_uid, owner_gid) = owner
3845
    return (False, "'%s' is not owned by %s:%s" % (path, owner_uid, owner_gid))
3846

    
3847
  return (True, st)
3848

    
3849

    
3850
def _VerifyRestrictedCmdDirectory(path, _owner=None):
3851
  """Verifies restricted command directory.
3852

3853
  @type path: string
3854
  @param path: Path to check
3855
  @rtype: tuple; (boolean, string or None)
3856
  @return: The tuple's first element is the status; if C{False}, the second
3857
    element is an error message string, otherwise it's C{None}
3858

3859
  """
3860
  (status, value) = _CommonRestrictedCmdCheck(path, _owner)
3861

    
3862
  if not status:
3863
    return (False, value)
3864

    
3865
  if not stat.S_ISDIR(value.st_mode):
3866
    return (False, "Path '%s' is not a directory" % path)
3867

    
3868
  return (True, None)
3869

    
3870

    
3871
def _VerifyRestrictedCmd(path, cmd, _owner=None):
3872
  """Verifies a whole restricted command and returns its executable filename.
3873

3874
  @type path: string
3875
  @param path: Directory containing restricted commands
3876
  @type cmd: string
3877
  @param cmd: Command name
3878
  @rtype: tuple; (boolean, string)
3879
  @return: The tuple's first element is the status; if C{False}, the second
3880
    element is an error message string, otherwise the second element is the
3881
    absolute path to the executable
3882

3883
  """
3884
  executable = utils.PathJoin(path, cmd)
3885

    
3886
  (status, msg) = _CommonRestrictedCmdCheck(executable, _owner)
3887

    
3888
  if not status:
3889
    return (False, msg)
3890

    
3891
  if not utils.IsExecutable(executable):
3892
    return (False, "access(2) thinks '%s' can't be executed" % executable)
3893

    
3894
  return (True, executable)
3895

    
3896

    
3897
def _PrepareRestrictedCmd(path, cmd,
3898
                          _verify_dir=_VerifyRestrictedCmdDirectory,
3899
                          _verify_name=_VerifyRestrictedCmdName,
3900
                          _verify_cmd=_VerifyRestrictedCmd):
3901
  """Performs a number of tests on a restricted command.
3902

3903
  @type path: string
3904
  @param path: Directory containing restricted commands
3905
  @type cmd: string
3906
  @param cmd: Command name
3907
  @return: Same as L{_VerifyRestrictedCmd}
3908

3909
  """
3910
  # Verify the directory first
3911
  (status, msg) = _verify_dir(path)
3912
  if status:
3913
    # Check command if everything was alright
3914
    (status, msg) = _verify_name(cmd)
3915

    
3916
  if not status:
3917
    return (False, msg)
3918

    
3919
  # Check actual executable
3920
  return _verify_cmd(path, cmd)
3921

    
3922

    
3923
def RunRestrictedCmd(cmd,
3924
                     _lock_timeout=_RCMD_LOCK_TIMEOUT,
3925
                     _lock_file=pathutils.RESTRICTED_COMMANDS_LOCK_FILE,
3926
                     _path=pathutils.RESTRICTED_COMMANDS_DIR,
3927
                     _sleep_fn=time.sleep,
3928
                     _prepare_fn=_PrepareRestrictedCmd,
3929
                     _runcmd_fn=utils.RunCmd,
3930
                     _enabled=constants.ENABLE_RESTRICTED_COMMANDS):
3931
  """Executes a restricted command after performing strict tests.
3932

3933
  @type cmd: string
3934
  @param cmd: Command name
3935
  @rtype: string
3936
  @return: Command output
3937
  @raise RPCFail: In case of an error
3938

3939
  """
3940
  logging.info("Preparing to run restricted command '%s'", cmd)
3941

    
3942
  if not _enabled:
3943
    _Fail("Restricted commands disabled at configure time")
3944

    
3945
  lock = None
3946
  try:
3947
    cmdresult = None
3948
    try:
3949
      lock = utils.FileLock.Open(_lock_file)
3950
      lock.Exclusive(blocking=True, timeout=_lock_timeout)
3951

    
3952
      (status, value) = _prepare_fn(_path, cmd)
3953

    
3954
      if status:
3955
        cmdresult = _runcmd_fn([value], env={}, reset_env=True,
3956
                               postfork_fn=lambda _: lock.Unlock())
3957
      else:
3958
        logging.error(value)
3959
    except Exception: # pylint: disable=W0703
3960
      # Keep original error in log
3961
      logging.exception("Caught exception")
3962

    
3963
    if cmdresult is None:
3964
      logging.info("Sleeping for %0.1f seconds before returning",
3965
                   _RCMD_INVALID_DELAY)
3966
      _sleep_fn(_RCMD_INVALID_DELAY)
3967

    
3968
      # Do not include original error message in returned error
3969
      _Fail("Executing command '%s' failed" % cmd)
3970
    elif cmdresult.failed or cmdresult.fail_reason:
3971
      _Fail("Restricted command '%s' failed: %s; output: %s",
3972
            cmd, cmdresult.fail_reason, cmdresult.output)
3973
    else:
3974
      return cmdresult.output
3975
  finally:
3976
    if lock is not None:
3977
      # Release lock at last
3978
      lock.Close()
3979
      lock = None
3980

    
3981

    
3982
def SetWatcherPause(until, _filename=pathutils.WATCHER_PAUSEFILE):
3983
  """Creates or removes the watcher pause file.
3984

3985
  @type until: None or number
3986
  @param until: Unix timestamp saying until when the watcher shouldn't run
3987

3988
  """
3989
  if until is None:
3990
    logging.info("Received request to no longer pause watcher")
3991
    utils.RemoveFile(_filename)
3992
  else:
3993
    logging.info("Received request to pause watcher until %s", until)
3994

    
3995
    if not ht.TNumber(until):
3996
      _Fail("Duration must be numeric")
3997

    
3998
    utils.WriteFile(_filename, data="%d\n" % (until, ), mode=0644)
3999

    
4000

    
4001
class HooksRunner(object):
4002
  """Hook runner.
4003

4004
  This class is instantiated on the node side (ganeti-noded) and not
4005
  on the master side.
4006

4007
  """
4008
  def __init__(self, hooks_base_dir=None):
4009
    """Constructor for hooks runner.
4010

4011
    @type hooks_base_dir: str or None
4012
    @param hooks_base_dir: if not None, this overrides the
4013
        L{pathutils.HOOKS_BASE_DIR} (useful for unittests)
4014

4015
    """
4016
    if hooks_base_dir is None:
4017
      hooks_base_dir = pathutils.HOOKS_BASE_DIR
4018
    # yeah, _BASE_DIR is not valid for attributes, we use it like a
4019
    # constant
4020
    self._BASE_DIR = hooks_base_dir # pylint: disable=C0103
4021

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

4025
    """
4026
    assert len(node_list) == 1
4027
    node = node_list[0]
4028
    _, myself = ssconf.GetMasterAndMyself()
4029
    assert node == myself
4030

    
4031
    results = self.RunHooks(hpath, phase, env)
4032

    
4033
    # Return values in the form expected by HooksMaster
4034
    return {node: (None, False, results)}
4035

    
4036
  def RunHooks(self, hpath, phase, env):
4037
    """Run the scripts in the hooks directory.
4038

4039
    @type hpath: str
4040
    @param hpath: the path to the hooks directory which
4041
        holds the scripts
4042
    @type phase: str
4043
    @param phase: either L{constants.HOOKS_PHASE_PRE} or
4044
        L{constants.HOOKS_PHASE_POST}
4045
    @type env: dict
4046
    @param env: dictionary with the environment for the hook
4047
    @rtype: list
4048
    @return: list of 3-element tuples:
4049
      - script path
4050
      - script result, either L{constants.HKR_SUCCESS} or
4051
        L{constants.HKR_FAIL}
4052
      - output of the script
4053

4054
    @raise errors.ProgrammerError: for invalid input
4055
        parameters
4056

4057
    """
4058
    if phase == constants.HOOKS_PHASE_PRE:
4059
      suffix = "pre"
4060
    elif phase == constants.HOOKS_PHASE_POST:
4061
      suffix = "post"
4062
    else:
4063
      _Fail("Unknown hooks phase '%s'", phase)
4064

    
4065
    subdir = "%s-%s.d" % (hpath, suffix)
4066
    dir_name = utils.PathJoin(self._BASE_DIR, subdir)
4067

    
4068
    results = []
4069

    
4070
    if not os.path.isdir(dir_name):
4071
      # for non-existing/non-dirs, we simply exit instead of logging a
4072
      # warning at every operation
4073
      return results
4074

    
4075
    runparts_results = utils.RunParts(dir_name, env=env, reset_env=True)
4076

    
4077
    for (relname, relstatus, runresult) in runparts_results:
4078
      if relstatus == constants.RUNPARTS_SKIP:
4079
        rrval = constants.HKR_SKIP
4080
        output = ""
4081
      elif relstatus == constants.RUNPARTS_ERR:
4082
        rrval = constants.HKR_FAIL
4083
        output = "Hook script execution error: %s" % runresult
4084
      elif relstatus == constants.RUNPARTS_RUN:
4085
        if runresult.failed:
4086
          rrval = constants.HKR_FAIL
4087
        else:
4088
          rrval = constants.HKR_SUCCESS
4089
        output = utils.SafeEncode(runresult.output.strip())
4090
      results.append(("%s/%s" % (subdir, relname), rrval, output))
4091

    
4092
    return results
4093

    
4094

    
4095
class IAllocatorRunner(object):
4096
  """IAllocator runner.
4097

4098
  This class is instantiated on the node side (ganeti-noded) and not on
4099
  the master side.
4100

4101
  """
4102
  @staticmethod
4103
  def Run(name, idata):
4104
    """Run an iallocator script.
4105

4106
    @type name: str
4107
    @param name: the iallocator script name
4108
    @type idata: str
4109
    @param idata: the allocator input data
4110

4111
    @rtype: tuple
4112
    @return: two element tuple of:
4113
       - status
4114
       - either error message or stdout of allocator (for success)
4115

4116
    """
4117
    alloc_script = utils.FindFile(name, constants.IALLOCATOR_SEARCH_PATH,
4118
                                  os.path.isfile)
4119
    if alloc_script is None:
4120
      _Fail("iallocator module '%s' not found in the search path", name)
4121

    
4122
    fd, fin_name = tempfile.mkstemp(prefix="ganeti-iallocator.")
4123
    try:
4124
      os.write(fd, idata)
4125
      os.close(fd)
4126
      result = utils.RunCmd([alloc_script, fin_name])
4127
      if result.failed:
4128
        _Fail("iallocator module '%s' failed: %s, output '%s'",
4129
              name, result.fail_reason, result.output)
4130
    finally:
4131
      os.unlink(fin_name)
4132

    
4133
    return result.stdout
4134

    
4135

    
4136
class DevCacheManager(object):
4137
  """Simple class for managing a cache of block device information.
4138

4139
  """
4140
  _DEV_PREFIX = "/dev/"
4141
  _ROOT_DIR = pathutils.BDEV_CACHE_DIR
4142

    
4143
  @classmethod
4144
  def _ConvertPath(cls, dev_path):
4145
    """Converts a /dev/name path to the cache file name.
4146

4147
    This replaces slashes with underscores and strips the /dev
4148