Statistics
| Branch: | Tag: | Revision:

root / lib / backend.py @ 2209788e

History | View | Annotate | Download (127.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
  if action == constants.HOTPLUG_ACTION_ADD:
1650
    fn = hyper.HotAddDevice
1651
  elif action == constants.HOTPLUG_ACTION_REMOVE:
1652
    fn = hyper.HotDelDevice
1653
  elif action == constants.HOTPLUG_ACTION_MODIFY:
1654
    fn = hyper.HotModDevice
1655
  else:
1656
    assert action in constants.HOTPLUG_ALL_ACTIONS
1657
  # This will raise an exception if hotplug is no supported for this case
1658
  hyper.HotplugSupported(instance, action, dev_type)
1659
  return fn(instance, dev_type, device, extra, seq)
1660

    
1661

    
1662
def BlockdevCreate(disk, size, owner, on_primary, info, excl_stor):
1663
  """Creates a block device for an instance.
1664

1665
  @type disk: L{objects.Disk}
1666
  @param disk: the object describing the disk we should create
1667
  @type size: int
1668
  @param size: the size of the physical underlying device, in MiB
1669
  @type owner: str
1670
  @param owner: the name of the instance for which disk is created,
1671
      used for device cache data
1672
  @type on_primary: boolean
1673
  @param on_primary:  indicates if it is the primary node or not
1674
  @type info: string
1675
  @param info: string that will be sent to the physical device
1676
      creation, used for example to set (LVM) tags on LVs
1677
  @type excl_stor: boolean
1678
  @param excl_stor: Whether exclusive_storage is active
1679

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

1684
  """
1685
  # TODO: remove the obsolete "size" argument
1686
  # pylint: disable=W0613
1687
  clist = []
1688
  if disk.children:
1689
    for child in disk.children:
1690
      try:
1691
        crdev = _RecursiveAssembleBD(child, owner, on_primary)
1692
      except errors.BlockDeviceError, err:
1693
        _Fail("Can't assemble device %s: %s", child, err)
1694
      if on_primary or disk.AssembleOnSecondary():
1695
        # we need the children open in case the device itself has to
1696
        # be assembled
1697
        try:
1698
          # pylint: disable=E1103
1699
          crdev.Open()
1700
        except errors.BlockDeviceError, err:
1701
          _Fail("Can't make child '%s' read-write: %s", child, err)
1702
      clist.append(crdev)
1703

    
1704
  try:
1705
    device = bdev.Create(disk, clist, excl_stor)
1706
  except errors.BlockDeviceError, err:
1707
    _Fail("Can't create block device: %s", err)
1708

    
1709
  if on_primary or disk.AssembleOnSecondary():
1710
    try:
1711
      device.Assemble()
1712
    except errors.BlockDeviceError, err:
1713
      _Fail("Can't assemble device after creation, unusual event: %s", err)
1714
    if on_primary or disk.OpenOnSecondary():
1715
      try:
1716
        device.Open(force=True)
1717
      except errors.BlockDeviceError, err:
1718
        _Fail("Can't make device r/w after creation, unusual event: %s", err)
1719
    DevCacheManager.UpdateCache(device.dev_path, owner,
1720
                                on_primary, disk.iv_name)
1721

    
1722
  device.SetInfo(info)
1723

    
1724
  return device.unique_id
1725

    
1726

    
1727
def _WipeDevice(path, offset, size):
1728
  """This function actually wipes the device.
1729

1730
  @param path: The path to the device to wipe
1731
  @param offset: The offset in MiB in the file
1732
  @param size: The size in MiB to write
1733

1734
  """
1735
  # Internal sizes are always in Mebibytes; if the following "dd" command
1736
  # should use a different block size the offset and size given to this
1737
  # function must be adjusted accordingly before being passed to "dd".
1738
  block_size = 1024 * 1024
1739

    
1740
  cmd = [constants.DD_CMD, "if=/dev/zero", "seek=%d" % offset,
1741
         "bs=%s" % block_size, "oflag=direct", "of=%s" % path,
1742
         "count=%d" % size]
1743
  result = utils.RunCmd(cmd)
1744

    
1745
  if result.failed:
1746
    _Fail("Wipe command '%s' exited with error: %s; output: %s", result.cmd,
1747
          result.fail_reason, result.output)
1748

    
1749

    
1750
def BlockdevWipe(disk, offset, size):
1751
  """Wipes a block device.
1752

1753
  @type disk: L{objects.Disk}
1754
  @param disk: the disk object we want to wipe
1755
  @type offset: int
1756
  @param offset: The offset in MiB in the file
1757
  @type size: int
1758
  @param size: The size in MiB to write
1759

1760
  """
1761
  try:
1762
    rdev = _RecursiveFindBD(disk)
1763
  except errors.BlockDeviceError:
1764
    rdev = None
1765

    
1766
  if not rdev:
1767
    _Fail("Cannot execute wipe for device %s: device not found", disk.iv_name)
1768

    
1769
  # Do cross verify some of the parameters
1770
  if offset < 0:
1771
    _Fail("Negative offset")
1772
  if size < 0:
1773
    _Fail("Negative size")
1774
  if offset > rdev.size:
1775
    _Fail("Offset is bigger than device size")
1776
  if (offset + size) > rdev.size:
1777
    _Fail("The provided offset and size to wipe is bigger than device size")
1778

    
1779
  _WipeDevice(rdev.dev_path, offset, size)
1780

    
1781

    
1782
def BlockdevPauseResumeSync(disks, pause):
1783
  """Pause or resume the sync of the block device.
1784

1785
  @type disks: list of L{objects.Disk}
1786
  @param disks: the disks object we want to pause/resume
1787
  @type pause: bool
1788
  @param pause: Wheater to pause or resume
1789

1790
  """
1791
  success = []
1792
  for disk in disks:
1793
    try:
1794
      rdev = _RecursiveFindBD(disk)
1795
    except errors.BlockDeviceError:
1796
      rdev = None
1797

    
1798
    if not rdev:
1799
      success.append((False, ("Cannot change sync for device %s:"
1800
                              " device not found" % disk.iv_name)))
1801
      continue
1802

    
1803
    result = rdev.PauseResumeSync(pause)
1804

    
1805
    if result:
1806
      success.append((result, None))
1807
    else:
1808
      if pause:
1809
        msg = "Pause"
1810
      else:
1811
        msg = "Resume"
1812
      success.append((result, "%s for device %s failed" % (msg, disk.iv_name)))
1813

    
1814
  return success
1815

    
1816

    
1817
def BlockdevRemove(disk):
1818
  """Remove a block device.
1819

1820
  @note: This is intended to be called recursively.
1821

1822
  @type disk: L{objects.Disk}
1823
  @param disk: the disk object we should remove
1824
  @rtype: boolean
1825
  @return: the success of the operation
1826

1827
  """
1828
  msgs = []
1829
  try:
1830
    rdev = _RecursiveFindBD(disk)
1831
  except errors.BlockDeviceError, err:
1832
    # probably can't attach
1833
    logging.info("Can't attach to device %s in remove", disk)
1834
    rdev = None
1835
  if rdev is not None:
1836
    r_path = rdev.dev_path
1837
    try:
1838
      rdev.Remove()
1839
    except errors.BlockDeviceError, err:
1840
      msgs.append(str(err))
1841
    if not msgs:
1842
      DevCacheManager.RemoveCache(r_path)
1843

    
1844
  if disk.children:
1845
    for child in disk.children:
1846
      try:
1847
        BlockdevRemove(child)
1848
      except RPCFail, err:
1849
        msgs.append(str(err))
1850

    
1851
  if msgs:
1852
    _Fail("; ".join(msgs))
1853

    
1854

    
1855
def _RecursiveAssembleBD(disk, owner, as_primary):
1856
  """Activate a block device for an instance.
1857

1858
  This is run on the primary and secondary nodes for an instance.
1859

1860
  @note: this function is called recursively.
1861

1862
  @type disk: L{objects.Disk}
1863
  @param disk: the disk we try to assemble
1864
  @type owner: str
1865
  @param owner: the name of the instance which owns the disk
1866
  @type as_primary: boolean
1867
  @param as_primary: if we should make the block device
1868
      read/write
1869

1870
  @return: the assembled device or None (in case no device
1871
      was assembled)
1872
  @raise errors.BlockDeviceError: in case there is an error
1873
      during the activation of the children or the device
1874
      itself
1875

1876
  """
1877
  children = []
1878
  if disk.children:
1879
    mcn = disk.ChildrenNeeded()
1880
    if mcn == -1:
1881
      mcn = 0 # max number of Nones allowed
1882
    else:
1883
      mcn = len(disk.children) - mcn # max number of Nones
1884
    for chld_disk in disk.children:
1885
      try:
1886
        cdev = _RecursiveAssembleBD(chld_disk, owner, as_primary)
1887
      except errors.BlockDeviceError, err:
1888
        if children.count(None) >= mcn:
1889
          raise
1890
        cdev = None
1891
        logging.error("Error in child activation (but continuing): %s",
1892
                      str(err))
1893
      children.append(cdev)
1894

    
1895
  if as_primary or disk.AssembleOnSecondary():
1896
    r_dev = bdev.Assemble(disk, children)
1897
    result = r_dev
1898
    if as_primary or disk.OpenOnSecondary():
1899
      r_dev.Open()
1900
    DevCacheManager.UpdateCache(r_dev.dev_path, owner,
1901
                                as_primary, disk.iv_name)
1902

    
1903
  else:
1904
    result = True
1905
  return result
1906

    
1907

    
1908
def BlockdevAssemble(disk, owner, as_primary, idx):
1909
  """Activate a block device for an instance.
1910

1911
  This is a wrapper over _RecursiveAssembleBD.
1912

1913
  @rtype: str or boolean
1914
  @return: a C{/dev/...} path for primary nodes, and
1915
      C{True} for secondary nodes
1916

1917
  """
1918
  try:
1919
    device = _RecursiveAssembleBD(disk, owner, as_primary)
1920
    if isinstance(device, bdev.BlockDev):
1921
      dev_path = device.dev_path
1922
      link_name = None
1923
      if as_primary:
1924
        link_name = _SymlinkBlockDev(owner, dev_path, idx)
1925
  except errors.BlockDeviceError, err:
1926
    _Fail("Error while assembling disk: %s", err, exc=True)
1927
  except OSError, err:
1928
    _Fail("Error while symlinking disk: %s", err, exc=True)
1929

    
1930
  return dev_path, link_name
1931

    
1932

    
1933
def BlockdevShutdown(disk):
1934
  """Shut down a block device.
1935

1936
  First, if the device is assembled (Attach() is successful), then
1937
  the device is shutdown. Then the children of the device are
1938
  shutdown.
1939

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

1944
  @type disk: L{objects.Disk}
1945
  @param disk: the description of the disk we should
1946
      shutdown
1947
  @rtype: None
1948

1949
  """
1950
  msgs = []
1951
  r_dev = _RecursiveFindBD(disk)
1952
  if r_dev is not None:
1953
    r_path = r_dev.dev_path
1954
    try:
1955
      r_dev.Shutdown()
1956
      DevCacheManager.RemoveCache(r_path)
1957
    except errors.BlockDeviceError, err:
1958
      msgs.append(str(err))
1959

    
1960
  if disk.children:
1961
    for child in disk.children:
1962
      try:
1963
        BlockdevShutdown(child)
1964
      except RPCFail, err:
1965
        msgs.append(str(err))
1966

    
1967
  if msgs:
1968
    _Fail("; ".join(msgs))
1969

    
1970

    
1971
def BlockdevAddchildren(parent_cdev, new_cdevs):
1972
  """Extend a mirrored block device.
1973

1974
  @type parent_cdev: L{objects.Disk}
1975
  @param parent_cdev: the disk to which we should add children
1976
  @type new_cdevs: list of L{objects.Disk}
1977
  @param new_cdevs: the list of children which we should add
1978
  @rtype: None
1979

1980
  """
1981
  parent_bdev = _RecursiveFindBD(parent_cdev)
1982
  if parent_bdev is None:
1983
    _Fail("Can't find parent device '%s' in add children", parent_cdev)
1984
  new_bdevs = [_RecursiveFindBD(disk) for disk in new_cdevs]
1985
  if new_bdevs.count(None) > 0:
1986
    _Fail("Can't find new device(s) to add: %s:%s", new_bdevs, new_cdevs)
1987
  parent_bdev.AddChildren(new_bdevs)
1988

    
1989

    
1990
def BlockdevRemovechildren(parent_cdev, new_cdevs):
1991
  """Shrink a mirrored block device.
1992

1993
  @type parent_cdev: L{objects.Disk}
1994
  @param parent_cdev: the disk from which we should remove children
1995
  @type new_cdevs: list of L{objects.Disk}
1996
  @param new_cdevs: the list of children which we should remove
1997
  @rtype: None
1998

1999
  """
2000
  parent_bdev = _RecursiveFindBD(parent_cdev)
2001
  if parent_bdev is None:
2002
    _Fail("Can't find parent device '%s' in remove children", parent_cdev)
2003
  devs = []
2004
  for disk in new_cdevs:
2005
    rpath = disk.StaticDevPath()
2006
    if rpath is None:
2007
      bd = _RecursiveFindBD(disk)
2008
      if bd is None:
2009
        _Fail("Can't find device %s while removing children", disk)
2010
      else:
2011
        devs.append(bd.dev_path)
2012
    else:
2013
      if not utils.IsNormAbsPath(rpath):
2014
        _Fail("Strange path returned from StaticDevPath: '%s'", rpath)
2015
      devs.append(rpath)
2016
  parent_bdev.RemoveChildren(devs)
2017

    
2018

    
2019
def BlockdevGetmirrorstatus(disks):
2020
  """Get the mirroring status of a list of devices.
2021

2022
  @type disks: list of L{objects.Disk}
2023
  @param disks: the list of disks which we should query
2024
  @rtype: disk
2025
  @return: List of L{objects.BlockDevStatus}, one for each disk
2026
  @raise errors.BlockDeviceError: if any of the disks cannot be
2027
      found
2028

2029
  """
2030
  stats = []
2031
  for dsk in disks:
2032
    rbd = _RecursiveFindBD(dsk)
2033
    if rbd is None:
2034
      _Fail("Can't find device %s", dsk)
2035

    
2036
    stats.append(rbd.CombinedSyncStatus())
2037

    
2038
  return stats
2039

    
2040

    
2041
def BlockdevGetmirrorstatusMulti(disks):
2042
  """Get the mirroring status of a list of devices.
2043

2044
  @type disks: list of L{objects.Disk}
2045
  @param disks: the list of disks which we should query
2046
  @rtype: disk
2047
  @return: List of tuples, (bool, status), one for each disk; bool denotes
2048
    success/failure, status is L{objects.BlockDevStatus} on success, string
2049
    otherwise
2050

2051
  """
2052
  result = []
2053
  for disk in disks:
2054
    try:
2055
      rbd = _RecursiveFindBD(disk)
2056
      if rbd is None:
2057
        result.append((False, "Can't find device %s" % disk))
2058
        continue
2059

    
2060
      status = rbd.CombinedSyncStatus()
2061
    except errors.BlockDeviceError, err:
2062
      logging.exception("Error while getting disk status")
2063
      result.append((False, str(err)))
2064
    else:
2065
      result.append((True, status))
2066

    
2067
  assert len(disks) == len(result)
2068

    
2069
  return result
2070

    
2071

    
2072
def _RecursiveFindBD(disk):
2073
  """Check if a device is activated.
2074

2075
  If so, return information about the real device.
2076

2077
  @type disk: L{objects.Disk}
2078
  @param disk: the disk object we need to find
2079

2080
  @return: None if the device can't be found,
2081
      otherwise the device instance
2082

2083
  """
2084
  children = []
2085
  if disk.children:
2086
    for chdisk in disk.children:
2087
      children.append(_RecursiveFindBD(chdisk))
2088

    
2089
  return bdev.FindDevice(disk, children)
2090

    
2091

    
2092
def _OpenRealBD(disk):
2093
  """Opens the underlying block device of a disk.
2094

2095
  @type disk: L{objects.Disk}
2096
  @param disk: the disk object we want to open
2097

2098
  """
2099
  real_disk = _RecursiveFindBD(disk)
2100
  if real_disk is None:
2101
    _Fail("Block device '%s' is not set up", disk)
2102

    
2103
  real_disk.Open()
2104

    
2105
  return real_disk
2106

    
2107

    
2108
def BlockdevFind(disk):
2109
  """Check if a device is activated.
2110

2111
  If it is, return information about the real device.
2112

2113
  @type disk: L{objects.Disk}
2114
  @param disk: the disk to find
2115
  @rtype: None or objects.BlockDevStatus
2116
  @return: None if the disk cannot be found, otherwise a the current
2117
           information
2118

2119
  """
2120
  try:
2121
    rbd = _RecursiveFindBD(disk)
2122
  except errors.BlockDeviceError, err:
2123
    _Fail("Failed to find device: %s", err, exc=True)
2124

    
2125
  if rbd is None:
2126
    return None
2127

    
2128
  return rbd.GetSyncStatus()
2129

    
2130

    
2131
def BlockdevGetsize(disks):
2132
  """Computes the size of the given disks.
2133

2134
  If a disk is not found, returns None instead.
2135

2136
  @type disks: list of L{objects.Disk}
2137
  @param disks: the list of disk to compute the size for
2138
  @rtype: list
2139
  @return: list with elements None if the disk cannot be found,
2140
      otherwise the size
2141

2142
  """
2143
  result = []
2144
  for cf in disks:
2145
    try:
2146
      rbd = _RecursiveFindBD(cf)
2147
    except errors.BlockDeviceError:
2148
      result.append(None)
2149
      continue
2150
    if rbd is None:
2151
      result.append(None)
2152
    else:
2153
      result.append(rbd.GetActualSize())
2154
  return result
2155

    
2156

    
2157
def BlockdevExport(disk, dest_node, dest_path, cluster_name):
2158
  """Export a block device to a remote node.
2159

2160
  @type disk: L{objects.Disk}
2161
  @param disk: the description of the disk to export
2162
  @type dest_node: str
2163
  @param dest_node: the destination node to export to
2164
  @type dest_path: str
2165
  @param dest_path: the destination path on the target node
2166
  @type cluster_name: str
2167
  @param cluster_name: the cluster name, needed for SSH hostalias
2168
  @rtype: None
2169

2170
  """
2171
  real_disk = _OpenRealBD(disk)
2172

    
2173
  # the block size on the read dd is 1MiB to match our units
2174
  expcmd = utils.BuildShellCmd("set -e; set -o pipefail; "
2175
                               "dd if=%s bs=1048576 count=%s",
2176
                               real_disk.dev_path, str(disk.size))
2177

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

    
2187
  remotecmd = _GetSshRunner(cluster_name).BuildCmd(dest_node,
2188
                                                   constants.SSH_LOGIN_USER,
2189
                                                   destcmd)
2190

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

    
2194
  result = utils.RunCmd(["bash", "-c", command])
2195

    
2196
  if result.failed:
2197
    _Fail("Disk copy command '%s' returned error: %s"
2198
          " output: %s", command, result.fail_reason, result.output)
2199

    
2200

    
2201
def UploadFile(file_name, data, mode, uid, gid, atime, mtime):
2202
  """Write a file to the filesystem.
2203

2204
  This allows the master to overwrite(!) a file. It will only perform
2205
  the operation if the file belongs to a list of configuration files.
2206

2207
  @type file_name: str
2208
  @param file_name: the target file name
2209
  @type data: str
2210
  @param data: the new contents of the file
2211
  @type mode: int
2212
  @param mode: the mode to give the file (can be None)
2213
  @type uid: string
2214
  @param uid: the owner of the file
2215
  @type gid: string
2216
  @param gid: the group of the file
2217
  @type atime: float
2218
  @param atime: the atime to set on the file (can be None)
2219
  @type mtime: float
2220
  @param mtime: the mtime to set on the file (can be None)
2221
  @rtype: None
2222

2223
  """
2224
  file_name = vcluster.LocalizeVirtualPath(file_name)
2225

    
2226
  if not os.path.isabs(file_name):
2227
    _Fail("Filename passed to UploadFile is not absolute: '%s'", file_name)
2228

    
2229
  if file_name not in _ALLOWED_UPLOAD_FILES:
2230
    _Fail("Filename passed to UploadFile not in allowed upload targets: '%s'",
2231
          file_name)
2232

    
2233
  raw_data = _Decompress(data)
2234

    
2235
  if not (isinstance(uid, basestring) and isinstance(gid, basestring)):
2236
    _Fail("Invalid username/groupname type")
2237

    
2238
  getents = runtime.GetEnts()
2239
  uid = getents.LookupUser(uid)
2240
  gid = getents.LookupGroup(gid)
2241

    
2242
  utils.SafeWriteFile(file_name, None,
2243
                      data=raw_data, mode=mode, uid=uid, gid=gid,
2244
                      atime=atime, mtime=mtime)
2245

    
2246

    
2247
def RunOob(oob_program, command, node, timeout):
2248
  """Executes oob_program with given command on given node.
2249

2250
  @param oob_program: The path to the executable oob_program
2251
  @param command: The command to invoke on oob_program
2252
  @param node: The node given as an argument to the program
2253
  @param timeout: Timeout after which we kill the oob program
2254

2255
  @return: stdout
2256
  @raise RPCFail: If execution fails for some reason
2257

2258
  """
2259
  result = utils.RunCmd([oob_program, command, node], timeout=timeout)
2260

    
2261
  if result.failed:
2262
    _Fail("'%s' failed with reason '%s'; output: %s", result.cmd,
2263
          result.fail_reason, result.output)
2264

    
2265
  return result.stdout
2266

    
2267

    
2268
def _OSOndiskAPIVersion(os_dir):
2269
  """Compute and return the API version of a given OS.
2270

2271
  This function will try to read the API version of the OS residing in
2272
  the 'os_dir' directory.
2273

2274
  @type os_dir: str
2275
  @param os_dir: the directory in which we should look for the OS
2276
  @rtype: tuple
2277
  @return: tuple (status, data) with status denoting the validity and
2278
      data holding either the vaid versions or an error message
2279

2280
  """
2281
  api_file = utils.PathJoin(os_dir, constants.OS_API_FILE)
2282

    
2283
  try:
2284
    st = os.stat(api_file)
2285
  except EnvironmentError, err:
2286
    return False, ("Required file '%s' not found under path %s: %s" %
2287
                   (constants.OS_API_FILE, os_dir, utils.ErrnoOrStr(err)))
2288

    
2289
  if not stat.S_ISREG(stat.S_IFMT(st.st_mode)):
2290
    return False, ("File '%s' in %s is not a regular file" %
2291
                   (constants.OS_API_FILE, os_dir))
2292

    
2293
  try:
2294
    api_versions = utils.ReadFile(api_file).splitlines()
2295
  except EnvironmentError, err:
2296
    return False, ("Error while reading the API version file at %s: %s" %
2297
                   (api_file, utils.ErrnoOrStr(err)))
2298

    
2299
  try:
2300
    api_versions = [int(version.strip()) for version in api_versions]
2301
  except (TypeError, ValueError), err:
2302
    return False, ("API version(s) can't be converted to integer: %s" %
2303
                   str(err))
2304

    
2305
  return True, api_versions
2306

    
2307

    
2308
def DiagnoseOS(top_dirs=None):
2309
  """Compute the validity for all OSes.
2310

2311
  @type top_dirs: list
2312
  @param top_dirs: the list of directories in which to
2313
      search (if not given defaults to
2314
      L{pathutils.OS_SEARCH_PATH})
2315
  @rtype: list of L{objects.OS}
2316
  @return: a list of tuples (name, path, status, diagnose, variants,
2317
      parameters, api_version) for all (potential) OSes under all
2318
      search paths, where:
2319
          - name is the (potential) OS name
2320
          - path is the full path to the OS
2321
          - status True/False is the validity of the OS
2322
          - diagnose is the error message for an invalid OS, otherwise empty
2323
          - variants is a list of supported OS variants, if any
2324
          - parameters is a list of (name, help) parameters, if any
2325
          - api_version is a list of support OS API versions
2326

2327
  """
2328
  if top_dirs is None:
2329
    top_dirs = pathutils.OS_SEARCH_PATH
2330

    
2331
  result = []
2332
  for dir_name in top_dirs:
2333
    if os.path.isdir(dir_name):
2334
      try:
2335
        f_names = utils.ListVisibleFiles(dir_name)
2336
      except EnvironmentError, err:
2337
        logging.exception("Can't list the OS directory %s: %s", dir_name, err)
2338
        break
2339
      for name in f_names:
2340
        os_path = utils.PathJoin(dir_name, name)
2341
        status, os_inst = _TryOSFromDisk(name, base_dir=dir_name)
2342
        if status:
2343
          diagnose = ""
2344
          variants = os_inst.supported_variants
2345
          parameters = os_inst.supported_parameters
2346
          api_versions = os_inst.api_versions
2347
        else:
2348
          diagnose = os_inst
2349
          variants = parameters = api_versions = []
2350
        result.append((name, os_path, status, diagnose, variants,
2351
                       parameters, api_versions))
2352

    
2353
  return result
2354

    
2355

    
2356
def _TryOSFromDisk(name, base_dir=None):
2357
  """Create an OS instance from disk.
2358

2359
  This function will return an OS instance if the given name is a
2360
  valid OS name.
2361

2362
  @type base_dir: string
2363
  @keyword base_dir: Base directory containing OS installations.
2364
                     Defaults to a search in all the OS_SEARCH_PATH dirs.
2365
  @rtype: tuple
2366
  @return: success and either the OS instance if we find a valid one,
2367
      or error message
2368

2369
  """
2370
  if base_dir is None:
2371
    os_dir = utils.FindFile(name, pathutils.OS_SEARCH_PATH, os.path.isdir)
2372
  else:
2373
    os_dir = utils.FindFile(name, [base_dir], os.path.isdir)
2374

    
2375
  if os_dir is None:
2376
    return False, "Directory for OS %s not found in search path" % name
2377

    
2378
  status, api_versions = _OSOndiskAPIVersion(os_dir)
2379
  if not status:
2380
    # push the error up
2381
    return status, api_versions
2382

    
2383
  if not constants.OS_API_VERSIONS.intersection(api_versions):
2384
    return False, ("API version mismatch for path '%s': found %s, want %s." %
2385
                   (os_dir, api_versions, constants.OS_API_VERSIONS))
2386

    
2387
  # OS Files dictionary, we will populate it with the absolute path
2388
  # names; if the value is True, then it is a required file, otherwise
2389
  # an optional one
2390
  os_files = dict.fromkeys(constants.OS_SCRIPTS, True)
2391

    
2392
  if max(api_versions) >= constants.OS_API_V15:
2393
    os_files[constants.OS_VARIANTS_FILE] = False
2394

    
2395
  if max(api_versions) >= constants.OS_API_V20:
2396
    os_files[constants.OS_PARAMETERS_FILE] = True
2397
  else:
2398
    del os_files[constants.OS_SCRIPT_VERIFY]
2399

    
2400
  for (filename, required) in os_files.items():
2401
    os_files[filename] = utils.PathJoin(os_dir, filename)
2402

    
2403
    try:
2404
      st = os.stat(os_files[filename])
2405
    except EnvironmentError, err:
2406
      if err.errno == errno.ENOENT and not required:
2407
        del os_files[filename]
2408
        continue
2409
      return False, ("File '%s' under path '%s' is missing (%s)" %
2410
                     (filename, os_dir, utils.ErrnoOrStr(err)))
2411

    
2412
    if not stat.S_ISREG(stat.S_IFMT(st.st_mode)):
2413
      return False, ("File '%s' under path '%s' is not a regular file" %
2414
                     (filename, os_dir))
2415

    
2416
    if filename in constants.OS_SCRIPTS:
2417
      if stat.S_IMODE(st.st_mode) & stat.S_IXUSR != stat.S_IXUSR:
2418
        return False, ("File '%s' under path '%s' is not executable" %
2419
                       (filename, os_dir))
2420

    
2421
  variants = []
2422
  if constants.OS_VARIANTS_FILE in os_files:
2423
    variants_file = os_files[constants.OS_VARIANTS_FILE]
2424
    try:
2425
      variants = \
2426
        utils.FilterEmptyLinesAndComments(utils.ReadFile(variants_file))
2427
    except EnvironmentError, err:
2428
      # we accept missing files, but not other errors
2429
      if err.errno != errno.ENOENT:
2430
        return False, ("Error while reading the OS variants file at %s: %s" %
2431
                       (variants_file, utils.ErrnoOrStr(err)))
2432

    
2433
  parameters = []
2434
  if constants.OS_PARAMETERS_FILE in os_files:
2435
    parameters_file = os_files[constants.OS_PARAMETERS_FILE]
2436
    try:
2437
      parameters = utils.ReadFile(parameters_file).splitlines()
2438
    except EnvironmentError, err:
2439
      return False, ("Error while reading the OS parameters file at %s: %s" %
2440
                     (parameters_file, utils.ErrnoOrStr(err)))
2441
    parameters = [v.split(None, 1) for v in parameters]
2442

    
2443
  os_obj = objects.OS(name=name, path=os_dir,
2444
                      create_script=os_files[constants.OS_SCRIPT_CREATE],
2445
                      export_script=os_files[constants.OS_SCRIPT_EXPORT],
2446
                      import_script=os_files[constants.OS_SCRIPT_IMPORT],
2447
                      rename_script=os_files[constants.OS_SCRIPT_RENAME],
2448
                      verify_script=os_files.get(constants.OS_SCRIPT_VERIFY,
2449
                                                 None),
2450
                      supported_variants=variants,
2451
                      supported_parameters=parameters,
2452
                      api_versions=api_versions)
2453
  return True, os_obj
2454

    
2455

    
2456
def OSFromDisk(name, base_dir=None):
2457
  """Create an OS instance from disk.
2458

2459
  This function will return an OS instance if the given name is a
2460
  valid OS name. Otherwise, it will raise an appropriate
2461
  L{RPCFail} exception, detailing why this is not a valid OS.
2462

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

2466
  @type base_dir: string
2467
  @keyword base_dir: Base directory containing OS installations.
2468
                     Defaults to a search in all the OS_SEARCH_PATH dirs.
2469
  @rtype: L{objects.OS}
2470
  @return: the OS instance if we find a valid one
2471
  @raise RPCFail: if we don't find a valid OS
2472

2473
  """
2474
  name_only = objects.OS.GetName(name)
2475
  status, payload = _TryOSFromDisk(name_only, base_dir)
2476

    
2477
  if not status:
2478
    _Fail(payload)
2479

    
2480
  return payload
2481

    
2482

    
2483
def OSCoreEnv(os_name, inst_os, os_params, debug=0):
2484
  """Calculate the basic environment for an os script.
2485

2486
  @type os_name: str
2487
  @param os_name: full operating system name (including variant)
2488
  @type inst_os: L{objects.OS}
2489
  @param inst_os: operating system for which the environment is being built
2490
  @type os_params: dict
2491
  @param os_params: the OS parameters
2492
  @type debug: integer
2493
  @param debug: debug level (0 or 1, for OS Api 10)
2494
  @rtype: dict
2495
  @return: dict of environment variables
2496
  @raise errors.BlockDeviceError: if the block device
2497
      cannot be found
2498

2499
  """
2500
  result = {}
2501
  api_version = \
2502
    max(constants.OS_API_VERSIONS.intersection(inst_os.api_versions))
2503
  result["OS_API_VERSION"] = "%d" % api_version
2504
  result["OS_NAME"] = inst_os.name
2505
  result["DEBUG_LEVEL"] = "%d" % debug
2506

    
2507
  # OS variants
2508
  if api_version >= constants.OS_API_V15 and inst_os.supported_variants:
2509
    variant = objects.OS.GetVariant(os_name)
2510
    if not variant:
2511
      variant = inst_os.supported_variants[0]
2512
  else:
2513
    variant = ""
2514
  result["OS_VARIANT"] = variant
2515

    
2516
  # OS params
2517
  for pname, pvalue in os_params.items():
2518
    result["OSP_%s" % pname.upper()] = pvalue
2519

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

    
2525
  return result
2526

    
2527

    
2528
def OSEnvironment(instance, inst_os, debug=0):
2529
  """Calculate the environment for an os script.
2530

2531
  @type instance: L{objects.Instance}
2532
  @param instance: target instance for the os script run
2533
  @type inst_os: L{objects.OS}
2534
  @param inst_os: operating system for which the environment is being built
2535
  @type debug: integer
2536
  @param debug: debug level (0 or 1, for OS Api 10)
2537
  @rtype: dict
2538
  @return: dict of environment variables
2539
  @raise errors.BlockDeviceError: if the block device
2540
      cannot be found
2541

2542
  """
2543
  result = OSCoreEnv(instance.os, inst_os, instance.osparams, debug=debug)
2544

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

    
2548
  result["HYPERVISOR"] = instance.hypervisor
2549
  result["DISK_COUNT"] = "%d" % len(instance.disks)
2550
  result["NIC_COUNT"] = "%d" % len(instance.nics)
2551
  result["INSTANCE_SECONDARY_NODES"] = \
2552
      ("%s" % " ".join(instance.secondary_nodes))
2553

    
2554
  # Disks
2555
  for idx, disk in enumerate(instance.disks):
2556
    real_disk = _OpenRealBD(disk)
2557
    result["DISK_%d_PATH" % idx] = real_disk.dev_path
2558
    result["DISK_%d_ACCESS" % idx] = disk.mode
2559
    result["DISK_%d_UUID" % idx] = disk.uuid
2560
    if disk.name:
2561
      result["DISK_%d_NAME" % idx] = disk.name
2562
    if constants.HV_DISK_TYPE in instance.hvparams:
2563
      result["DISK_%d_FRONTEND_TYPE" % idx] = \
2564
        instance.hvparams[constants.HV_DISK_TYPE]
2565
    if disk.dev_type in constants.LDS_BLOCK:
2566
      result["DISK_%d_BACKEND_TYPE" % idx] = "block"
2567
    elif disk.dev_type == constants.LD_FILE:
2568
      result["DISK_%d_BACKEND_TYPE" % idx] = \
2569
        "file:%s" % disk.physical_id[0]
2570

    
2571
  # NICs
2572
  for idx, nic in enumerate(instance.nics):
2573
    result["NIC_%d_MAC" % idx] = nic.mac
2574
    result["NIC_%d_UUID" % idx] = nic.uuid
2575
    if nic.name:
2576
      result["NIC_%d_NAME" % idx] = nic.name
2577
    if nic.ip:
2578
      result["NIC_%d_IP" % idx] = nic.ip
2579
    result["NIC_%d_MODE" % idx] = nic.nicparams[constants.NIC_MODE]
2580
    if nic.nicparams[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED:
2581
      result["NIC_%d_BRIDGE" % idx] = nic.nicparams[constants.NIC_LINK]
2582
    if nic.nicparams[constants.NIC_LINK]:
2583
      result["NIC_%d_LINK" % idx] = nic.nicparams[constants.NIC_LINK]
2584
    if nic.netinfo:
2585
      nobj = objects.Network.FromDict(nic.netinfo)
2586
      result.update(nobj.HooksDict("NIC_%d_" % idx))
2587
    if constants.HV_NIC_TYPE in instance.hvparams:
2588
      result["NIC_%d_FRONTEND_TYPE" % idx] = \
2589
        instance.hvparams[constants.HV_NIC_TYPE]
2590

    
2591
  # HV/BE params
2592
  for source, kind in [(instance.beparams, "BE"), (instance.hvparams, "HV")]:
2593
    for key, value in source.items():
2594
      result["INSTANCE_%s_%s" % (kind, key)] = str(value)
2595

    
2596
  return result
2597

    
2598

    
2599
def DiagnoseExtStorage(top_dirs=None):
2600
  """Compute the validity for all ExtStorage Providers.
2601

2602
  @type top_dirs: list
2603
  @param top_dirs: the list of directories in which to
2604
      search (if not given defaults to
2605
      L{pathutils.ES_SEARCH_PATH})
2606
  @rtype: list of L{objects.ExtStorage}
2607
  @return: a list of tuples (name, path, status, diagnose, parameters)
2608
      for all (potential) ExtStorage Providers under all
2609
      search paths, where:
2610
          - name is the (potential) ExtStorage Provider
2611
          - path is the full path to the ExtStorage Provider
2612
          - status True/False is the validity of the ExtStorage Provider
2613
          - diagnose is the error message for an invalid ExtStorage Provider,
2614
            otherwise empty
2615
          - parameters is a list of (name, help) parameters, if any
2616

2617
  """
2618
  if top_dirs is None:
2619
    top_dirs = pathutils.ES_SEARCH_PATH
2620

    
2621
  result = []
2622
  for dir_name in top_dirs:
2623
    if os.path.isdir(dir_name):
2624
      try:
2625
        f_names = utils.ListVisibleFiles(dir_name)
2626
      except EnvironmentError, err:
2627
        logging.exception("Can't list the ExtStorage directory %s: %s",
2628
                          dir_name, err)
2629
        break
2630
      for name in f_names:
2631
        es_path = utils.PathJoin(dir_name, name)
2632
        status, es_inst = bdev.ExtStorageFromDisk(name, base_dir=dir_name)
2633
        if status:
2634
          diagnose = ""
2635
          parameters = es_inst.supported_parameters
2636
        else:
2637
          diagnose = es_inst
2638
          parameters = []
2639
        result.append((name, es_path, status, diagnose, parameters))
2640

    
2641
  return result
2642

    
2643

    
2644
def BlockdevGrow(disk, amount, dryrun, backingstore):
2645
  """Grow a stack of block devices.
2646

2647
  This function is called recursively, with the childrens being the
2648
  first ones to resize.
2649

2650
  @type disk: L{objects.Disk}
2651
  @param disk: the disk to be grown
2652
  @type amount: integer
2653
  @param amount: the amount (in mebibytes) to grow with
2654
  @type dryrun: boolean
2655
  @param dryrun: whether to execute the operation in simulation mode
2656
      only, without actually increasing the size
2657
  @param backingstore: whether to execute the operation on backing storage
2658
      only, or on "logical" storage only; e.g. DRBD is logical storage,
2659
      whereas LVM, file, RBD are backing storage
2660
  @rtype: (status, result)
2661
  @return: a tuple with the status of the operation (True/False), and
2662
      the errors message if status is False
2663

2664
  """
2665
  r_dev = _RecursiveFindBD(disk)
2666
  if r_dev is None:
2667
    _Fail("Cannot find block device %s", disk)
2668

    
2669
  try:
2670
    r_dev.Grow(amount, dryrun, backingstore)
2671
  except errors.BlockDeviceError, err:
2672
    _Fail("Failed to grow block device: %s", err, exc=True)
2673

    
2674

    
2675
def BlockdevSnapshot(disk):
2676
  """Create a snapshot copy of a block device.
2677

2678
  This function is called recursively, and the snapshot is actually created
2679
  just for the leaf lvm backend device.
2680

2681
  @type disk: L{objects.Disk}
2682
  @param disk: the disk to be snapshotted
2683
  @rtype: string
2684
  @return: snapshot disk ID as (vg, lv)
2685

2686
  """
2687
  if disk.dev_type == constants.LD_DRBD8:
2688
    if not disk.children:
2689
      _Fail("DRBD device '%s' without backing storage cannot be snapshotted",
2690
            disk.unique_id)
2691
    return BlockdevSnapshot(disk.children[0])
2692
  elif disk.dev_type == constants.LD_LV:
2693
    r_dev = _RecursiveFindBD(disk)
2694
    if r_dev is not None:
2695
      # FIXME: choose a saner value for the snapshot size
2696
      # let's stay on the safe side and ask for the full size, for now
2697
      return r_dev.Snapshot(disk.size)
2698
    else:
2699
      _Fail("Cannot find block device %s", disk)
2700
  else:
2701
    _Fail("Cannot snapshot non-lvm block device '%s' of type '%s'",
2702
          disk.unique_id, disk.dev_type)
2703

    
2704

    
2705
def BlockdevSetInfo(disk, info):
2706
  """Sets 'metadata' information on block devices.
2707

2708
  This function sets 'info' metadata on block devices. Initial
2709
  information is set at device creation; this function should be used
2710
  for example after renames.
2711

2712
  @type disk: L{objects.Disk}
2713
  @param disk: the disk to be grown
2714
  @type info: string
2715
  @param info: new 'info' metadata
2716
  @rtype: (status, result)
2717
  @return: a tuple with the status of the operation (True/False), and
2718
      the errors message if status is False
2719

2720
  """
2721
  r_dev = _RecursiveFindBD(disk)
2722
  if r_dev is None:
2723
    _Fail("Cannot find block device %s", disk)
2724

    
2725
  try:
2726
    r_dev.SetInfo(info)
2727
  except errors.BlockDeviceError, err:
2728
    _Fail("Failed to set information on block device: %s", err, exc=True)
2729

    
2730

    
2731
def FinalizeExport(instance, snap_disks):
2732
  """Write out the export configuration information.
2733

2734
  @type instance: L{objects.Instance}
2735
  @param instance: the instance which we export, used for
2736
      saving configuration
2737
  @type snap_disks: list of L{objects.Disk}
2738
  @param snap_disks: list of snapshot block devices, which
2739
      will be used to get the actual name of the dump file
2740

2741
  @rtype: None
2742

2743
  """
2744
  destdir = utils.PathJoin(pathutils.EXPORT_DIR, instance.name + ".new")
2745
  finaldestdir = utils.PathJoin(pathutils.EXPORT_DIR, instance.name)
2746

    
2747
  config = objects.SerializableConfigParser()
2748

    
2749
  config.add_section(constants.INISECT_EXP)
2750
  config.set(constants.INISECT_EXP, "version", "0")
2751
  config.set(constants.INISECT_EXP, "timestamp", "%d" % int(time.time()))
2752
  config.set(constants.INISECT_EXP, "source", instance.primary_node)
2753
  config.set(constants.INISECT_EXP, "os", instance.os)
2754
  config.set(constants.INISECT_EXP, "compression", "none")
2755

    
2756
  config.add_section(constants.INISECT_INS)
2757
  config.set(constants.INISECT_INS, "name", instance.name)
2758
  config.set(constants.INISECT_INS, "maxmem", "%d" %
2759
             instance.beparams[constants.BE_MAXMEM])
2760
  config.set(constants.INISECT_INS, "minmem", "%d" %
2761
             instance.beparams[constants.BE_MINMEM])
2762
  # "memory" is deprecated, but useful for exporting to old ganeti versions
2763
  config.set(constants.INISECT_INS, "memory", "%d" %
2764
             instance.beparams[constants.BE_MAXMEM])
2765
  config.set(constants.INISECT_INS, "vcpus", "%d" %
2766
             instance.beparams[constants.BE_VCPUS])
2767
  config.set(constants.INISECT_INS, "disk_template", instance.disk_template)
2768
  config.set(constants.INISECT_INS, "hypervisor", instance.hypervisor)
2769
  config.set(constants.INISECT_INS, "tags", " ".join(instance.GetTags()))
2770

    
2771
  nic_total = 0
2772
  for nic_count, nic in enumerate(instance.nics):
2773
    nic_total += 1
2774
    config.set(constants.INISECT_INS, "nic%d_mac" %
2775
               nic_count, "%s" % nic.mac)
2776
    config.set(constants.INISECT_INS, "nic%d_ip" % nic_count, "%s" % nic.ip)
2777
    config.set(constants.INISECT_INS, "nic%d_network" % nic_count,
2778
               "%s" % nic.network)
2779
    for param in constants.NICS_PARAMETER_TYPES:
2780
      config.set(constants.INISECT_INS, "nic%d_%s" % (nic_count, param),
2781
                 "%s" % nic.nicparams.get(param, None))
2782
  # TODO: redundant: on load can read nics until it doesn't exist
2783
  config.set(constants.INISECT_INS, "nic_count", "%d" % nic_total)
2784

    
2785
  disk_total = 0
2786
  for disk_count, disk in enumerate(snap_disks):
2787
    if disk:
2788
      disk_total += 1
2789
      config.set(constants.INISECT_INS, "disk%d_ivname" % disk_count,
2790
                 ("%s" % disk.iv_name))
2791
      config.set(constants.INISECT_INS, "disk%d_dump" % disk_count,
2792
                 ("%s" % disk.physical_id[1]))
2793
      config.set(constants.INISECT_INS, "disk%d_size" % disk_count,
2794
                 ("%d" % disk.size))
2795

    
2796
  config.set(constants.INISECT_INS, "disk_count", "%d" % disk_total)
2797

    
2798
  # New-style hypervisor/backend parameters
2799

    
2800
  config.add_section(constants.INISECT_HYP)
2801
  for name, value in instance.hvparams.items():
2802
    if name not in constants.HVC_GLOBALS:
2803
      config.set(constants.INISECT_HYP, name, str(value))
2804

    
2805
  config.add_section(constants.INISECT_BEP)
2806
  for name, value in instance.beparams.items():
2807
    config.set(constants.INISECT_BEP, name, str(value))
2808

    
2809
  config.add_section(constants.INISECT_OSP)
2810
  for name, value in instance.osparams.items():
2811
    config.set(constants.INISECT_OSP, name, str(value))
2812

    
2813
  utils.WriteFile(utils.PathJoin(destdir, constants.EXPORT_CONF_FILE),
2814
                  data=config.Dumps())
2815
  shutil.rmtree(finaldestdir, ignore_errors=True)
2816
  shutil.move(destdir, finaldestdir)
2817

    
2818

    
2819
def ExportInfo(dest):
2820
  """Get export configuration information.
2821

2822
  @type dest: str
2823
  @param dest: directory containing the export
2824

2825
  @rtype: L{objects.SerializableConfigParser}
2826
  @return: a serializable config file containing the
2827
      export info
2828

2829
  """
2830
  cff = utils.PathJoin(dest, constants.EXPORT_CONF_FILE)
2831

    
2832
  config = objects.SerializableConfigParser()
2833
  config.read(cff)
2834

    
2835
  if (not config.has_section(constants.INISECT_EXP) or
2836
      not config.has_section(constants.INISECT_INS)):
2837
    _Fail("Export info file doesn't have the required fields")
2838

    
2839
  return config.Dumps()
2840

    
2841

    
2842
def ListExports():
2843
  """Return a list of exports currently available on this machine.
2844

2845
  @rtype: list
2846
  @return: list of the exports
2847

2848
  """
2849
  if os.path.isdir(pathutils.EXPORT_DIR):
2850
    return sorted(utils.ListVisibleFiles(pathutils.EXPORT_DIR))
2851
  else:
2852
    _Fail("No exports directory")
2853

    
2854

    
2855
def RemoveExport(export):
2856
  """Remove an existing export from the node.
2857

2858
  @type export: str
2859
  @param export: the name of the export to remove
2860
  @rtype: None
2861

2862
  """
2863
  target = utils.PathJoin(pathutils.EXPORT_DIR, export)
2864

    
2865
  try:
2866
    shutil.rmtree(target)
2867
  except EnvironmentError, err:
2868
    _Fail("Error while removing the export: %s", err, exc=True)
2869

    
2870

    
2871
def BlockdevRename(devlist):
2872
  """Rename a list of block devices.
2873

2874
  @type devlist: list of tuples
2875
  @param devlist: list of tuples of the form  (disk,
2876
      new_logical_id, new_physical_id); disk is an
2877
      L{objects.Disk} object describing the current disk,
2878
      and new logical_id/physical_id is the name we
2879
      rename it to
2880
  @rtype: boolean
2881
  @return: True if all renames succeeded, False otherwise
2882

2883
  """
2884
  msgs = []
2885
  result = True
2886
  for disk, unique_id in devlist:
2887
    dev = _RecursiveFindBD(disk)
2888
    if dev is None:
2889
      msgs.append("Can't find device %s in rename" % str(disk))
2890
      result = False
2891
      continue
2892
    try:
2893
      old_rpath = dev.dev_path
2894
      dev.Rename(unique_id)
2895
      new_rpath = dev.dev_path
2896
      if old_rpath != new_rpath:
2897
        DevCacheManager.RemoveCache(old_rpath)
2898
        # FIXME: we should add the new cache information here, like:
2899
        # DevCacheManager.UpdateCache(new_rpath, owner, ...)
2900
        # but we don't have the owner here - maybe parse from existing
2901
        # cache? for now, we only lose lvm data when we rename, which
2902
        # is less critical than DRBD or MD
2903
    except errors.BlockDeviceError, err:
2904
      msgs.append("Can't rename device '%s' to '%s': %s" %
2905
                  (dev, unique_id, err))
2906
      logging.exception("Can't rename device '%s' to '%s'", dev, unique_id)
2907
      result = False
2908
  if not result:
2909
    _Fail("; ".join(msgs))
2910

    
2911

    
2912
def _TransformFileStorageDir(fs_dir):
2913
  """Checks whether given file_storage_dir is valid.
2914

2915
  Checks wheter the given fs_dir is within the cluster-wide default
2916
  file_storage_dir or the shared_file_storage_dir, which are stored in
2917
  SimpleStore. Only paths under those directories are allowed.
2918

2919
  @type fs_dir: str
2920
  @param fs_dir: the path to check
2921

2922
  @return: the normalized path if valid, None otherwise
2923

2924
  """
2925
  if not (constants.ENABLE_FILE_STORAGE or
2926
          constants.ENABLE_SHARED_FILE_STORAGE):
2927
    _Fail("File storage disabled at configure time")
2928

    
2929
  bdev.CheckFileStoragePath(fs_dir)
2930

    
2931
  return os.path.normpath(fs_dir)
2932

    
2933

    
2934
def CreateFileStorageDir(file_storage_dir):
2935
  """Create file storage directory.
2936

2937
  @type file_storage_dir: str
2938
  @param file_storage_dir: directory to create
2939

2940
  @rtype: tuple
2941
  @return: tuple with first element a boolean indicating wheter dir
2942
      creation was successful or not
2943

2944
  """
2945
  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
2946
  if os.path.exists(file_storage_dir):
2947
    if not os.path.isdir(file_storage_dir):
2948
      _Fail("Specified storage dir '%s' is not a directory",
2949
            file_storage_dir)
2950
  else:
2951
    try:
2952
      os.makedirs(file_storage_dir, 0750)
2953
    except OSError, err:
2954
      _Fail("Cannot create file storage directory '%s': %s",
2955
            file_storage_dir, err, exc=True)
2956

    
2957

    
2958
def RemoveFileStorageDir(file_storage_dir):
2959
  """Remove file storage directory.
2960

2961
  Remove it only if it's empty. If not log an error and return.
2962

2963
  @type file_storage_dir: str
2964
  @param file_storage_dir: the directory we should cleanup
2965
  @rtype: tuple (success,)
2966
  @return: tuple of one element, C{success}, denoting
2967
      whether the operation was successful
2968

2969
  """
2970
  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
2971
  if os.path.exists(file_storage_dir):
2972
    if not os.path.isdir(file_storage_dir):
2973
      _Fail("Specified Storage directory '%s' is not a directory",
2974
            file_storage_dir)
2975
    # deletes dir only if empty, otherwise we want to fail the rpc call
2976
    try:
2977
      os.rmdir(file_storage_dir)
2978
    except OSError, err:
2979
      _Fail("Cannot remove file storage directory '%s': %s",
2980
            file_storage_dir, err)
2981

    
2982

    
2983
def RenameFileStorageDir(old_file_storage_dir, new_file_storage_dir):
2984
  """Rename the file storage directory.
2985

2986
  @type old_file_storage_dir: str
2987
  @param old_file_storage_dir: the current path
2988
  @type new_file_storage_dir: str
2989
  @param new_file_storage_dir: the name we should rename to
2990
  @rtype: tuple (success,)
2991
  @return: tuple of one element, C{success}, denoting
2992
      whether the operation was successful
2993

2994
  """
2995
  old_file_storage_dir = _TransformFileStorageDir(old_file_storage_dir)
2996
  new_file_storage_dir = _TransformFileStorageDir(new_file_storage_dir)
2997
  if not os.path.exists(new_file_storage_dir):
2998
    if os.path.isdir(old_file_storage_dir):
2999
      try:
3000
        os.rename(old_file_storage_dir, new_file_storage_dir)
3001
      except OSError, err:
3002
        _Fail("Cannot rename '%s' to '%s': %s",
3003
              old_file_storage_dir, new_file_storage_dir, err)
3004
    else:
3005
      _Fail("Specified storage dir '%s' is not a directory",
3006
            old_file_storage_dir)
3007
  else:
3008
    if os.path.exists(old_file_storage_dir):
3009
      _Fail("Cannot rename '%s' to '%s': both locations exist",
3010
            old_file_storage_dir, new_file_storage_dir)
3011

    
3012

    
3013
def _EnsureJobQueueFile(file_name):
3014
  """Checks whether the given filename is in the queue directory.
3015

3016
  @type file_name: str
3017
  @param file_name: the file name we should check
3018
  @rtype: None
3019
  @raises RPCFail: if the file is not valid
3020

3021
  """
3022
  if not utils.IsBelowDir(pathutils.QUEUE_DIR, file_name):
3023
    _Fail("Passed job queue file '%s' does not belong to"
3024
          " the queue directory '%s'", file_name, pathutils.QUEUE_DIR)
3025

    
3026

    
3027
def JobQueueUpdate(file_name, content):
3028
  """Updates a file in the queue directory.
3029

3030
  This is just a wrapper over L{utils.io.WriteFile}, with proper
3031
  checking.
3032

3033
  @type file_name: str
3034
  @param file_name: the job file name
3035
  @type content: str
3036
  @param content: the new job contents
3037
  @rtype: boolean
3038
  @return: the success of the operation
3039

3040
  """
3041
  file_name = vcluster.LocalizeVirtualPath(file_name)
3042

    
3043
  _EnsureJobQueueFile(file_name)
3044
  getents = runtime.GetEnts()
3045

    
3046
  # Write and replace the file atomically
3047
  utils.WriteFile(file_name, data=_Decompress(content), uid=getents.masterd_uid,
3048
                  gid=getents.daemons_gid, mode=constants.JOB_QUEUE_FILES_PERMS)
3049

    
3050

    
3051
def JobQueueRename(old, new):
3052
  """Renames a job queue file.
3053

3054
  This is just a wrapper over os.rename with proper checking.
3055

3056
  @type old: str
3057
  @param old: the old (actual) file name
3058
  @type new: str
3059
  @param new: the desired file name
3060
  @rtype: tuple
3061
  @return: the success of the operation and payload
3062

3063
  """
3064
  old = vcluster.LocalizeVirtualPath(old)
3065
  new = vcluster.LocalizeVirtualPath(new)
3066

    
3067
  _EnsureJobQueueFile(old)
3068
  _EnsureJobQueueFile(new)
3069

    
3070
  getents = runtime.GetEnts()
3071

    
3072
  utils.RenameFile(old, new, mkdir=True, mkdir_mode=0750,
3073
                   dir_uid=getents.masterd_uid, dir_gid=getents.daemons_gid)
3074

    
3075

    
3076
def BlockdevClose(instance_name, disks):
3077
  """Closes the given block devices.
3078

3079
  This means they will be switched to secondary mode (in case of
3080
  DRBD).
3081

3082
  @param instance_name: if the argument is not empty, the symlinks
3083
      of this instance will be removed
3084
  @type disks: list of L{objects.Disk}
3085
  @param disks: the list of disks to be closed
3086
  @rtype: tuple (success, message)
3087
  @return: a tuple of success and message, where success
3088
      indicates the succes of the operation, and message
3089
      which will contain the error details in case we
3090
      failed
3091

3092
  """
3093
  bdevs = []
3094
  for cf in disks:
3095
    rd = _RecursiveFindBD(cf)
3096
    if rd is None:
3097
      _Fail("Can't find device %s", cf)
3098
    bdevs.append(rd)
3099

    
3100
  msg = []
3101
  for rd in bdevs:
3102
    try:
3103
      rd.Close()
3104
    except errors.BlockDeviceError, err:
3105
      msg.append(str(err))
3106
  if msg:
3107
    _Fail("Can't make devices secondary: %s", ",".join(msg))
3108
  else:
3109
    if instance_name:
3110
      _RemoveBlockDevLinks(instance_name, disks)
3111

    
3112

    
3113
def ValidateHVParams(hvname, hvparams):
3114
  """Validates the given hypervisor parameters.
3115

3116
  @type hvname: string
3117
  @param hvname: the hypervisor name
3118
  @type hvparams: dict
3119
  @param hvparams: the hypervisor parameters to be validated
3120
  @rtype: None
3121

3122
  """
3123
  try:
3124
    hv_type = hypervisor.GetHypervisor(hvname)
3125
    hv_type.ValidateParameters(hvparams)
3126
  except errors.HypervisorError, err:
3127
    _Fail(str(err), log=False)
3128

    
3129

    
3130
def _CheckOSPList(os_obj, parameters):
3131
  """Check whether a list of parameters is supported by the OS.
3132

3133
  @type os_obj: L{objects.OS}
3134
  @param os_obj: OS object to check
3135
  @type parameters: list
3136
  @param parameters: the list of parameters to check
3137

3138
  """
3139
  supported = [v[0] for v in os_obj.supported_parameters]
3140
  delta = frozenset(parameters).difference(supported)
3141
  if delta:
3142
    _Fail("The following parameters are not supported"
3143
          " by the OS %s: %s" % (os_obj.name, utils.CommaJoin(delta)))
3144

    
3145

    
3146
def ValidateOS(required, osname, checks, osparams):
3147
  """Validate the given OS' parameters.
3148

3149
  @type required: boolean
3150
  @param required: whether absence of the OS should translate into
3151
      failure or not
3152
  @type osname: string
3153
  @param osname: the OS to be validated
3154
  @type checks: list
3155
  @param checks: list of the checks to run (currently only 'parameters')
3156
  @type osparams: dict
3157
  @param osparams: dictionary with OS parameters
3158
  @rtype: boolean
3159
  @return: True if the validation passed, or False if the OS was not
3160
      found and L{required} was false
3161

3162
  """
3163
  if not constants.OS_VALIDATE_CALLS.issuperset(checks):
3164
    _Fail("Unknown checks required for OS %s: %s", osname,
3165
          set(checks).difference(constants.OS_VALIDATE_CALLS))
3166

    
3167
  name_only = objects.OS.GetName(osname)
3168
  status, tbv = _TryOSFromDisk(name_only, None)
3169

    
3170
  if not status:
3171
    if required:
3172
      _Fail(tbv)
3173
    else:
3174
      return False
3175

    
3176
  if max(tbv.api_versions) < constants.OS_API_V20:
3177
    return True
3178

    
3179
  if constants.OS_VALIDATE_PARAMETERS in checks:
3180
    _CheckOSPList(tbv, osparams.keys())
3181

    
3182
  validate_env = OSCoreEnv(osname, tbv, osparams)
3183
  result = utils.RunCmd([tbv.verify_script] + checks, env=validate_env,
3184
                        cwd=tbv.path, reset_env=True)
3185
  if result.failed:
3186
    logging.error("os validate command '%s' returned error: %s output: %s",
3187
                  result.cmd, result.fail_reason, result.output)
3188
    _Fail("OS validation script failed (%s), output: %s",
3189
          result.fail_reason, result.output, log=False)
3190

    
3191
  return True
3192

    
3193

    
3194
def DemoteFromMC():
3195
  """Demotes the current node from master candidate role.
3196

3197
  """
3198
  # try to ensure we're not the master by mistake
3199
  master, myself = ssconf.GetMasterAndMyself()
3200
  if master == myself:
3201
    _Fail("ssconf status shows I'm the master node, will not demote")
3202

    
3203
  result = utils.RunCmd([pathutils.DAEMON_UTIL, "check", constants.MASTERD])
3204
  if not result.failed:
3205
    _Fail("The master daemon is running, will not demote")
3206

    
3207
  try:
3208
    if os.path.isfile(pathutils.CLUSTER_CONF_FILE):
3209
      utils.CreateBackup(pathutils.CLUSTER_CONF_FILE)
3210
  except EnvironmentError, err:
3211
    if err.errno != errno.ENOENT:
3212
      _Fail("Error while backing up cluster file: %s", err, exc=True)
3213

    
3214
  utils.RemoveFile(pathutils.CLUSTER_CONF_FILE)
3215

    
3216

    
3217
def _GetX509Filenames(cryptodir, name):
3218
  """Returns the full paths for the private key and certificate.
3219

3220
  """
3221
  return (utils.PathJoin(cryptodir, name),
3222
          utils.PathJoin(cryptodir, name, _X509_KEY_FILE),
3223
          utils.PathJoin(cryptodir, name, _X509_CERT_FILE))
3224

    
3225

    
3226
def CreateX509Certificate(validity, cryptodir=pathutils.CRYPTO_KEYS_DIR):
3227
  """Creates a new X509 certificate for SSL/TLS.
3228

3229
  @type validity: int
3230
  @param validity: Validity in seconds
3231
  @rtype: tuple; (string, string)
3232
  @return: Certificate name and public part
3233

3234
  """
3235
  (key_pem, cert_pem) = \
3236
    utils.GenerateSelfSignedX509Cert(netutils.Hostname.GetSysName(),
3237
                                     min(validity, _MAX_SSL_CERT_VALIDITY))
3238

    
3239
  cert_dir = tempfile.mkdtemp(dir=cryptodir,
3240
                              prefix="x509-%s-" % utils.TimestampForFilename())
3241
  try:
3242
    name = os.path.basename(cert_dir)
3243
    assert len(name) > 5
3244

    
3245
    (_, key_file, cert_file) = _GetX509Filenames(cryptodir, name)
3246

    
3247
    utils.WriteFile(key_file, mode=0400, data=key_pem)
3248
    utils.WriteFile(cert_file, mode=0400, data=cert_pem)
3249

    
3250
    # Never return private key as it shouldn't leave the node
3251
    return (name, cert_pem)
3252
  except Exception:
3253
    shutil.rmtree(cert_dir, ignore_errors=True)
3254
    raise
3255

    
3256

    
3257
def RemoveX509Certificate(name, cryptodir=pathutils.CRYPTO_KEYS_DIR):
3258
  """Removes a X509 certificate.
3259

3260
  @type name: string
3261
  @param name: Certificate name
3262

3263
  """
3264
  (cert_dir, key_file, cert_file) = _GetX509Filenames(cryptodir, name)
3265

    
3266
  utils.RemoveFile(key_file)
3267
  utils.RemoveFile(cert_file)
3268

    
3269
  try:
3270
    os.rmdir(cert_dir)
3271
  except EnvironmentError, err:
3272
    _Fail("Cannot remove certificate directory '%s': %s",
3273
          cert_dir, err)
3274

    
3275

    
3276
def _GetImportExportIoCommand(instance, mode, ieio, ieargs):
3277
  """Returns the command for the requested input/output.
3278

3279
  @type instance: L{objects.Instance}
3280
  @param instance: The instance object
3281
  @param mode: Import/export mode
3282
  @param ieio: Input/output type
3283
  @param ieargs: Input/output arguments
3284

3285
  """
3286
  assert mode in (constants.IEM_IMPORT, constants.IEM_EXPORT)
3287

    
3288
  env = None
3289
  prefix = None
3290
  suffix = None
3291
  exp_size = None
3292

    
3293
  if ieio == constants.IEIO_FILE:
3294
    (filename, ) = ieargs
3295

    
3296
    if not utils.IsNormAbsPath(filename):
3297
      _Fail("Path '%s' is not normalized or absolute", filename)
3298

    
3299
    real_filename = os.path.realpath(filename)
3300
    directory = os.path.dirname(real_filename)
3301

    
3302
    if not utils.IsBelowDir(pathutils.EXPORT_DIR, real_filename):
3303
      _Fail("File '%s' is not under exports directory '%s': %s",
3304
            filename, pathutils.EXPORT_DIR, real_filename)
3305

    
3306
    # Create directory
3307
    utils.Makedirs(directory, mode=0750)
3308

    
3309
    quoted_filename = utils.ShellQuote(filename)
3310

    
3311
    if mode == constants.IEM_IMPORT:
3312
      suffix = "> %s" % quoted_filename
3313
    elif mode == constants.IEM_EXPORT:
3314
      suffix = "< %s" % quoted_filename
3315

    
3316
      # Retrieve file size
3317
      try:
3318
        st = os.stat(filename)
3319
      except EnvironmentError, err:
3320
        logging.error("Can't stat(2) %s: %s", filename, err)
3321
      else:
3322
        exp_size = utils.BytesToMebibyte(st.st_size)
3323

    
3324
  elif ieio == constants.IEIO_RAW_DISK:
3325
    (disk, ) = ieargs
3326

    
3327
    real_disk = _OpenRealBD(disk)
3328

    
3329
    if mode == constants.IEM_IMPORT:
3330
      # we set here a smaller block size as, due to transport buffering, more
3331
      # than 64-128k will mostly ignored; we use nocreat to fail if the device
3332
      # is not already there or we pass a wrong path; we use notrunc to no
3333
      # attempt truncate on an LV device; we use oflag=dsync to not buffer too
3334
      # much memory; this means that at best, we flush every 64k, which will
3335
      # not be very fast
3336
      suffix = utils.BuildShellCmd(("| dd of=%s conv=nocreat,notrunc"
3337
                                    " bs=%s oflag=dsync"),
3338
                                    real_disk.dev_path,
3339
                                    str(64 * 1024))
3340

    
3341
    elif mode == constants.IEM_EXPORT:
3342
      # the block size on the read dd is 1MiB to match our units
3343
      prefix = utils.BuildShellCmd("dd if=%s bs=%s count=%s |",
3344
                                   real_disk.dev_path,
3345
                                   str(1024 * 1024), # 1 MB
3346
                                   str(disk.size))
3347
      exp_size = disk.size
3348

    
3349
  elif ieio == constants.IEIO_SCRIPT:
3350
    (disk, disk_index, ) = ieargs
3351

    
3352
    assert isinstance(disk_index, (int, long))
3353

    
3354
    real_disk = _OpenRealBD(disk)
3355

    
3356
    inst_os = OSFromDisk(instance.os)
3357
    env = OSEnvironment(instance, inst_os)
3358

    
3359
    if mode == constants.IEM_IMPORT:
3360
      env["IMPORT_DEVICE"] = env["DISK_%d_PATH" % disk_index]
3361
      env["IMPORT_INDEX"] = str(disk_index)
3362
      script = inst_os.import_script
3363

    
3364
    elif mode == constants.IEM_EXPORT:
3365
      env["EXPORT_DEVICE"] = real_disk.dev_path
3366
      env["EXPORT_INDEX"] = str(disk_index)
3367
      script = inst_os.export_script
3368

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

    
3372
    if mode == constants.IEM_IMPORT:
3373
      suffix = "| %s" % script_cmd
3374

    
3375
    elif mode == constants.IEM_EXPORT:
3376
      prefix = "%s |" % script_cmd
3377

    
3378
    # Let script predict size
3379
    exp_size = constants.IE_CUSTOM_SIZE
3380

    
3381
  else:
3382
    _Fail("Invalid %s I/O mode %r", mode, ieio)
3383

    
3384
  return (env, prefix, suffix, exp_size)
3385

    
3386

    
3387
def _CreateImportExportStatusDir(prefix):
3388
  """Creates status directory for import/export.
3389

3390
  """
3391
  return tempfile.mkdtemp(dir=pathutils.IMPORT_EXPORT_DIR,
3392
                          prefix=("%s-%s-" %
3393
                                  (prefix, utils.TimestampForFilename())))
3394

    
3395

    
3396
def StartImportExportDaemon(mode, opts, host, port, instance, component,
3397
                            ieio, ieioargs):
3398
  """Starts an import or export daemon.
3399

3400
  @param mode: Import/output mode
3401
  @type opts: L{objects.ImportExportOptions}
3402
  @param opts: Daemon options
3403
  @type host: string
3404
  @param host: Remote host for export (None for import)
3405
  @type port: int
3406
  @param port: Remote port for export (None for import)
3407
  @type instance: L{objects.Instance}
3408
  @param instance: Instance object
3409
  @type component: string
3410
  @param component: which part of the instance is transferred now,
3411
      e.g. 'disk/0'
3412
  @param ieio: Input/output type
3413
  @param ieioargs: Input/output arguments
3414

3415
  """
3416
  if mode == constants.IEM_IMPORT:
3417
    prefix = "import"
3418

    
3419
    if not (host is None and port is None):
3420
      _Fail("Can not specify host or port on import")
3421

    
3422
  elif mode == constants.IEM_EXPORT:
3423
    prefix = "export"
3424

    
3425
    if host is None or port is None:
3426
      _Fail("Host and port must be specified for an export")
3427

    
3428
  else:
3429
    _Fail("Invalid mode %r", mode)
3430

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

    
3434
  (cmd_env, cmd_prefix, cmd_suffix, exp_size) = \
3435
    _GetImportExportIoCommand(instance, mode, ieio, ieioargs)
3436

    
3437
  if opts.key_name is None:
3438
    # Use server.pem
3439
    key_path = pathutils.NODED_CERT_FILE
3440
    cert_path = pathutils.NODED_CERT_FILE
3441
    assert opts.ca_pem is None
3442
  else:
3443
    (_, key_path, cert_path) = _GetX509Filenames(pathutils.CRYPTO_KEYS_DIR,
3444
                                                 opts.key_name)
3445
    assert opts.ca_pem is not None
3446

    
3447
  for i in [key_path, cert_path]:
3448
    if not os.path.exists(i):
3449
      _Fail("File '%s' does not exist" % i)
3450

    
3451
  status_dir = _CreateImportExportStatusDir("%s-%s" % (prefix, component))
3452
  try:
3453
    status_file = utils.PathJoin(status_dir, _IES_STATUS_FILE)
3454
    pid_file = utils.PathJoin(status_dir, _IES_PID_FILE)
3455
    ca_file = utils.PathJoin(status_dir, _IES_CA_FILE)
3456

    
3457
    if opts.ca_pem is None:
3458
      # Use server.pem
3459
      ca = utils.ReadFile(pathutils.NODED_CERT_FILE)
3460
    else:
3461
      ca = opts.ca_pem
3462

    
3463
    # Write CA file
3464
    utils.WriteFile(ca_file, data=ca, mode=0400)
3465

    
3466
    cmd = [
3467
      pathutils.IMPORT_EXPORT_DAEMON,
3468
      status_file, mode,
3469
      "--key=%s" % key_path,
3470
      "--cert=%s" % cert_path,
3471
      "--ca=%s" % ca_file,
3472
      ]
3473

    
3474
    if host:
3475
      cmd.append("--host=%s" % host)
3476

    
3477
    if port:
3478
      cmd.append("--port=%s" % port)
3479

    
3480
    if opts.ipv6:
3481
      cmd.append("--ipv6")
3482
    else:
3483
      cmd.append("--ipv4")
3484

    
3485
    if opts.compress:
3486
      cmd.append("--compress=%s" % opts.compress)
3487

    
3488
    if opts.magic:
3489
      cmd.append("--magic=%s" % opts.magic)
3490

    
3491
    if exp_size is not None:
3492
      cmd.append("--expected-size=%s" % exp_size)
3493

    
3494
    if cmd_prefix:
3495
      cmd.append("--cmd-prefix=%s" % cmd_prefix)
3496

    
3497
    if cmd_suffix:
3498
      cmd.append("--cmd-suffix=%s" % cmd_suffix)
3499

    
3500
    if mode == constants.IEM_EXPORT:
3501
      # Retry connection a few times when connecting to remote peer
3502
      cmd.append("--connect-retries=%s" % constants.RIE_CONNECT_RETRIES)
3503
      cmd.append("--connect-timeout=%s" % constants.RIE_CONNECT_ATTEMPT_TIMEOUT)
3504
    elif opts.connect_timeout is not None:
3505
      assert mode == constants.IEM_IMPORT
3506
      # Overall timeout for establishing connection while listening
3507
      cmd.append("--connect-timeout=%s" % opts.connect_timeout)
3508

    
3509
    logfile = _InstanceLogName(prefix, instance.os, instance.name, component)
3510

    
3511
    # TODO: Once _InstanceLogName uses tempfile.mkstemp, StartDaemon has
3512
    # support for receiving a file descriptor for output
3513
    utils.StartDaemon(cmd, env=cmd_env, pidfile=pid_file,
3514
                      output=logfile)
3515

    
3516
    # The import/export name is simply the status directory name
3517
    return os.path.basename(status_dir)
3518

    
3519
  except Exception:
3520
    shutil.rmtree(status_dir, ignore_errors=True)
3521
    raise
3522

    
3523

    
3524
def GetImportExportStatus(names):
3525
  """Returns import/export daemon status.
3526

3527
  @type names: sequence
3528
  @param names: List of names
3529
  @rtype: List of dicts
3530
  @return: Returns a list of the state of each named import/export or None if a
3531
           status couldn't be read
3532

3533
  """
3534
  result = []
3535

    
3536
  for name in names:
3537
    status_file = utils.PathJoin(pathutils.IMPORT_EXPORT_DIR, name,
3538
                                 _IES_STATUS_FILE)
3539

    
3540
    try:
3541
      data = utils.ReadFile(status_file)
3542
    except EnvironmentError, err:
3543
      if err.errno != errno.ENOENT:
3544
        raise
3545
      data = None
3546

    
3547
    if not data:
3548
      result.append(None)
3549
      continue
3550

    
3551
    result.append(serializer.LoadJson(data))
3552

    
3553
  return result
3554

    
3555

    
3556
def AbortImportExport(name):
3557
  """Sends SIGTERM to a running import/export daemon.
3558

3559
  """
3560
  logging.info("Abort import/export %s", name)
3561

    
3562
  status_dir = utils.PathJoin(pathutils.IMPORT_EXPORT_DIR, name)
3563
  pid = utils.ReadLockedPidFile(utils.PathJoin(status_dir, _IES_PID_FILE))
3564

    
3565
  if pid:
3566
    logging.info("Import/export %s is running with PID %s, sending SIGTERM",
3567
                 name, pid)
3568
    utils.IgnoreProcessNotFound(os.kill, pid, signal.SIGTERM)
3569

    
3570

    
3571
def CleanupImportExport(name):
3572
  """Cleanup after an import or export.
3573

3574
  If the import/export daemon is still running it's killed. Afterwards the
3575
  whole status directory is removed.
3576

3577
  """
3578
  logging.info("Finalizing import/export %s", name)
3579

    
3580
  status_dir = utils.PathJoin(pathutils.IMPORT_EXPORT_DIR, name)
3581

    
3582
  pid = utils.ReadLockedPidFile(utils.PathJoin(status_dir, _IES_PID_FILE))
3583

    
3584
  if pid:
3585
    logging.info("Import/export %s is still running with PID %s",
3586
                 name, pid)
3587
    utils.KillProcess(pid, waitpid=False)
3588

    
3589
  shutil.rmtree(status_dir, ignore_errors=True)
3590

    
3591

    
3592
def _FindDisks(nodes_ip, disks):
3593
  """Sets the physical ID on disks and returns the block devices.
3594

3595
  """
3596
  # set the correct physical ID
3597
  my_name = netutils.Hostname.GetSysName()
3598
  for cf in disks:
3599
    cf.SetPhysicalID(my_name, nodes_ip)
3600

    
3601
  bdevs = []
3602

    
3603
  for cf in disks:
3604
    rd = _RecursiveFindBD(cf)
3605
    if rd is None:
3606
      _Fail("Can't find device %s", cf)
3607
    bdevs.append(rd)
3608
  return bdevs
3609

    
3610

    
3611
def DrbdDisconnectNet(nodes_ip, disks):
3612
  """Disconnects the network on a list of drbd devices.
3613

3614
  """
3615
  bdevs = _FindDisks(nodes_ip, disks)
3616

    
3617
  # disconnect disks
3618
  for rd in bdevs:
3619
    try:
3620
      rd.DisconnectNet()
3621
    except errors.BlockDeviceError, err:
3622
      _Fail("Can't change network configuration to standalone mode: %s",
3623
            err, exc=True)
3624

    
3625

    
3626
def DrbdAttachNet(nodes_ip, disks, instance_name, multimaster):
3627
  """Attaches the network on a list of drbd devices.
3628

3629
  """
3630
  bdevs = _FindDisks(nodes_ip, disks)
3631

    
3632
  if multimaster:
3633
    for idx, rd in enumerate(bdevs):
3634
      try:
3635
        _SymlinkBlockDev(instance_name, rd.dev_path, idx)
3636
      except EnvironmentError, err:
3637
        _Fail("Can't create symlink: %s", err)
3638
  # reconnect disks, switch to new master configuration and if
3639
  # needed primary mode
3640
  for rd in bdevs:
3641
    try:
3642
      rd.AttachNet(multimaster)
3643
    except errors.BlockDeviceError, err:
3644
      _Fail("Can't change network configuration: %s", err)
3645

    
3646
  # wait until the disks are connected; we need to retry the re-attach
3647
  # if the device becomes standalone, as this might happen if the one
3648
  # node disconnects and reconnects in a different mode before the
3649
  # other node reconnects; in this case, one or both of the nodes will
3650
  # decide it has wrong configuration and switch to standalone
3651

    
3652
  def _Attach():
3653
    all_connected = True
3654

    
3655
    for rd in bdevs:
3656
      stats = rd.GetProcStatus()
3657

    
3658
      if multimaster:
3659
        # In the multimaster case we have to wait explicitly until
3660
        # the resource is Connected and UpToDate/UpToDate, because
3661
        # we promote *both nodes* to primary directly afterwards.
3662
        # Being in resync is not enough, since there is a race during which we
3663
        # may promote a node with an Outdated disk to primary, effectively
3664
        # tearing down the connection.
3665
        all_connected = (all_connected and
3666
                         stats.is_connected and
3667
                         stats.is_disk_uptodate and
3668
                         stats.peer_disk_uptodate)
3669
      else:
3670
        all_connected = (all_connected and
3671
                         (stats.is_connected or stats.is_in_resync))
3672

    
3673
      if stats.is_standalone:
3674
        # peer had different config info and this node became
3675
        # standalone, even though this should not happen with the
3676
        # new staged way of changing disk configs
3677
        try:
3678
          rd.AttachNet(multimaster)
3679
        except errors.BlockDeviceError, err:
3680
          _Fail("Can't change network configuration: %s", err)
3681

    
3682
    if not all_connected:
3683
      raise utils.RetryAgain()
3684

    
3685
  try:
3686
    # Start with a delay of 100 miliseconds and go up to 5 seconds
3687
    utils.Retry(_Attach, (0.1, 1.5, 5.0), 2 * 60)
3688
  except utils.RetryTimeout:
3689
    _Fail("Timeout in disk reconnecting")
3690

    
3691
  if multimaster:
3692
    # change to primary mode
3693
    for rd in bdevs:
3694
      try:
3695
        rd.Open()
3696
      except errors.BlockDeviceError, err:
3697
        _Fail("Can't change to primary mode: %s", err)
3698

    
3699

    
3700
def DrbdWaitSync(nodes_ip, disks):
3701
  """Wait until DRBDs have synchronized.
3702

3703
  """
3704
  def _helper(rd):
3705
    stats = rd.GetProcStatus()
3706
    if not (stats.is_connected or stats.is_in_resync):
3707
      raise utils.RetryAgain()
3708
    return stats
3709

    
3710
  bdevs = _FindDisks(nodes_ip, disks)
3711

    
3712
  min_resync = 100
3713
  alldone = True
3714
  for rd in bdevs:
3715
    try:
3716
      # poll each second for 15 seconds
3717
      stats = utils.Retry(_helper, 1, 15, args=[rd])
3718
    except utils.RetryTimeout:
3719
      stats = rd.GetProcStatus()
3720
      # last check
3721
      if not (stats.is_connected or stats.is_in_resync):
3722
        _Fail("DRBD device %s is not in sync: stats=%s", rd, stats)
3723
    alldone = alldone and (not stats.is_in_resync)
3724
    if stats.sync_percent is not None:
3725
      min_resync = min(min_resync, stats.sync_percent)
3726

    
3727
  return (alldone, min_resync)
3728

    
3729

    
3730
def GetDrbdUsermodeHelper():
3731
  """Returns DRBD usermode helper currently configured.
3732

3733
  """
3734
  try:
3735
    return bdev.BaseDRBD.GetUsermodeHelper()
3736
  except errors.BlockDeviceError, err:
3737
    _Fail(str(err))
3738

    
3739

    
3740
def PowercycleNode(hypervisor_type):
3741
  """Hard-powercycle the node.
3742

3743
  Because we need to return first, and schedule the powercycle in the
3744
  background, we won't be able to report failures nicely.
3745

3746
  """
3747
  hyper = hypervisor.GetHypervisor(hypervisor_type)
3748
  try:
3749
    pid = os.fork()
3750
  except OSError:
3751
    # if we can't fork, we'll pretend that we're in the child process
3752
    pid = 0
3753
  if pid > 0:
3754
    return "Reboot scheduled in 5 seconds"
3755
  # ensure the child is running on ram
3756
  try:
3757
    utils.Mlockall()
3758
  except Exception: # pylint: disable=W0703
3759
    pass
3760
  time.sleep(5)
3761
  hyper.PowercycleNode()
3762

    
3763

    
3764
def _VerifyRestrictedCmdName(cmd):
3765
  """Verifies a restricted command name.
3766

3767
  @type cmd: string
3768
  @param cmd: Command name
3769
  @rtype: tuple; (boolean, string or None)
3770
  @return: The tuple's first element is the status; if C{False}, the second
3771
    element is an error message string, otherwise it's C{None}
3772

3773
  """
3774
  if not cmd.strip():
3775
    return (False, "Missing command name")
3776

    
3777
  if os.path.basename(cmd) != cmd:
3778
    return (False, "Invalid command name")
3779

    
3780
  if not constants.EXT_PLUGIN_MASK.match(cmd):
3781
    return (False, "Command name contains forbidden characters")
3782

    
3783
  return (True, None)
3784

    
3785

    
3786
def _CommonRestrictedCmdCheck(path, owner):
3787
  """Common checks for restricted command file system directories and files.
3788

3789
  @type path: string
3790
  @param path: Path to check
3791
  @param owner: C{None} or tuple containing UID and GID
3792
  @rtype: tuple; (boolean, string or C{os.stat} result)
3793
  @return: The tuple's first element is the status; if C{False}, the second
3794
    element is an error message string, otherwise it's the result of C{os.stat}
3795

3796
  """
3797
  if owner is None:
3798
    # Default to root as owner
3799
    owner = (0, 0)
3800

    
3801
  try:
3802
    st = os.stat(path)
3803
  except EnvironmentError, err:
3804
    return (False, "Can't stat(2) '%s': %s" % (path, err))
3805

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

    
3809
  if (st.st_uid, st.st_gid) != owner:
3810
    (owner_uid, owner_gid) = owner
3811
    return (False, "'%s' is not owned by %s:%s" % (path, owner_uid, owner_gid))
3812

    
3813
  return (True, st)
3814

    
3815

    
3816
def _VerifyRestrictedCmdDirectory(path, _owner=None):
3817
  """Verifies restricted command directory.
3818

3819
  @type path: string
3820
  @param path: Path to check
3821
  @rtype: tuple; (boolean, string or None)
3822
  @return: The tuple's first element is the status; if C{False}, the second
3823
    element is an error message string, otherwise it's C{None}
3824

3825
  """
3826
  (status, value) = _CommonRestrictedCmdCheck(path, _owner)
3827

    
3828
  if not status:
3829
    return (False, value)
3830

    
3831
  if not stat.S_ISDIR(value.st_mode):
3832
    return (False, "Path '%s' is not a directory" % path)
3833

    
3834
  return (True, None)
3835

    
3836

    
3837
def _VerifyRestrictedCmd(path, cmd, _owner=None):
3838
  """Verifies a whole restricted command and returns its executable filename.
3839

3840
  @type path: string
3841
  @param path: Directory containing restricted commands
3842
  @type cmd: string
3843
  @param cmd: Command name
3844
  @rtype: tuple; (boolean, string)
3845
  @return: The tuple's first element is the status; if C{False}, the second
3846
    element is an error message string, otherwise the second element is the
3847
    absolute path to the executable
3848

3849
  """
3850
  executable = utils.PathJoin(path, cmd)
3851

    
3852
  (status, msg) = _CommonRestrictedCmdCheck(executable, _owner)
3853

    
3854
  if not status:
3855
    return (False, msg)
3856

    
3857
  if not utils.IsExecutable(executable):
3858
    return (False, "access(2) thinks '%s' can't be executed" % executable)
3859

    
3860
  return (True, executable)
3861

    
3862

    
3863
def _PrepareRestrictedCmd(path, cmd,
3864
                          _verify_dir=_VerifyRestrictedCmdDirectory,
3865
                          _verify_name=_VerifyRestrictedCmdName,
3866
                          _verify_cmd=_VerifyRestrictedCmd):
3867
  """Performs a number of tests on a restricted command.
3868

3869
  @type path: string
3870
  @param path: Directory containing restricted commands
3871
  @type cmd: string
3872
  @param cmd: Command name
3873
  @return: Same as L{_VerifyRestrictedCmd}
3874

3875
  """
3876
  # Verify the directory first
3877
  (status, msg) = _verify_dir(path)
3878
  if status:
3879
    # Check command if everything was alright
3880
    (status, msg) = _verify_name(cmd)
3881

    
3882
  if not status:
3883
    return (False, msg)
3884

    
3885
  # Check actual executable
3886
  return _verify_cmd(path, cmd)
3887

    
3888

    
3889
def RunRestrictedCmd(cmd,
3890
                     _lock_timeout=_RCMD_LOCK_TIMEOUT,
3891
                     _lock_file=pathutils.RESTRICTED_COMMANDS_LOCK_FILE,
3892
                     _path=pathutils.RESTRICTED_COMMANDS_DIR,
3893
                     _sleep_fn=time.sleep,
3894
                     _prepare_fn=_PrepareRestrictedCmd,
3895
                     _runcmd_fn=utils.RunCmd,
3896
                     _enabled=constants.ENABLE_RESTRICTED_COMMANDS):
3897
  """Executes a restricted command after performing strict tests.
3898

3899
  @type cmd: string
3900
  @param cmd: Command name
3901
  @rtype: string
3902
  @return: Command output
3903
  @raise RPCFail: In case of an error
3904

3905
  """
3906
  logging.info("Preparing to run restricted command '%s'", cmd)
3907

    
3908
  if not _enabled:
3909
    _Fail("Restricted commands disabled at configure time")
3910

    
3911
  lock = None
3912
  try:
3913
    cmdresult = None
3914
    try:
3915
      lock = utils.FileLock.Open(_lock_file)
3916
      lock.Exclusive(blocking=True, timeout=_lock_timeout)
3917

    
3918
      (status, value) = _prepare_fn(_path, cmd)
3919

    
3920
      if status:
3921
        cmdresult = _runcmd_fn([value], env={}, reset_env=True,
3922
                               postfork_fn=lambda _: lock.Unlock())
3923
      else:
3924
        logging.error(value)
3925
    except Exception: # pylint: disable=W0703
3926
      # Keep original error in log
3927
      logging.exception("Caught exception")
3928

    
3929
    if cmdresult is None:
3930
      logging.info("Sleeping for %0.1f seconds before returning",
3931
                   _RCMD_INVALID_DELAY)
3932
      _sleep_fn(_RCMD_INVALID_DELAY)
3933

    
3934
      # Do not include original error message in returned error
3935
      _Fail("Executing command '%s' failed" % cmd)
3936
    elif cmdresult.failed or cmdresult.fail_reason:
3937
      _Fail("Restricted command '%s' failed: %s; output: %s",
3938
            cmd, cmdresult.fail_reason, cmdresult.output)
3939
    else:
3940
      return cmdresult.output
3941
  finally:
3942
    if lock is not None:
3943
      # Release lock at last
3944
      lock.Close()
3945
      lock = None
3946

    
3947

    
3948
def SetWatcherPause(until, _filename=pathutils.WATCHER_PAUSEFILE):
3949
  """Creates or removes the watcher pause file.
3950

3951
  @type until: None or number
3952
  @param until: Unix timestamp saying until when the watcher shouldn't run
3953

3954
  """
3955
  if until is None:
3956
    logging.info("Received request to no longer pause watcher")
3957
    utils.RemoveFile(_filename)
3958
  else:
3959
    logging.info("Received request to pause watcher until %s", until)
3960

    
3961
    if not ht.TNumber(until):
3962
      _Fail("Duration must be numeric")
3963

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

    
3966

    
3967
class HooksRunner(object):
3968
  """Hook runner.
3969

3970
  This class is instantiated on the node side (ganeti-noded) and not
3971
  on the master side.
3972

3973
  """
3974
  def __init__(self, hooks_base_dir=None):
3975
    """Constructor for hooks runner.
3976

3977
    @type hooks_base_dir: str or None
3978
    @param hooks_base_dir: if not None, this overrides the
3979
        L{pathutils.HOOKS_BASE_DIR} (useful for unittests)
3980

3981
    """
3982
    if hooks_base_dir is None:
3983
      hooks_base_dir = pathutils.HOOKS_BASE_DIR
3984
    # yeah, _BASE_DIR is not valid for attributes, we use it like a
3985
    # constant
3986
    self._BASE_DIR = hooks_base_dir # pylint: disable=C0103
3987

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

3991
    """
3992
    assert len(node_list) == 1
3993
    node = node_list[0]
3994
    _, myself = ssconf.GetMasterAndMyself()
3995
    assert node == myself
3996

    
3997
    results = self.RunHooks(hpath, phase, env)
3998

    
3999
    # Return values in the form expected by HooksMaster
4000
    return {node: (None, False, results)}
4001

    
4002
  def RunHooks(self, hpath, phase, env):
4003
    """Run the scripts in the hooks directory.
4004

4005
    @type hpath: str
4006
    @param hpath: the path to the hooks directory which
4007
        holds the scripts
4008
    @type phase: str
4009
    @param phase: either L{constants.HOOKS_PHASE_PRE} or
4010
        L{constants.HOOKS_PHASE_POST}
4011
    @type env: dict
4012
    @param env: dictionary with the environment for the hook
4013
    @rtype: list
4014
    @return: list of 3-element tuples:
4015
      - script path
4016
      - script result, either L{constants.HKR_SUCCESS} or
4017
        L{constants.HKR_FAIL}
4018
      - output of the script
4019

4020
    @raise errors.ProgrammerError: for invalid input
4021
        parameters
4022

4023
    """
4024
    if phase == constants.HOOKS_PHASE_PRE:
4025
      suffix = "pre"
4026
    elif phase == constants.HOOKS_PHASE_POST:
4027
      suffix = "post"
4028
    else:
4029
      _Fail("Unknown hooks phase '%s'", phase)
4030

    
4031
    subdir = "%s-%s.d" % (hpath, suffix)
4032
    dir_name = utils.PathJoin(self._BASE_DIR, subdir)
4033

    
4034
    results = []
4035

    
4036
    if not os.path.isdir(dir_name):
4037
      # for non-existing/non-dirs, we simply exit instead of logging a
4038
      # warning at every operation
4039
      return results
4040

    
4041
    runparts_results = utils.RunParts(dir_name, env=env, reset_env=True)
4042

    
4043
    for (relname, relstatus, runresult) in runparts_results:
4044
      if relstatus == constants.RUNPARTS_SKIP:
4045
        rrval = constants.HKR_SKIP
4046
        output = ""
4047
      elif relstatus == constants.RUNPARTS_ERR:
4048
        rrval = constants.HKR_FAIL
4049
        output = "Hook script execution error: %s" % runresult
4050
      elif relstatus == constants.RUNPARTS_RUN:
4051
        if runresult.failed:
4052
          rrval = constants.HKR_FAIL
4053
        else:
4054
          rrval = constants.HKR_SUCCESS
4055
        output = utils.SafeEncode(runresult.output.strip())
4056
      results.append(("%s/%s" % (subdir, relname), rrval, output))
4057

    
4058
    return results
4059

    
4060

    
4061
class IAllocatorRunner(object):
4062
  """IAllocator runner.
4063

4064
  This class is instantiated on the node side (ganeti-noded) and not on
4065
  the master side.
4066

4067
  """
4068
  @staticmethod
4069
  def Run(name, idata):
4070
    """Run an iallocator script.
4071

4072
    @type name: str
4073
    @param name: the iallocator script name
4074
    @type idata: str
4075
    @param idata: the allocator input data
4076

4077
    @rtype: tuple
4078
    @return: two element tuple of:
4079
       - status
4080
       - either error message or stdout of allocator (for success)
4081

4082
    """
4083
    alloc_script = utils.FindFile(name, constants.IALLOCATOR_SEARCH_PATH,
4084
                                  os.path.isfile)
4085
    if alloc_script is None:
4086
      _Fail("iallocator module '%s' not found in the search path", name)
4087

    
4088
    fd, fin_name = tempfile.mkstemp(prefix="ganeti-iallocator.")
4089
    try:
4090
      os.write(fd, idata)
4091
      os.close(fd)
4092
      result = utils.RunCmd([alloc_script, fin_name])
4093
      if result.failed:
4094
        _Fail("iallocator module '%s' failed: %s, output '%s'",
4095
              name, result.fail_reason, result.output)
4096
    finally:
4097
      os.unlink(fin_name)
4098

    
4099
    return result.stdout
4100

    
4101

    
4102
class DevCacheManager(object):
4103
  """Simple class for managing a cache of block device information.
4104

4105
  """
4106
  _DEV_PREFIX = "/dev/"
4107
  _ROOT_DIR = pathutils.BDEV_CACHE_DIR
4108

    
4109
  @classmethod
4110
  def _ConvertPath(cls, dev_path):
4111
    """Converts a /dev/name path to the cache file name.
4112

4113
    This replaces slashes with underscores and strips the /dev
4114
    prefix. It then returns the full path to the cache file.
4115

4116
    @type dev_path: str
4117
    @param dev_path: the C{/dev/} path name
4118
    @rtype: str
4119
    @return: the converted path name
4120

4121
    """