Statistics
| Branch: | Tag: | Revision:

root / lib / backend.py @ 1f350e0f

History | View | Annotate | Download (125.1 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 mcpu
64
from ganeti import compat
65
from ganeti import pathutils
66
from ganeti import vcluster
67
from ganeti import ht
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 = mcpu.HooksMaster(hook_opcode, hooks_path, nodes, hr.RunLocalHooks,
329
                            None, env_fn, logging.warning, cfg.GetClusterName(),
330
                            cfg.GetMasterNode())
331

    
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):
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
  @rtype: None
1339

1340
  """
1341
  running_instances = GetInstanceList([instance.hypervisor])
1342

    
1343
  if instance.name in running_instances:
1344
    logging.info("Instance %s already running, not starting", instance.name)
1345
    return
1346

    
1347
  try:
1348
    block_devices = _GatherAndLinkBlockDevs(instance)
1349
    hyper = hypervisor.GetHypervisor(instance.hypervisor)
1350
    hyper.StartInstance(instance, block_devices, startup_paused)
1351
  except errors.BlockDeviceError, err:
1352
    _Fail("Block device error: %s", err, exc=True)
1353
  except errors.HypervisorError, err:
1354
    _RemoveBlockDevLinks(instance.name, instance.disks)
1355
    _Fail("Hypervisor error: %s", err, exc=True)
1356

    
1357

    
1358
def InstanceShutdown(instance, timeout, reason, store_reason=True):
1359
  """Shut an instance down.
1360

1361
  @note: this functions uses polling with a hardcoded timeout.
1362

1363
  @type instance: L{objects.Instance}
1364
  @param instance: the instance object
1365
  @type timeout: integer
1366
  @param timeout: maximum timeout for soft shutdown
1367
  @type reason: list of reasons
1368
  @param reason: the reason trail for this shutdown
1369
  @type store_reason: boolean
1370
  @param store_reason: whether to store the shutdown reason trail on file
1371
  @rtype: None
1372

1373
  """
1374
  hv_name = instance.hypervisor
1375
  hyper = hypervisor.GetHypervisor(hv_name)
1376
  iname = instance.name
1377

    
1378
  if instance.name not in hyper.ListInstances():
1379
    logging.info("Instance %s not running, doing nothing", iname)
1380
    return
1381

    
1382
  class _TryShutdown:
1383
    def __init__(self):
1384
      self.tried_once = False
1385

    
1386
    def __call__(self):
1387
      if iname not in hyper.ListInstances():
1388
        return
1389

    
1390
      try:
1391
        hyper.StopInstance(instance, retry=self.tried_once)
1392
        if store_reason:
1393
          _StoreInstReasonTrail(instance.name, reason)
1394
      except errors.HypervisorError, err:
1395
        if iname not in hyper.ListInstances():
1396
          # if the instance is no longer existing, consider this a
1397
          # success and go to cleanup
1398
          return
1399

    
1400
        _Fail("Failed to stop instance %s: %s", iname, err)
1401

    
1402
      self.tried_once = True
1403

    
1404
      raise utils.RetryAgain()
1405

    
1406
  try:
1407
    utils.Retry(_TryShutdown(), 5, timeout)
1408
  except utils.RetryTimeout:
1409
    # the shutdown did not succeed
1410
    logging.error("Shutdown of '%s' unsuccessful, forcing", iname)
1411

    
1412
    try:
1413
      hyper.StopInstance(instance, force=True)
1414
    except errors.HypervisorError, err:
1415
      if iname in hyper.ListInstances():
1416
        # only raise an error if the instance still exists, otherwise
1417
        # the error could simply be "instance ... unknown"!
1418
        _Fail("Failed to force stop instance %s: %s", iname, err)
1419

    
1420
    time.sleep(1)
1421

    
1422
    if iname in hyper.ListInstances():
1423
      _Fail("Could not shutdown instance %s even by destroy", iname)
1424

    
1425
  try:
1426
    hyper.CleanupInstance(instance.name)
1427
  except errors.HypervisorError, err:
1428
    logging.warning("Failed to execute post-shutdown cleanup step: %s", err)
1429

    
1430
  _RemoveBlockDevLinks(iname, instance.disks)
1431

    
1432

    
1433
def InstanceReboot(instance, reboot_type, shutdown_timeout, reason):
1434
  """Reboot an instance.
1435

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

1455
  """
1456
  running_instances = GetInstanceList([instance.hypervisor])
1457

    
1458
  if instance.name not in running_instances:
1459
    _Fail("Cannot reboot instance %s that is not running", instance.name)
1460

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

    
1478

    
1479
def InstanceBalloonMemory(instance, memory):
1480
  """Resize an instance's memory.
1481

1482
  @type instance: L{objects.Instance}
1483
  @param instance: the instance object
1484
  @type memory: int
1485
  @param memory: new memory amount in MB
1486
  @rtype: None
1487

1488
  """
1489
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1490
  running = hyper.ListInstances()
1491
  if instance.name not in running:
1492
    logging.info("Instance %s is not running, cannot balloon", instance.name)
1493
    return
1494
  try:
1495
    hyper.BalloonInstanceMemory(instance, memory)
1496
  except errors.HypervisorError, err:
1497
    _Fail("Failed to balloon instance memory: %s", err, exc=True)
1498

    
1499

    
1500
def MigrationInfo(instance):
1501
  """Gather information about an instance to be migrated.
1502

1503
  @type instance: L{objects.Instance}
1504
  @param instance: the instance definition
1505

1506
  """
1507
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1508
  try:
1509
    info = hyper.MigrationInfo(instance)
1510
  except errors.HypervisorError, err:
1511
    _Fail("Failed to fetch migration information: %s", err, exc=True)
1512
  return info
1513

    
1514

    
1515
def AcceptInstance(instance, info, target):
1516
  """Prepare the node to accept an instance.
1517

1518
  @type instance: L{objects.Instance}
1519
  @param instance: the instance definition
1520
  @type info: string/data (opaque)
1521
  @param info: migration information, from the source node
1522
  @type target: string
1523
  @param target: target host (usually ip), on this node
1524

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

    
1535
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1536
  try:
1537
    hyper.AcceptInstance(instance, info, target)
1538
  except errors.HypervisorError, err:
1539
    if instance.disk_template in constants.DTS_EXT_MIRROR:
1540
      _RemoveBlockDevLinks(instance.name, instance.disks)
1541
    _Fail("Failed to accept instance: %s", err, exc=True)
1542

    
1543

    
1544
def FinalizeMigrationDst(instance, info, success):
1545
  """Finalize any preparation to accept an instance.
1546

1547
  @type instance: L{objects.Instance}
1548
  @param instance: the instance definition
1549
  @type info: string/data (opaque)
1550
  @param info: migration information, from the source node
1551
  @type success: boolean
1552
  @param success: whether the migration was a success or a failure
1553

1554
  """
1555
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1556
  try:
1557
    hyper.FinalizeMigrationDst(instance, info, success)
1558
  except errors.HypervisorError, err:
1559
    _Fail("Failed to finalize migration on the target node: %s", err, exc=True)
1560

    
1561

    
1562
def MigrateInstance(instance, target, live):
1563
  """Migrates an instance to another node.
1564

1565
  @type instance: L{objects.Instance}
1566
  @param instance: the instance definition
1567
  @type target: string
1568
  @param target: the target node name
1569
  @type live: boolean
1570
  @param live: whether the migration should be done live or not (the
1571
      interpretation of this parameter is left to the hypervisor)
1572
  @raise RPCFail: if migration fails for some reason
1573

1574
  """
1575
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1576

    
1577
  try:
1578
    hyper.MigrateInstance(instance, target, live)
1579
  except errors.HypervisorError, err:
1580
    _Fail("Failed to migrate instance: %s", err, exc=True)
1581

    
1582

    
1583
def FinalizeMigrationSource(instance, success, live):
1584
  """Finalize the instance migration on the source node.
1585

1586
  @type instance: L{objects.Instance}
1587
  @param instance: the instance definition of the migrated instance
1588
  @type success: bool
1589
  @param success: whether the migration succeeded or not
1590
  @type live: bool
1591
  @param live: whether the user requested a live migration or not
1592
  @raise RPCFail: If the execution fails for some reason
1593

1594
  """
1595
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1596

    
1597
  try:
1598
    hyper.FinalizeMigrationSource(instance, success, live)
1599
  except Exception, err:  # pylint: disable=W0703
1600
    _Fail("Failed to finalize the migration on the source node: %s", err,
1601
          exc=True)
1602

    
1603

    
1604
def GetMigrationStatus(instance):
1605
  """Get the migration status
1606

1607
  @type instance: L{objects.Instance}
1608
  @param instance: the instance that is being migrated
1609
  @rtype: L{objects.MigrationStatus}
1610
  @return: the status of the current migration (one of
1611
           L{constants.HV_MIGRATION_VALID_STATUSES}), plus any additional
1612
           progress info that can be retrieved from the hypervisor
1613
  @raise RPCFail: If the migration status cannot be retrieved
1614

1615
  """
1616
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1617
  try:
1618
    return hyper.GetMigrationStatus(instance)
1619
  except Exception, err:  # pylint: disable=W0703
1620
    _Fail("Failed to get migration status: %s", err, exc=True)
1621

    
1622

    
1623
def BlockdevCreate(disk, size, owner, on_primary, info, excl_stor):
1624
  """Creates a block device for an instance.
1625

1626
  @type disk: L{objects.Disk}
1627
  @param disk: the object describing the disk we should create
1628
  @type size: int
1629
  @param size: the size of the physical underlying device, in MiB
1630
  @type owner: str
1631
  @param owner: the name of the instance for which disk is created,
1632
      used for device cache data
1633
  @type on_primary: boolean
1634
  @param on_primary:  indicates if it is the primary node or not
1635
  @type info: string
1636
  @param info: string that will be sent to the physical device
1637
      creation, used for example to set (LVM) tags on LVs
1638
  @type excl_stor: boolean
1639
  @param excl_stor: Whether exclusive_storage is active
1640

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

1645
  """
1646
  # TODO: remove the obsolete "size" argument
1647
  # pylint: disable=W0613
1648
  clist = []
1649
  if disk.children:
1650
    for child in disk.children:
1651
      try:
1652
        crdev = _RecursiveAssembleBD(child, owner, on_primary)
1653
      except errors.BlockDeviceError, err:
1654
        _Fail("Can't assemble device %s: %s", child, err)
1655
      if on_primary or disk.AssembleOnSecondary():
1656
        # we need the children open in case the device itself has to
1657
        # be assembled
1658
        try:
1659
          # pylint: disable=E1103
1660
          crdev.Open()
1661
        except errors.BlockDeviceError, err:
1662
          _Fail("Can't make child '%s' read-write: %s", child, err)
1663
      clist.append(crdev)
1664

    
1665
  try:
1666
    device = bdev.Create(disk, clist, excl_stor)
1667
  except errors.BlockDeviceError, err:
1668
    _Fail("Can't create block device: %s", err)
1669

    
1670
  if on_primary or disk.AssembleOnSecondary():
1671
    try:
1672
      device.Assemble()
1673
    except errors.BlockDeviceError, err:
1674
      _Fail("Can't assemble device after creation, unusual event: %s", err)
1675
    if on_primary or disk.OpenOnSecondary():
1676
      try:
1677
        device.Open(force=True)
1678
      except errors.BlockDeviceError, err:
1679
        _Fail("Can't make device r/w after creation, unusual event: %s", err)
1680
    DevCacheManager.UpdateCache(device.dev_path, owner,
1681
                                on_primary, disk.iv_name)
1682

    
1683
  device.SetInfo(info)
1684

    
1685
  return device.unique_id
1686

    
1687

    
1688
def _WipeDevice(path, offset, size):
1689
  """This function actually wipes the device.
1690

1691
  @param path: The path to the device to wipe
1692
  @param offset: The offset in MiB in the file
1693
  @param size: The size in MiB to write
1694

1695
  """
1696
  # Internal sizes are always in Mebibytes; if the following "dd" command
1697
  # should use a different block size the offset and size given to this
1698
  # function must be adjusted accordingly before being passed to "dd".
1699
  block_size = 1024 * 1024
1700

    
1701
  cmd = [constants.DD_CMD, "if=/dev/zero", "seek=%d" % offset,
1702
         "bs=%s" % block_size, "oflag=direct", "of=%s" % path,
1703
         "count=%d" % size]
1704
  result = utils.RunCmd(cmd)
1705

    
1706
  if result.failed:
1707
    _Fail("Wipe command '%s' exited with error: %s; output: %s", result.cmd,
1708
          result.fail_reason, result.output)
1709

    
1710

    
1711
def BlockdevWipe(disk, offset, size):
1712
  """Wipes a block device.
1713

1714
  @type disk: L{objects.Disk}
1715
  @param disk: the disk object we want to wipe
1716
  @type offset: int
1717
  @param offset: The offset in MiB in the file
1718
  @type size: int
1719
  @param size: The size in MiB to write
1720

1721
  """
1722
  try:
1723
    rdev = _RecursiveFindBD(disk)
1724
  except errors.BlockDeviceError:
1725
    rdev = None
1726

    
1727
  if not rdev:
1728
    _Fail("Cannot execute wipe for device %s: device not found", disk.iv_name)
1729

    
1730
  # Do cross verify some of the parameters
1731
  if offset < 0:
1732
    _Fail("Negative offset")
1733
  if size < 0:
1734
    _Fail("Negative size")
1735
  if offset > rdev.size:
1736
    _Fail("Offset is bigger than device size")
1737
  if (offset + size) > rdev.size:
1738
    _Fail("The provided offset and size to wipe is bigger than device size")
1739

    
1740
  _WipeDevice(rdev.dev_path, offset, size)
1741

    
1742

    
1743
def BlockdevPauseResumeSync(disks, pause):
1744
  """Pause or resume the sync of the block device.
1745

1746
  @type disks: list of L{objects.Disk}
1747
  @param disks: the disks object we want to pause/resume
1748
  @type pause: bool
1749
  @param pause: Wheater to pause or resume
1750

1751
  """
1752
  success = []
1753
  for disk in disks:
1754
    try:
1755
      rdev = _RecursiveFindBD(disk)
1756
    except errors.BlockDeviceError:
1757
      rdev = None
1758

    
1759
    if not rdev:
1760
      success.append((False, ("Cannot change sync for device %s:"
1761
                              " device not found" % disk.iv_name)))
1762
      continue
1763

    
1764
    result = rdev.PauseResumeSync(pause)
1765

    
1766
    if result:
1767
      success.append((result, None))
1768
    else:
1769
      if pause:
1770
        msg = "Pause"
1771
      else:
1772
        msg = "Resume"
1773
      success.append((result, "%s for device %s failed" % (msg, disk.iv_name)))
1774

    
1775
  return success
1776

    
1777

    
1778
def BlockdevRemove(disk):
1779
  """Remove a block device.
1780

1781
  @note: This is intended to be called recursively.
1782

1783
  @type disk: L{objects.Disk}
1784
  @param disk: the disk object we should remove
1785
  @rtype: boolean
1786
  @return: the success of the operation
1787

1788
  """
1789
  msgs = []
1790
  try:
1791
    rdev = _RecursiveFindBD(disk)
1792
  except errors.BlockDeviceError, err:
1793
    # probably can't attach
1794
    logging.info("Can't attach to device %s in remove", disk)
1795
    rdev = None
1796
  if rdev is not None:
1797
    r_path = rdev.dev_path
1798
    try:
1799
      rdev.Remove()
1800
    except errors.BlockDeviceError, err:
1801
      msgs.append(str(err))
1802
    if not msgs:
1803
      DevCacheManager.RemoveCache(r_path)
1804

    
1805
  if disk.children:
1806
    for child in disk.children:
1807
      try:
1808
        BlockdevRemove(child)
1809
      except RPCFail, err:
1810
        msgs.append(str(err))
1811

    
1812
  if msgs:
1813
    _Fail("; ".join(msgs))
1814

    
1815

    
1816
def _RecursiveAssembleBD(disk, owner, as_primary):
1817
  """Activate a block device for an instance.
1818

1819
  This is run on the primary and secondary nodes for an instance.
1820

1821
  @note: this function is called recursively.
1822

1823
  @type disk: L{objects.Disk}
1824
  @param disk: the disk we try to assemble
1825
  @type owner: str
1826
  @param owner: the name of the instance which owns the disk
1827
  @type as_primary: boolean
1828
  @param as_primary: if we should make the block device
1829
      read/write
1830

1831
  @return: the assembled device or None (in case no device
1832
      was assembled)
1833
  @raise errors.BlockDeviceError: in case there is an error
1834
      during the activation of the children or the device
1835
      itself
1836

1837
  """
1838
  children = []
1839
  if disk.children:
1840
    mcn = disk.ChildrenNeeded()
1841
    if mcn == -1:
1842
      mcn = 0 # max number of Nones allowed
1843
    else:
1844
      mcn = len(disk.children) - mcn # max number of Nones
1845
    for chld_disk in disk.children:
1846
      try:
1847
        cdev = _RecursiveAssembleBD(chld_disk, owner, as_primary)
1848
      except errors.BlockDeviceError, err:
1849
        if children.count(None) >= mcn:
1850
          raise
1851
        cdev = None
1852
        logging.error("Error in child activation (but continuing): %s",
1853
                      str(err))
1854
      children.append(cdev)
1855

    
1856
  if as_primary or disk.AssembleOnSecondary():
1857
    r_dev = bdev.Assemble(disk, children)
1858
    result = r_dev
1859
    if as_primary or disk.OpenOnSecondary():
1860
      r_dev.Open()
1861
    DevCacheManager.UpdateCache(r_dev.dev_path, owner,
1862
                                as_primary, disk.iv_name)
1863

    
1864
  else:
1865
    result = True
1866
  return result
1867

    
1868

    
1869
def BlockdevAssemble(disk, owner, as_primary, idx):
1870
  """Activate a block device for an instance.
1871

1872
  This is a wrapper over _RecursiveAssembleBD.
1873

1874
  @rtype: str or boolean
1875
  @return: a C{/dev/...} path for primary nodes, and
1876
      C{True} for secondary nodes
1877

1878
  """
1879
  try:
1880
    result = _RecursiveAssembleBD(disk, owner, as_primary)
1881
    if isinstance(result, bdev.BlockDev):
1882
      # pylint: disable=E1103
1883
      result = result.dev_path
1884
      if as_primary:
1885
        _SymlinkBlockDev(owner, result, idx)
1886
  except errors.BlockDeviceError, err:
1887
    _Fail("Error while assembling disk: %s", err, exc=True)
1888
  except OSError, err:
1889
    _Fail("Error while symlinking disk: %s", err, exc=True)
1890

    
1891
  return result
1892

    
1893

    
1894
def BlockdevShutdown(disk):
1895
  """Shut down a block device.
1896

1897
  First, if the device is assembled (Attach() is successful), then
1898
  the device is shutdown. Then the children of the device are
1899
  shutdown.
1900

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

1905
  @type disk: L{objects.Disk}
1906
  @param disk: the description of the disk we should
1907
      shutdown
1908
  @rtype: None
1909

1910
  """
1911
  msgs = []
1912
  r_dev = _RecursiveFindBD(disk)
1913
  if r_dev is not None:
1914
    r_path = r_dev.dev_path
1915
    try:
1916
      r_dev.Shutdown()
1917
      DevCacheManager.RemoveCache(r_path)
1918
    except errors.BlockDeviceError, err:
1919
      msgs.append(str(err))
1920

    
1921
  if disk.children:
1922
    for child in disk.children:
1923
      try:
1924
        BlockdevShutdown(child)
1925
      except RPCFail, err:
1926
        msgs.append(str(err))
1927

    
1928
  if msgs:
1929
    _Fail("; ".join(msgs))
1930

    
1931

    
1932
def BlockdevAddchildren(parent_cdev, new_cdevs):
1933
  """Extend a mirrored block device.
1934

1935
  @type parent_cdev: L{objects.Disk}
1936
  @param parent_cdev: the disk to which we should add children
1937
  @type new_cdevs: list of L{objects.Disk}
1938
  @param new_cdevs: the list of children which we should add
1939
  @rtype: None
1940

1941
  """
1942
  parent_bdev = _RecursiveFindBD(parent_cdev)
1943
  if parent_bdev is None:
1944
    _Fail("Can't find parent device '%s' in add children", parent_cdev)
1945
  new_bdevs = [_RecursiveFindBD(disk) for disk in new_cdevs]
1946
  if new_bdevs.count(None) > 0:
1947
    _Fail("Can't find new device(s) to add: %s:%s", new_bdevs, new_cdevs)
1948
  parent_bdev.AddChildren(new_bdevs)
1949

    
1950

    
1951
def BlockdevRemovechildren(parent_cdev, new_cdevs):
1952
  """Shrink a mirrored block device.
1953

1954
  @type parent_cdev: L{objects.Disk}
1955
  @param parent_cdev: the disk from which we should remove children
1956
  @type new_cdevs: list of L{objects.Disk}
1957
  @param new_cdevs: the list of children which we should remove
1958
  @rtype: None
1959

1960
  """
1961
  parent_bdev = _RecursiveFindBD(parent_cdev)
1962
  if parent_bdev is None:
1963
    _Fail("Can't find parent device '%s' in remove children", parent_cdev)
1964
  devs = []
1965
  for disk in new_cdevs:
1966
    rpath = disk.StaticDevPath()
1967
    if rpath is None:
1968
      bd = _RecursiveFindBD(disk)
1969
      if bd is None:
1970
        _Fail("Can't find device %s while removing children", disk)
1971
      else:
1972
        devs.append(bd.dev_path)
1973
    else:
1974
      if not utils.IsNormAbsPath(rpath):
1975
        _Fail("Strange path returned from StaticDevPath: '%s'", rpath)
1976
      devs.append(rpath)
1977
  parent_bdev.RemoveChildren(devs)
1978

    
1979

    
1980
def BlockdevGetmirrorstatus(disks):
1981
  """Get the mirroring status of a list of devices.
1982

1983
  @type disks: list of L{objects.Disk}
1984
  @param disks: the list of disks which we should query
1985
  @rtype: disk
1986
  @return: List of L{objects.BlockDevStatus}, one for each disk
1987
  @raise errors.BlockDeviceError: if any of the disks cannot be
1988
      found
1989

1990
  """
1991
  stats = []
1992
  for dsk in disks:
1993
    rbd = _RecursiveFindBD(dsk)
1994
    if rbd is None:
1995
      _Fail("Can't find device %s", dsk)
1996

    
1997
    stats.append(rbd.CombinedSyncStatus())
1998

    
1999
  return stats
2000

    
2001

    
2002
def BlockdevGetmirrorstatusMulti(disks):
2003
  """Get the mirroring status of a list of devices.
2004

2005
  @type disks: list of L{objects.Disk}
2006
  @param disks: the list of disks which we should query
2007
  @rtype: disk
2008
  @return: List of tuples, (bool, status), one for each disk; bool denotes
2009
    success/failure, status is L{objects.BlockDevStatus} on success, string
2010
    otherwise
2011

2012
  """
2013
  result = []
2014
  for disk in disks:
2015
    try:
2016
      rbd = _RecursiveFindBD(disk)
2017
      if rbd is None:
2018
        result.append((False, "Can't find device %s" % disk))
2019
        continue
2020

    
2021
      status = rbd.CombinedSyncStatus()
2022
    except errors.BlockDeviceError, err:
2023
      logging.exception("Error while getting disk status")
2024
      result.append((False, str(err)))
2025
    else:
2026
      result.append((True, status))
2027

    
2028
  assert len(disks) == len(result)
2029

    
2030
  return result
2031

    
2032

    
2033
def _RecursiveFindBD(disk):
2034
  """Check if a device is activated.
2035

2036
  If so, return information about the real device.
2037

2038
  @type disk: L{objects.Disk}
2039
  @param disk: the disk object we need to find
2040

2041
  @return: None if the device can't be found,
2042
      otherwise the device instance
2043

2044
  """
2045
  children = []
2046
  if disk.children:
2047
    for chdisk in disk.children:
2048
      children.append(_RecursiveFindBD(chdisk))
2049

    
2050
  return bdev.FindDevice(disk, children)
2051

    
2052

    
2053
def _OpenRealBD(disk):
2054
  """Opens the underlying block device of a disk.
2055

2056
  @type disk: L{objects.Disk}
2057
  @param disk: the disk object we want to open
2058

2059
  """
2060
  real_disk = _RecursiveFindBD(disk)
2061
  if real_disk is None:
2062
    _Fail("Block device '%s' is not set up", disk)
2063

    
2064
  real_disk.Open()
2065

    
2066
  return real_disk
2067

    
2068

    
2069
def BlockdevFind(disk):
2070
  """Check if a device is activated.
2071

2072
  If it is, return information about the real device.
2073

2074
  @type disk: L{objects.Disk}
2075
  @param disk: the disk to find
2076
  @rtype: None or objects.BlockDevStatus
2077
  @return: None if the disk cannot be found, otherwise a the current
2078
           information
2079

2080
  """
2081
  try:
2082
    rbd = _RecursiveFindBD(disk)
2083
  except errors.BlockDeviceError, err:
2084
    _Fail("Failed to find device: %s", err, exc=True)
2085

    
2086
  if rbd is None:
2087
    return None
2088

    
2089
  return rbd.GetSyncStatus()
2090

    
2091

    
2092
def BlockdevGetsize(disks):
2093
  """Computes the size of the given disks.
2094

2095
  If a disk is not found, returns None instead.
2096

2097
  @type disks: list of L{objects.Disk}
2098
  @param disks: the list of disk to compute the size for
2099
  @rtype: list
2100
  @return: list with elements None if the disk cannot be found,
2101
      otherwise the size
2102

2103
  """
2104
  result = []
2105
  for cf in disks:
2106
    try:
2107
      rbd = _RecursiveFindBD(cf)
2108
    except errors.BlockDeviceError:
2109
      result.append(None)
2110
      continue
2111
    if rbd is None:
2112
      result.append(None)
2113
    else:
2114
      result.append(rbd.GetActualSize())
2115
  return result
2116

    
2117

    
2118
def BlockdevExport(disk, dest_node, dest_path, cluster_name):
2119
  """Export a block device to a remote node.
2120

2121
  @type disk: L{objects.Disk}
2122
  @param disk: the description of the disk to export
2123
  @type dest_node: str
2124
  @param dest_node: the destination node to export to
2125
  @type dest_path: str
2126
  @param dest_path: the destination path on the target node
2127
  @type cluster_name: str
2128
  @param cluster_name: the cluster name, needed for SSH hostalias
2129
  @rtype: None
2130

2131
  """
2132
  real_disk = _OpenRealBD(disk)
2133

    
2134
  # the block size on the read dd is 1MiB to match our units
2135
  expcmd = utils.BuildShellCmd("set -e; set -o pipefail; "
2136
                               "dd if=%s bs=1048576 count=%s",
2137
                               real_disk.dev_path, str(disk.size))
2138

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

    
2148
  remotecmd = _GetSshRunner(cluster_name).BuildCmd(dest_node,
2149
                                                   constants.SSH_LOGIN_USER,
2150
                                                   destcmd)
2151

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

    
2155
  result = utils.RunCmd(["bash", "-c", command])
2156

    
2157
  if result.failed:
2158
    _Fail("Disk copy command '%s' returned error: %s"
2159
          " output: %s", command, result.fail_reason, result.output)
2160

    
2161

    
2162
def UploadFile(file_name, data, mode, uid, gid, atime, mtime):
2163
  """Write a file to the filesystem.
2164

2165
  This allows the master to overwrite(!) a file. It will only perform
2166
  the operation if the file belongs to a list of configuration files.
2167

2168
  @type file_name: str
2169
  @param file_name: the target file name
2170
  @type data: str
2171
  @param data: the new contents of the file
2172
  @type mode: int
2173
  @param mode: the mode to give the file (can be None)
2174
  @type uid: string
2175
  @param uid: the owner of the file
2176
  @type gid: string
2177
  @param gid: the group of the file
2178
  @type atime: float
2179
  @param atime: the atime to set on the file (can be None)
2180
  @type mtime: float
2181
  @param mtime: the mtime to set on the file (can be None)
2182
  @rtype: None
2183

2184
  """
2185
  file_name = vcluster.LocalizeVirtualPath(file_name)
2186

    
2187
  if not os.path.isabs(file_name):
2188
    _Fail("Filename passed to UploadFile is not absolute: '%s'", file_name)
2189

    
2190
  if file_name not in _ALLOWED_UPLOAD_FILES:
2191
    _Fail("Filename passed to UploadFile not in allowed upload targets: '%s'",
2192
          file_name)
2193

    
2194
  raw_data = _Decompress(data)
2195

    
2196
  if not (isinstance(uid, basestring) and isinstance(gid, basestring)):
2197
    _Fail("Invalid username/groupname type")
2198

    
2199
  getents = runtime.GetEnts()
2200
  uid = getents.LookupUser(uid)
2201
  gid = getents.LookupGroup(gid)
2202

    
2203
  utils.SafeWriteFile(file_name, None,
2204
                      data=raw_data, mode=mode, uid=uid, gid=gid,
2205
                      atime=atime, mtime=mtime)
2206

    
2207

    
2208
def RunOob(oob_program, command, node, timeout):
2209
  """Executes oob_program with given command on given node.
2210

2211
  @param oob_program: The path to the executable oob_program
2212
  @param command: The command to invoke on oob_program
2213
  @param node: The node given as an argument to the program
2214
  @param timeout: Timeout after which we kill the oob program
2215

2216
  @return: stdout
2217
  @raise RPCFail: If execution fails for some reason
2218

2219
  """
2220
  result = utils.RunCmd([oob_program, command, node], timeout=timeout)
2221

    
2222
  if result.failed:
2223
    _Fail("'%s' failed with reason '%s'; output: %s", result.cmd,
2224
          result.fail_reason, result.output)
2225

    
2226
  return result.stdout
2227

    
2228

    
2229
def _OSOndiskAPIVersion(os_dir):
2230
  """Compute and return the API version of a given OS.
2231

2232
  This function will try to read the API version of the OS residing in
2233
  the 'os_dir' directory.
2234

2235
  @type os_dir: str
2236
  @param os_dir: the directory in which we should look for the OS
2237
  @rtype: tuple
2238
  @return: tuple (status, data) with status denoting the validity and
2239
      data holding either the vaid versions or an error message
2240

2241
  """
2242
  api_file = utils.PathJoin(os_dir, constants.OS_API_FILE)
2243

    
2244
  try:
2245
    st = os.stat(api_file)
2246
  except EnvironmentError, err:
2247
    return False, ("Required file '%s' not found under path %s: %s" %
2248
                   (constants.OS_API_FILE, os_dir, utils.ErrnoOrStr(err)))
2249

    
2250
  if not stat.S_ISREG(stat.S_IFMT(st.st_mode)):
2251
    return False, ("File '%s' in %s is not a regular file" %
2252
                   (constants.OS_API_FILE, os_dir))
2253

    
2254
  try:
2255
    api_versions = utils.ReadFile(api_file).splitlines()
2256
  except EnvironmentError, err:
2257
    return False, ("Error while reading the API version file at %s: %s" %
2258
                   (api_file, utils.ErrnoOrStr(err)))
2259

    
2260
  try:
2261
    api_versions = [int(version.strip()) for version in api_versions]
2262
  except (TypeError, ValueError), err:
2263
    return False, ("API version(s) can't be converted to integer: %s" %
2264
                   str(err))
2265

    
2266
  return True, api_versions
2267

    
2268

    
2269
def DiagnoseOS(top_dirs=None):
2270
  """Compute the validity for all OSes.
2271

2272
  @type top_dirs: list
2273
  @param top_dirs: the list of directories in which to
2274
      search (if not given defaults to
2275
      L{pathutils.OS_SEARCH_PATH})
2276
  @rtype: list of L{objects.OS}
2277
  @return: a list of tuples (name, path, status, diagnose, variants,
2278
      parameters, api_version) for all (potential) OSes under all
2279
      search paths, where:
2280
          - name is the (potential) OS name
2281
          - path is the full path to the OS
2282
          - status True/False is the validity of the OS
2283
          - diagnose is the error message for an invalid OS, otherwise empty
2284
          - variants is a list of supported OS variants, if any
2285
          - parameters is a list of (name, help) parameters, if any
2286
          - api_version is a list of support OS API versions
2287

2288
  """
2289
  if top_dirs is None:
2290
    top_dirs = pathutils.OS_SEARCH_PATH
2291

    
2292
  result = []
2293
  for dir_name in top_dirs:
2294
    if os.path.isdir(dir_name):
2295
      try:
2296
        f_names = utils.ListVisibleFiles(dir_name)
2297
      except EnvironmentError, err:
2298
        logging.exception("Can't list the OS directory %s: %s", dir_name, err)
2299
        break
2300
      for name in f_names:
2301
        os_path = utils.PathJoin(dir_name, name)
2302
        status, os_inst = _TryOSFromDisk(name, base_dir=dir_name)
2303
        if status:
2304
          diagnose = ""
2305
          variants = os_inst.supported_variants
2306
          parameters = os_inst.supported_parameters
2307
          api_versions = os_inst.api_versions
2308
        else:
2309
          diagnose = os_inst
2310
          variants = parameters = api_versions = []
2311
        result.append((name, os_path, status, diagnose, variants,
2312
                       parameters, api_versions))
2313

    
2314
  return result
2315

    
2316

    
2317
def _TryOSFromDisk(name, base_dir=None):
2318
  """Create an OS instance from disk.
2319

2320
  This function will return an OS instance if the given name is a
2321
  valid OS name.
2322

2323
  @type base_dir: string
2324
  @keyword base_dir: Base directory containing OS installations.
2325
                     Defaults to a search in all the OS_SEARCH_PATH dirs.
2326
  @rtype: tuple
2327
  @return: success and either the OS instance if we find a valid one,
2328
      or error message
2329

2330
  """
2331
  if base_dir is None:
2332
    os_dir = utils.FindFile(name, pathutils.OS_SEARCH_PATH, os.path.isdir)
2333
  else:
2334
    os_dir = utils.FindFile(name, [base_dir], os.path.isdir)
2335

    
2336
  if os_dir is None:
2337
    return False, "Directory for OS %s not found in search path" % name
2338

    
2339
  status, api_versions = _OSOndiskAPIVersion(os_dir)
2340
  if not status:
2341
    # push the error up
2342
    return status, api_versions
2343

    
2344
  if not constants.OS_API_VERSIONS.intersection(api_versions):
2345
    return False, ("API version mismatch for path '%s': found %s, want %s." %
2346
                   (os_dir, api_versions, constants.OS_API_VERSIONS))
2347

    
2348
  # OS Files dictionary, we will populate it with the absolute path
2349
  # names; if the value is True, then it is a required file, otherwise
2350
  # an optional one
2351
  os_files = dict.fromkeys(constants.OS_SCRIPTS, True)
2352

    
2353
  if max(api_versions) >= constants.OS_API_V15:
2354
    os_files[constants.OS_VARIANTS_FILE] = False
2355

    
2356
  if max(api_versions) >= constants.OS_API_V20:
2357
    os_files[constants.OS_PARAMETERS_FILE] = True
2358
  else:
2359
    del os_files[constants.OS_SCRIPT_VERIFY]
2360

    
2361
  for (filename, required) in os_files.items():
2362
    os_files[filename] = utils.PathJoin(os_dir, filename)
2363

    
2364
    try:
2365
      st = os.stat(os_files[filename])
2366
    except EnvironmentError, err:
2367
      if err.errno == errno.ENOENT and not required:
2368
        del os_files[filename]
2369
        continue
2370
      return False, ("File '%s' under path '%s' is missing (%s)" %
2371
                     (filename, os_dir, utils.ErrnoOrStr(err)))
2372

    
2373
    if not stat.S_ISREG(stat.S_IFMT(st.st_mode)):
2374
      return False, ("File '%s' under path '%s' is not a regular file" %
2375
                     (filename, os_dir))
2376

    
2377
    if filename in constants.OS_SCRIPTS:
2378
      if stat.S_IMODE(st.st_mode) & stat.S_IXUSR != stat.S_IXUSR:
2379
        return False, ("File '%s' under path '%s' is not executable" %
2380
                       (filename, os_dir))
2381

    
2382
  variants = []
2383
  if constants.OS_VARIANTS_FILE in os_files:
2384
    variants_file = os_files[constants.OS_VARIANTS_FILE]
2385
    try:
2386
      variants = \
2387
        utils.FilterEmptyLinesAndComments(utils.ReadFile(variants_file))
2388
    except EnvironmentError, err:
2389
      # we accept missing files, but not other errors
2390
      if err.errno != errno.ENOENT:
2391
        return False, ("Error while reading the OS variants file at %s: %s" %
2392
                       (variants_file, utils.ErrnoOrStr(err)))
2393

    
2394
  parameters = []
2395
  if constants.OS_PARAMETERS_FILE in os_files:
2396
    parameters_file = os_files[constants.OS_PARAMETERS_FILE]
2397
    try:
2398
      parameters = utils.ReadFile(parameters_file).splitlines()
2399
    except EnvironmentError, err:
2400
      return False, ("Error while reading the OS parameters file at %s: %s" %
2401
                     (parameters_file, utils.ErrnoOrStr(err)))
2402
    parameters = [v.split(None, 1) for v in parameters]
2403

    
2404
  os_obj = objects.OS(name=name, path=os_dir,
2405
                      create_script=os_files[constants.OS_SCRIPT_CREATE],
2406
                      export_script=os_files[constants.OS_SCRIPT_EXPORT],
2407
                      import_script=os_files[constants.OS_SCRIPT_IMPORT],
2408
                      rename_script=os_files[constants.OS_SCRIPT_RENAME],
2409
                      verify_script=os_files.get(constants.OS_SCRIPT_VERIFY,
2410
                                                 None),
2411
                      supported_variants=variants,
2412
                      supported_parameters=parameters,
2413
                      api_versions=api_versions)
2414
  return True, os_obj
2415

    
2416

    
2417
def OSFromDisk(name, base_dir=None):
2418
  """Create an OS instance from disk.
2419

2420
  This function will return an OS instance if the given name is a
2421
  valid OS name. Otherwise, it will raise an appropriate
2422
  L{RPCFail} exception, detailing why this is not a valid OS.
2423

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

2427
  @type base_dir: string
2428
  @keyword base_dir: Base directory containing OS installations.
2429
                     Defaults to a search in all the OS_SEARCH_PATH dirs.
2430
  @rtype: L{objects.OS}
2431
  @return: the OS instance if we find a valid one
2432
  @raise RPCFail: if we don't find a valid OS
2433

2434
  """
2435
  name_only = objects.OS.GetName(name)
2436
  status, payload = _TryOSFromDisk(name_only, base_dir)
2437

    
2438
  if not status:
2439
    _Fail(payload)
2440

    
2441
  return payload
2442

    
2443

    
2444
def OSCoreEnv(os_name, inst_os, os_params, debug=0):
2445
  """Calculate the basic environment for an os script.
2446

2447
  @type os_name: str
2448
  @param os_name: full operating system name (including variant)
2449
  @type inst_os: L{objects.OS}
2450
  @param inst_os: operating system for which the environment is being built
2451
  @type os_params: dict
2452
  @param os_params: the OS parameters
2453
  @type debug: integer
2454
  @param debug: debug level (0 or 1, for OS Api 10)
2455
  @rtype: dict
2456
  @return: dict of environment variables
2457
  @raise errors.BlockDeviceError: if the block device
2458
      cannot be found
2459

2460
  """
2461
  result = {}
2462
  api_version = \
2463
    max(constants.OS_API_VERSIONS.intersection(inst_os.api_versions))
2464
  result["OS_API_VERSION"] = "%d" % api_version
2465
  result["OS_NAME"] = inst_os.name
2466
  result["DEBUG_LEVEL"] = "%d" % debug
2467

    
2468
  # OS variants
2469
  if api_version >= constants.OS_API_V15 and inst_os.supported_variants:
2470
    variant = objects.OS.GetVariant(os_name)
2471
    if not variant:
2472
      variant = inst_os.supported_variants[0]
2473
  else:
2474
    variant = ""
2475
  result["OS_VARIANT"] = variant
2476

    
2477
  # OS params
2478
  for pname, pvalue in os_params.items():
2479
    result["OSP_%s" % pname.upper()] = pvalue
2480

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

    
2486
  return result
2487

    
2488

    
2489
def OSEnvironment(instance, inst_os, debug=0):
2490
  """Calculate the environment for an os script.
2491

2492
  @type instance: L{objects.Instance}
2493
  @param instance: target instance for the os script run
2494
  @type inst_os: L{objects.OS}
2495
  @param inst_os: operating system for which the environment is being built
2496
  @type debug: integer
2497
  @param debug: debug level (0 or 1, for OS Api 10)
2498
  @rtype: dict
2499
  @return: dict of environment variables
2500
  @raise errors.BlockDeviceError: if the block device
2501
      cannot be found
2502

2503
  """
2504
  result = OSCoreEnv(instance.os, inst_os, instance.osparams, debug=debug)
2505

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

    
2509
  result["HYPERVISOR"] = instance.hypervisor
2510
  result["DISK_COUNT"] = "%d" % len(instance.disks)
2511
  result["NIC_COUNT"] = "%d" % len(instance.nics)
2512
  result["INSTANCE_SECONDARY_NODES"] = \
2513
      ("%s" % " ".join(instance.secondary_nodes))
2514

    
2515
  # Disks
2516
  for idx, disk in enumerate(instance.disks):
2517
    real_disk = _OpenRealBD(disk)
2518
    result["DISK_%d_PATH" % idx] = real_disk.dev_path
2519
    result["DISK_%d_ACCESS" % idx] = disk.mode
2520
    if constants.HV_DISK_TYPE in instance.hvparams:
2521
      result["DISK_%d_FRONTEND_TYPE" % idx] = \
2522
        instance.hvparams[constants.HV_DISK_TYPE]
2523
    if disk.dev_type in constants.LDS_BLOCK:
2524
      result["DISK_%d_BACKEND_TYPE" % idx] = "block"
2525
    elif disk.dev_type == constants.LD_FILE:
2526
      result["DISK_%d_BACKEND_TYPE" % idx] = \
2527
        "file:%s" % disk.physical_id[0]
2528

    
2529
  # NICs
2530
  for idx, nic in enumerate(instance.nics):
2531
    result["NIC_%d_MAC" % idx] = nic.mac
2532
    if nic.ip:
2533
      result["NIC_%d_IP" % idx] = nic.ip
2534
    result["NIC_%d_MODE" % idx] = nic.nicparams[constants.NIC_MODE]
2535
    if nic.nicparams[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED:
2536
      result["NIC_%d_BRIDGE" % idx] = nic.nicparams[constants.NIC_LINK]
2537
    if nic.nicparams[constants.NIC_LINK]:
2538
      result["NIC_%d_LINK" % idx] = nic.nicparams[constants.NIC_LINK]
2539
    if nic.netinfo:
2540
      nobj = objects.Network.FromDict(nic.netinfo)
2541
      result.update(nobj.HooksDict("NIC_%d_" % idx))
2542
    if constants.HV_NIC_TYPE in instance.hvparams:
2543
      result["NIC_%d_FRONTEND_TYPE" % idx] = \
2544
        instance.hvparams[constants.HV_NIC_TYPE]
2545

    
2546
  # HV/BE params
2547
  for source, kind in [(instance.beparams, "BE"), (instance.hvparams, "HV")]:
2548
    for key, value in source.items():
2549
      result["INSTANCE_%s_%s" % (kind, key)] = str(value)
2550

    
2551
  return result
2552

    
2553

    
2554
def DiagnoseExtStorage(top_dirs=None):
2555
  """Compute the validity for all ExtStorage Providers.
2556

2557
  @type top_dirs: list
2558
  @param top_dirs: the list of directories in which to
2559
      search (if not given defaults to
2560
      L{pathutils.ES_SEARCH_PATH})
2561
  @rtype: list of L{objects.ExtStorage}
2562
  @return: a list of tuples (name, path, status, diagnose, parameters)
2563
      for all (potential) ExtStorage Providers under all
2564
      search paths, where:
2565
          - name is the (potential) ExtStorage Provider
2566
          - path is the full path to the ExtStorage Provider
2567
          - status True/False is the validity of the ExtStorage Provider
2568
          - diagnose is the error message for an invalid ExtStorage Provider,
2569
            otherwise empty
2570
          - parameters is a list of (name, help) parameters, if any
2571

2572
  """
2573
  if top_dirs is None:
2574
    top_dirs = pathutils.ES_SEARCH_PATH
2575

    
2576
  result = []
2577
  for dir_name in top_dirs:
2578
    if os.path.isdir(dir_name):
2579
      try:
2580
        f_names = utils.ListVisibleFiles(dir_name)
2581
      except EnvironmentError, err:
2582
        logging.exception("Can't list the ExtStorage directory %s: %s",
2583
                          dir_name, err)
2584
        break
2585
      for name in f_names:
2586
        es_path = utils.PathJoin(dir_name, name)
2587
        status, es_inst = bdev.ExtStorageFromDisk(name, base_dir=dir_name)
2588
        if status:
2589
          diagnose = ""
2590
          parameters = es_inst.supported_parameters
2591
        else:
2592
          diagnose = es_inst
2593
          parameters = []
2594
        result.append((name, es_path, status, diagnose, parameters))
2595

    
2596
  return result
2597

    
2598

    
2599
def BlockdevGrow(disk, amount, dryrun, backingstore):
2600
  """Grow a stack of block devices.
2601

2602
  This function is called recursively, with the childrens being the
2603
  first ones to resize.
2604

2605
  @type disk: L{objects.Disk}
2606
  @param disk: the disk to be grown
2607
  @type amount: integer
2608
  @param amount: the amount (in mebibytes) to grow with
2609
  @type dryrun: boolean
2610
  @param dryrun: whether to execute the operation in simulation mode
2611
      only, without actually increasing the size
2612
  @param backingstore: whether to execute the operation on backing storage
2613
      only, or on "logical" storage only; e.g. DRBD is logical storage,
2614
      whereas LVM, file, RBD are backing storage
2615
  @rtype: (status, result)
2616
  @return: a tuple with the status of the operation (True/False), and
2617
      the errors message if status is False
2618

2619
  """
2620
  r_dev = _RecursiveFindBD(disk)
2621
  if r_dev is None:
2622
    _Fail("Cannot find block device %s", disk)
2623

    
2624
  try:
2625
    r_dev.Grow(amount, dryrun, backingstore)
2626
  except errors.BlockDeviceError, err:
2627
    _Fail("Failed to grow block device: %s", err, exc=True)
2628

    
2629

    
2630
def BlockdevSnapshot(disk):
2631
  """Create a snapshot copy of a block device.
2632

2633
  This function is called recursively, and the snapshot is actually created
2634
  just for the leaf lvm backend device.
2635

2636
  @type disk: L{objects.Disk}
2637
  @param disk: the disk to be snapshotted
2638
  @rtype: string
2639
  @return: snapshot disk ID as (vg, lv)
2640

2641
  """
2642
  if disk.dev_type == constants.LD_DRBD8:
2643
    if not disk.children:
2644
      _Fail("DRBD device '%s' without backing storage cannot be snapshotted",
2645
            disk.unique_id)
2646
    return BlockdevSnapshot(disk.children[0])
2647
  elif disk.dev_type == constants.LD_LV:
2648
    r_dev = _RecursiveFindBD(disk)
2649
    if r_dev is not None:
2650
      # FIXME: choose a saner value for the snapshot size
2651
      # let's stay on the safe side and ask for the full size, for now
2652
      return r_dev.Snapshot(disk.size)
2653
    else:
2654
      _Fail("Cannot find block device %s", disk)
2655
  else:
2656
    _Fail("Cannot snapshot non-lvm block device '%s' of type '%s'",
2657
          disk.unique_id, disk.dev_type)
2658

    
2659

    
2660
def BlockdevSetInfo(disk, info):
2661
  """Sets 'metadata' information on block devices.
2662

2663
  This function sets 'info' metadata on block devices. Initial
2664
  information is set at device creation; this function should be used
2665
  for example after renames.
2666

2667
  @type disk: L{objects.Disk}
2668
  @param disk: the disk to be grown
2669
  @type info: string
2670
  @param info: new 'info' metadata
2671
  @rtype: (status, result)
2672
  @return: a tuple with the status of the operation (True/False), and
2673
      the errors message if status is False
2674

2675
  """
2676
  r_dev = _RecursiveFindBD(disk)
2677
  if r_dev is None:
2678
    _Fail("Cannot find block device %s", disk)
2679

    
2680
  try:
2681
    r_dev.SetInfo(info)
2682
  except errors.BlockDeviceError, err:
2683
    _Fail("Failed to set information on block device: %s", err, exc=True)
2684

    
2685

    
2686
def FinalizeExport(instance, snap_disks):
2687
  """Write out the export configuration information.
2688

2689
  @type instance: L{objects.Instance}
2690
  @param instance: the instance which we export, used for
2691
      saving configuration
2692
  @type snap_disks: list of L{objects.Disk}
2693
  @param snap_disks: list of snapshot block devices, which
2694
      will be used to get the actual name of the dump file
2695

2696
  @rtype: None
2697

2698
  """
2699
  destdir = utils.PathJoin(pathutils.EXPORT_DIR, instance.name + ".new")
2700
  finaldestdir = utils.PathJoin(pathutils.EXPORT_DIR, instance.name)
2701

    
2702
  config = objects.SerializableConfigParser()
2703

    
2704
  config.add_section(constants.INISECT_EXP)
2705
  config.set(constants.INISECT_EXP, "version", "0")
2706
  config.set(constants.INISECT_EXP, "timestamp", "%d" % int(time.time()))
2707
  config.set(constants.INISECT_EXP, "source", instance.primary_node)
2708
  config.set(constants.INISECT_EXP, "os", instance.os)
2709
  config.set(constants.INISECT_EXP, "compression", "none")
2710

    
2711
  config.add_section(constants.INISECT_INS)
2712
  config.set(constants.INISECT_INS, "name", instance.name)
2713
  config.set(constants.INISECT_INS, "maxmem", "%d" %
2714
             instance.beparams[constants.BE_MAXMEM])
2715
  config.set(constants.INISECT_INS, "minmem", "%d" %
2716
             instance.beparams[constants.BE_MINMEM])
2717
  # "memory" is deprecated, but useful for exporting to old ganeti versions
2718
  config.set(constants.INISECT_INS, "memory", "%d" %
2719
             instance.beparams[constants.BE_MAXMEM])
2720
  config.set(constants.INISECT_INS, "vcpus", "%d" %
2721
             instance.beparams[constants.BE_VCPUS])
2722
  config.set(constants.INISECT_INS, "disk_template", instance.disk_template)
2723
  config.set(constants.INISECT_INS, "hypervisor", instance.hypervisor)
2724
  config.set(constants.INISECT_INS, "tags", " ".join(instance.GetTags()))
2725

    
2726
  nic_total = 0
2727
  for nic_count, nic in enumerate(instance.nics):
2728
    nic_total += 1
2729
    config.set(constants.INISECT_INS, "nic%d_mac" %
2730
               nic_count, "%s" % nic.mac)
2731
    config.set(constants.INISECT_INS, "nic%d_ip" % nic_count, "%s" % nic.ip)
2732
    config.set(constants.INISECT_INS, "nic%d_network" % nic_count,
2733
               "%s" % nic.network)
2734
    for param in constants.NICS_PARAMETER_TYPES:
2735
      config.set(constants.INISECT_INS, "nic%d_%s" % (nic_count, param),
2736
                 "%s" % nic.nicparams.get(param, None))
2737
  # TODO: redundant: on load can read nics until it doesn't exist
2738
  config.set(constants.INISECT_INS, "nic_count", "%d" % nic_total)
2739

    
2740
  disk_total = 0
2741
  for disk_count, disk in enumerate(snap_disks):
2742
    if disk:
2743
      disk_total += 1
2744
      config.set(constants.INISECT_INS, "disk%d_ivname" % disk_count,
2745
                 ("%s" % disk.iv_name))
2746
      config.set(constants.INISECT_INS, "disk%d_dump" % disk_count,
2747
                 ("%s" % disk.physical_id[1]))
2748
      config.set(constants.INISECT_INS, "disk%d_size" % disk_count,
2749
                 ("%d" % disk.size))
2750

    
2751
  config.set(constants.INISECT_INS, "disk_count", "%d" % disk_total)
2752

    
2753
  # New-style hypervisor/backend parameters
2754

    
2755
  config.add_section(constants.INISECT_HYP)
2756
  for name, value in instance.hvparams.items():
2757
    if name not in constants.HVC_GLOBALS:
2758
      config.set(constants.INISECT_HYP, name, str(value))
2759

    
2760
  config.add_section(constants.INISECT_BEP)
2761
  for name, value in instance.beparams.items():
2762
    config.set(constants.INISECT_BEP, name, str(value))
2763

    
2764
  config.add_section(constants.INISECT_OSP)
2765
  for name, value in instance.osparams.items():
2766
    config.set(constants.INISECT_OSP, name, str(value))
2767

    
2768
  utils.WriteFile(utils.PathJoin(destdir, constants.EXPORT_CONF_FILE),
2769
                  data=config.Dumps())
2770
  shutil.rmtree(finaldestdir, ignore_errors=True)
2771
  shutil.move(destdir, finaldestdir)
2772

    
2773

    
2774
def ExportInfo(dest):
2775
  """Get export configuration information.
2776

2777
  @type dest: str
2778
  @param dest: directory containing the export
2779

2780
  @rtype: L{objects.SerializableConfigParser}
2781
  @return: a serializable config file containing the
2782
      export info
2783

2784
  """
2785
  cff = utils.PathJoin(dest, constants.EXPORT_CONF_FILE)
2786

    
2787
  config = objects.SerializableConfigParser()
2788
  config.read(cff)
2789

    
2790
  if (not config.has_section(constants.INISECT_EXP) or
2791
      not config.has_section(constants.INISECT_INS)):
2792
    _Fail("Export info file doesn't have the required fields")
2793

    
2794
  return config.Dumps()
2795

    
2796

    
2797
def ListExports():
2798
  """Return a list of exports currently available on this machine.
2799

2800
  @rtype: list
2801
  @return: list of the exports
2802

2803
  """
2804
  if os.path.isdir(pathutils.EXPORT_DIR):
2805
    return sorted(utils.ListVisibleFiles(pathutils.EXPORT_DIR))
2806
  else:
2807
    _Fail("No exports directory")
2808

    
2809

    
2810
def RemoveExport(export):
2811
  """Remove an existing export from the node.
2812

2813
  @type export: str
2814
  @param export: the name of the export to remove
2815
  @rtype: None
2816

2817
  """
2818
  target = utils.PathJoin(pathutils.EXPORT_DIR, export)
2819

    
2820
  try:
2821
    shutil.rmtree(target)
2822
  except EnvironmentError, err:
2823
    _Fail("Error while removing the export: %s", err, exc=True)
2824

    
2825

    
2826
def BlockdevRename(devlist):
2827
  """Rename a list of block devices.
2828

2829
  @type devlist: list of tuples
2830
  @param devlist: list of tuples of the form  (disk,
2831
      new_logical_id, new_physical_id); disk is an
2832
      L{objects.Disk} object describing the current disk,
2833
      and new logical_id/physical_id is the name we
2834
      rename it to
2835
  @rtype: boolean
2836
  @return: True if all renames succeeded, False otherwise
2837

2838
  """
2839
  msgs = []
2840
  result = True
2841
  for disk, unique_id in devlist:
2842
    dev = _RecursiveFindBD(disk)
2843
    if dev is None:
2844
      msgs.append("Can't find device %s in rename" % str(disk))
2845
      result = False
2846
      continue
2847
    try:
2848
      old_rpath = dev.dev_path
2849
      dev.Rename(unique_id)
2850
      new_rpath = dev.dev_path
2851
      if old_rpath != new_rpath:
2852
        DevCacheManager.RemoveCache(old_rpath)
2853
        # FIXME: we should add the new cache information here, like:
2854
        # DevCacheManager.UpdateCache(new_rpath, owner, ...)
2855
        # but we don't have the owner here - maybe parse from existing
2856
        # cache? for now, we only lose lvm data when we rename, which
2857
        # is less critical than DRBD or MD
2858
    except errors.BlockDeviceError, err:
2859
      msgs.append("Can't rename device '%s' to '%s': %s" %
2860
                  (dev, unique_id, err))
2861
      logging.exception("Can't rename device '%s' to '%s'", dev, unique_id)
2862
      result = False
2863
  if not result:
2864
    _Fail("; ".join(msgs))
2865

    
2866

    
2867
def _TransformFileStorageDir(fs_dir):
2868
  """Checks whether given file_storage_dir is valid.
2869

2870
  Checks wheter the given fs_dir is within the cluster-wide default
2871
  file_storage_dir or the shared_file_storage_dir, which are stored in
2872
  SimpleStore. Only paths under those directories are allowed.
2873

2874
  @type fs_dir: str
2875
  @param fs_dir: the path to check
2876

2877
  @return: the normalized path if valid, None otherwise
2878

2879
  """
2880
  if not (constants.ENABLE_FILE_STORAGE or
2881
          constants.ENABLE_SHARED_FILE_STORAGE):
2882
    _Fail("File storage disabled at configure time")
2883

    
2884
  bdev.CheckFileStoragePath(fs_dir)
2885

    
2886
  return os.path.normpath(fs_dir)
2887

    
2888

    
2889
def CreateFileStorageDir(file_storage_dir):
2890
  """Create file storage directory.
2891

2892
  @type file_storage_dir: str
2893
  @param file_storage_dir: directory to create
2894

2895
  @rtype: tuple
2896
  @return: tuple with first element a boolean indicating wheter dir
2897
      creation was successful or not
2898

2899
  """
2900
  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
2901
  if os.path.exists(file_storage_dir):
2902
    if not os.path.isdir(file_storage_dir):
2903
      _Fail("Specified storage dir '%s' is not a directory",
2904
            file_storage_dir)
2905
  else:
2906
    try:
2907
      os.makedirs(file_storage_dir, 0750)
2908
    except OSError, err:
2909
      _Fail("Cannot create file storage directory '%s': %s",
2910
            file_storage_dir, err, exc=True)
2911

    
2912

    
2913
def RemoveFileStorageDir(file_storage_dir):
2914
  """Remove file storage directory.
2915

2916
  Remove it only if it's empty. If not log an error and return.
2917

2918
  @type file_storage_dir: str
2919
  @param file_storage_dir: the directory we should cleanup
2920
  @rtype: tuple (success,)
2921
  @return: tuple of one element, C{success}, denoting
2922
      whether the operation was successful
2923

2924
  """
2925
  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
2926
  if os.path.exists(file_storage_dir):
2927
    if not os.path.isdir(file_storage_dir):
2928
      _Fail("Specified Storage directory '%s' is not a directory",
2929
            file_storage_dir)
2930
    # deletes dir only if empty, otherwise we want to fail the rpc call
2931
    try:
2932
      os.rmdir(file_storage_dir)
2933
    except OSError, err:
2934
      _Fail("Cannot remove file storage directory '%s': %s",
2935
            file_storage_dir, err)
2936

    
2937

    
2938
def RenameFileStorageDir(old_file_storage_dir, new_file_storage_dir):
2939
  """Rename the file storage directory.
2940

2941
  @type old_file_storage_dir: str
2942
  @param old_file_storage_dir: the current path
2943
  @type new_file_storage_dir: str
2944
  @param new_file_storage_dir: the name we should rename to
2945
  @rtype: tuple (success,)
2946
  @return: tuple of one element, C{success}, denoting
2947
      whether the operation was successful
2948

2949
  """
2950
  old_file_storage_dir = _TransformFileStorageDir(old_file_storage_dir)
2951
  new_file_storage_dir = _TransformFileStorageDir(new_file_storage_dir)
2952
  if not os.path.exists(new_file_storage_dir):
2953
    if os.path.isdir(old_file_storage_dir):
2954
      try:
2955
        os.rename(old_file_storage_dir, new_file_storage_dir)
2956
      except OSError, err:
2957
        _Fail("Cannot rename '%s' to '%s': %s",
2958
              old_file_storage_dir, new_file_storage_dir, err)
2959
    else:
2960
      _Fail("Specified storage dir '%s' is not a directory",
2961
            old_file_storage_dir)
2962
  else:
2963
    if os.path.exists(old_file_storage_dir):
2964
      _Fail("Cannot rename '%s' to '%s': both locations exist",
2965
            old_file_storage_dir, new_file_storage_dir)
2966

    
2967

    
2968
def _EnsureJobQueueFile(file_name):
2969
  """Checks whether the given filename is in the queue directory.
2970

2971
  @type file_name: str
2972
  @param file_name: the file name we should check
2973
  @rtype: None
2974
  @raises RPCFail: if the file is not valid
2975

2976
  """
2977
  if not utils.IsBelowDir(pathutils.QUEUE_DIR, file_name):
2978
    _Fail("Passed job queue file '%s' does not belong to"
2979
          " the queue directory '%s'", file_name, pathutils.QUEUE_DIR)
2980

    
2981

    
2982
def JobQueueUpdate(file_name, content):
2983
  """Updates a file in the queue directory.
2984

2985
  This is just a wrapper over L{utils.io.WriteFile}, with proper
2986
  checking.
2987

2988
  @type file_name: str
2989
  @param file_name: the job file name
2990
  @type content: str
2991
  @param content: the new job contents
2992
  @rtype: boolean
2993
  @return: the success of the operation
2994

2995
  """
2996
  file_name = vcluster.LocalizeVirtualPath(file_name)
2997

    
2998
  _EnsureJobQueueFile(file_name)
2999
  getents = runtime.GetEnts()
3000

    
3001
  # Write and replace the file atomically
3002
  utils.WriteFile(file_name, data=_Decompress(content), uid=getents.masterd_uid,
3003
                  gid=getents.daemons_gid, mode=constants.JOB_QUEUE_FILES_PERMS)
3004

    
3005

    
3006
def JobQueueRename(old, new):
3007
  """Renames a job queue file.
3008

3009
  This is just a wrapper over os.rename with proper checking.
3010

3011
  @type old: str
3012
  @param old: the old (actual) file name
3013
  @type new: str
3014
  @param new: the desired file name
3015
  @rtype: tuple
3016
  @return: the success of the operation and payload
3017

3018
  """
3019
  old = vcluster.LocalizeVirtualPath(old)
3020
  new = vcluster.LocalizeVirtualPath(new)
3021

    
3022
  _EnsureJobQueueFile(old)
3023
  _EnsureJobQueueFile(new)
3024

    
3025
  getents = runtime.GetEnts()
3026

    
3027
  utils.RenameFile(old, new, mkdir=True, mkdir_mode=0750,
3028
                   dir_uid=getents.masterd_uid, dir_gid=getents.daemons_gid)
3029

    
3030

    
3031
def BlockdevClose(instance_name, disks):
3032
  """Closes the given block devices.
3033

3034
  This means they will be switched to secondary mode (in case of
3035
  DRBD).
3036

3037
  @param instance_name: if the argument is not empty, the symlinks
3038
      of this instance will be removed
3039
  @type disks: list of L{objects.Disk}
3040
  @param disks: the list of disks to be closed
3041
  @rtype: tuple (success, message)
3042
  @return: a tuple of success and message, where success
3043
      indicates the succes of the operation, and message
3044
      which will contain the error details in case we
3045
      failed
3046

3047
  """
3048
  bdevs = []
3049
  for cf in disks:
3050
    rd = _RecursiveFindBD(cf)
3051
    if rd is None:
3052
      _Fail("Can't find device %s", cf)
3053
    bdevs.append(rd)
3054

    
3055
  msg = []
3056
  for rd in bdevs:
3057
    try:
3058
      rd.Close()
3059
    except errors.BlockDeviceError, err:
3060
      msg.append(str(err))
3061
  if msg:
3062
    _Fail("Can't make devices secondary: %s", ",".join(msg))
3063
  else:
3064
    if instance_name:
3065
      _RemoveBlockDevLinks(instance_name, disks)
3066

    
3067

    
3068
def ValidateHVParams(hvname, hvparams):
3069
  """Validates the given hypervisor parameters.
3070

3071
  @type hvname: string
3072
  @param hvname: the hypervisor name
3073
  @type hvparams: dict
3074
  @param hvparams: the hypervisor parameters to be validated
3075
  @rtype: None
3076

3077
  """
3078
  try:
3079
    hv_type = hypervisor.GetHypervisor(hvname)
3080
    hv_type.ValidateParameters(hvparams)
3081
  except errors.HypervisorError, err:
3082
    _Fail(str(err), log=False)
3083

    
3084

    
3085
def _CheckOSPList(os_obj, parameters):
3086
  """Check whether a list of parameters is supported by the OS.
3087

3088
  @type os_obj: L{objects.OS}
3089
  @param os_obj: OS object to check
3090
  @type parameters: list
3091
  @param parameters: the list of parameters to check
3092

3093
  """
3094
  supported = [v[0] for v in os_obj.supported_parameters]
3095
  delta = frozenset(parameters).difference(supported)
3096
  if delta:
3097
    _Fail("The following parameters are not supported"
3098
          " by the OS %s: %s" % (os_obj.name, utils.CommaJoin(delta)))
3099

    
3100

    
3101
def ValidateOS(required, osname, checks, osparams):
3102
  """Validate the given OS' parameters.
3103

3104
  @type required: boolean
3105
  @param required: whether absence of the OS should translate into
3106
      failure or not
3107
  @type osname: string
3108
  @param osname: the OS to be validated
3109
  @type checks: list
3110
  @param checks: list of the checks to run (currently only 'parameters')
3111
  @type osparams: dict
3112
  @param osparams: dictionary with OS parameters
3113
  @rtype: boolean
3114
  @return: True if the validation passed, or False if the OS was not
3115
      found and L{required} was false
3116

3117
  """
3118
  if not constants.OS_VALIDATE_CALLS.issuperset(checks):
3119
    _Fail("Unknown checks required for OS %s: %s", osname,
3120
          set(checks).difference(constants.OS_VALIDATE_CALLS))
3121

    
3122
  name_only = objects.OS.GetName(osname)
3123
  status, tbv = _TryOSFromDisk(name_only, None)
3124

    
3125
  if not status:
3126
    if required:
3127
      _Fail(tbv)
3128
    else:
3129
      return False
3130

    
3131
  if max(tbv.api_versions) < constants.OS_API_V20:
3132
    return True
3133

    
3134
  if constants.OS_VALIDATE_PARAMETERS in checks:
3135
    _CheckOSPList(tbv, osparams.keys())
3136

    
3137
  validate_env = OSCoreEnv(osname, tbv, osparams)
3138
  result = utils.RunCmd([tbv.verify_script] + checks, env=validate_env,
3139
                        cwd=tbv.path, reset_env=True)
3140
  if result.failed:
3141
    logging.error("os validate command '%s' returned error: %s output: %s",
3142
                  result.cmd, result.fail_reason, result.output)
3143
    _Fail("OS validation script failed (%s), output: %s",
3144
          result.fail_reason, result.output, log=False)
3145

    
3146
  return True
3147

    
3148

    
3149
def DemoteFromMC():
3150
  """Demotes the current node from master candidate role.
3151

3152
  """
3153
  # try to ensure we're not the master by mistake
3154
  master, myself = ssconf.GetMasterAndMyself()
3155
  if master == myself:
3156
    _Fail("ssconf status shows I'm the master node, will not demote")
3157

    
3158
  result = utils.RunCmd([pathutils.DAEMON_UTIL, "check", constants.MASTERD])
3159
  if not result.failed:
3160
    _Fail("The master daemon is running, will not demote")
3161

    
3162
  try:
3163
    if os.path.isfile(pathutils.CLUSTER_CONF_FILE):
3164
      utils.CreateBackup(pathutils.CLUSTER_CONF_FILE)
3165
  except EnvironmentError, err:
3166
    if err.errno != errno.ENOENT:
3167
      _Fail("Error while backing up cluster file: %s", err, exc=True)
3168

    
3169
  utils.RemoveFile(pathutils.CLUSTER_CONF_FILE)
3170

    
3171

    
3172
def _GetX509Filenames(cryptodir, name):
3173
  """Returns the full paths for the private key and certificate.
3174

3175
  """
3176
  return (utils.PathJoin(cryptodir, name),
3177
          utils.PathJoin(cryptodir, name, _X509_KEY_FILE),
3178
          utils.PathJoin(cryptodir, name, _X509_CERT_FILE))
3179

    
3180

    
3181
def CreateX509Certificate(validity, cryptodir=pathutils.CRYPTO_KEYS_DIR):
3182
  """Creates a new X509 certificate for SSL/TLS.
3183

3184
  @type validity: int
3185
  @param validity: Validity in seconds
3186
  @rtype: tuple; (string, string)
3187
  @return: Certificate name and public part
3188

3189
  """
3190
  (key_pem, cert_pem) = \
3191
    utils.GenerateSelfSignedX509Cert(netutils.Hostname.GetSysName(),
3192
                                     min(validity, _MAX_SSL_CERT_VALIDITY))
3193

    
3194
  cert_dir = tempfile.mkdtemp(dir=cryptodir,
3195
                              prefix="x509-%s-" % utils.TimestampForFilename())
3196
  try:
3197
    name = os.path.basename(cert_dir)
3198
    assert len(name) > 5
3199

    
3200
    (_, key_file, cert_file) = _GetX509Filenames(cryptodir, name)
3201

    
3202
    utils.WriteFile(key_file, mode=0400, data=key_pem)
3203
    utils.WriteFile(cert_file, mode=0400, data=cert_pem)
3204

    
3205
    # Never return private key as it shouldn't leave the node
3206
    return (name, cert_pem)
3207
  except Exception:
3208
    shutil.rmtree(cert_dir, ignore_errors=True)
3209
    raise
3210

    
3211

    
3212
def RemoveX509Certificate(name, cryptodir=pathutils.CRYPTO_KEYS_DIR):
3213
  """Removes a X509 certificate.
3214

3215
  @type name: string
3216
  @param name: Certificate name
3217

3218
  """
3219
  (cert_dir, key_file, cert_file) = _GetX509Filenames(cryptodir, name)
3220

    
3221
  utils.RemoveFile(key_file)
3222
  utils.RemoveFile(cert_file)
3223

    
3224
  try:
3225
    os.rmdir(cert_dir)
3226
  except EnvironmentError, err:
3227
    _Fail("Cannot remove certificate directory '%s': %s",
3228
          cert_dir, err)
3229

    
3230

    
3231
def _GetImportExportIoCommand(instance, mode, ieio, ieargs):
3232
  """Returns the command for the requested input/output.
3233

3234
  @type instance: L{objects.Instance}
3235
  @param instance: The instance object
3236
  @param mode: Import/export mode
3237
  @param ieio: Input/output type
3238
  @param ieargs: Input/output arguments
3239

3240
  """
3241
  assert mode in (constants.IEM_IMPORT, constants.IEM_EXPORT)
3242

    
3243
  env = None
3244
  prefix = None
3245
  suffix = None
3246
  exp_size = None
3247

    
3248
  if ieio == constants.IEIO_FILE:
3249
    (filename, ) = ieargs
3250

    
3251
    if not utils.IsNormAbsPath(filename):
3252
      _Fail("Path '%s' is not normalized or absolute", filename)
3253

    
3254
    real_filename = os.path.realpath(filename)
3255
    directory = os.path.dirname(real_filename)
3256

    
3257
    if not utils.IsBelowDir(pathutils.EXPORT_DIR, real_filename):
3258
      _Fail("File '%s' is not under exports directory '%s': %s",
3259
            filename, pathutils.EXPORT_DIR, real_filename)
3260

    
3261
    # Create directory
3262
    utils.Makedirs(directory, mode=0750)
3263

    
3264
    quoted_filename = utils.ShellQuote(filename)
3265

    
3266
    if mode == constants.IEM_IMPORT:
3267
      suffix = "> %s" % quoted_filename
3268
    elif mode == constants.IEM_EXPORT:
3269
      suffix = "< %s" % quoted_filename
3270

    
3271
      # Retrieve file size
3272
      try:
3273
        st = os.stat(filename)
3274
      except EnvironmentError, err:
3275
        logging.error("Can't stat(2) %s: %s", filename, err)
3276
      else:
3277
        exp_size = utils.BytesToMebibyte(st.st_size)
3278

    
3279
  elif ieio == constants.IEIO_RAW_DISK:
3280
    (disk, ) = ieargs
3281

    
3282
    real_disk = _OpenRealBD(disk)
3283

    
3284
    if mode == constants.IEM_IMPORT:
3285
      # we set here a smaller block size as, due to transport buffering, more
3286
      # than 64-128k will mostly ignored; we use nocreat to fail if the device
3287
      # is not already there or we pass a wrong path; we use notrunc to no
3288
      # attempt truncate on an LV device; we use oflag=dsync to not buffer too
3289
      # much memory; this means that at best, we flush every 64k, which will
3290
      # not be very fast
3291
      suffix = utils.BuildShellCmd(("| dd of=%s conv=nocreat,notrunc"
3292
                                    " bs=%s oflag=dsync"),
3293
                                    real_disk.dev_path,
3294
                                    str(64 * 1024))
3295

    
3296
    elif mode == constants.IEM_EXPORT:
3297
      # the block size on the read dd is 1MiB to match our units
3298
      prefix = utils.BuildShellCmd("dd if=%s bs=%s count=%s |",
3299
                                   real_disk.dev_path,
3300
                                   str(1024 * 1024), # 1 MB
3301
                                   str(disk.size))
3302
      exp_size = disk.size
3303

    
3304
  elif ieio == constants.IEIO_SCRIPT:
3305
    (disk, disk_index, ) = ieargs
3306

    
3307
    assert isinstance(disk_index, (int, long))
3308

    
3309
    real_disk = _OpenRealBD(disk)
3310

    
3311
    inst_os = OSFromDisk(instance.os)
3312
    env = OSEnvironment(instance, inst_os)
3313

    
3314
    if mode == constants.IEM_IMPORT:
3315
      env["IMPORT_DEVICE"] = env["DISK_%d_PATH" % disk_index]
3316
      env["IMPORT_INDEX"] = str(disk_index)
3317
      script = inst_os.import_script
3318

    
3319
    elif mode == constants.IEM_EXPORT:
3320
      env["EXPORT_DEVICE"] = real_disk.dev_path
3321
      env["EXPORT_INDEX"] = str(disk_index)
3322
      script = inst_os.export_script
3323

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

    
3327
    if mode == constants.IEM_IMPORT:
3328
      suffix = "| %s" % script_cmd
3329

    
3330
    elif mode == constants.IEM_EXPORT:
3331
      prefix = "%s |" % script_cmd
3332

    
3333
    # Let script predict size
3334
    exp_size = constants.IE_CUSTOM_SIZE
3335

    
3336
  else:
3337
    _Fail("Invalid %s I/O mode %r", mode, ieio)
3338

    
3339
  return (env, prefix, suffix, exp_size)
3340

    
3341

    
3342
def _CreateImportExportStatusDir(prefix):
3343
  """Creates status directory for import/export.
3344

3345
  """
3346
  return tempfile.mkdtemp(dir=pathutils.IMPORT_EXPORT_DIR,
3347
                          prefix=("%s-%s-" %
3348
                                  (prefix, utils.TimestampForFilename())))
3349

    
3350

    
3351
def StartImportExportDaemon(mode, opts, host, port, instance, component,
3352
                            ieio, ieioargs):
3353
  """Starts an import or export daemon.
3354

3355
  @param mode: Import/output mode
3356
  @type opts: L{objects.ImportExportOptions}
3357
  @param opts: Daemon options
3358
  @type host: string
3359
  @param host: Remote host for export (None for import)
3360
  @type port: int
3361
  @param port: Remote port for export (None for import)
3362
  @type instance: L{objects.Instance}
3363
  @param instance: Instance object
3364
  @type component: string
3365
  @param component: which part of the instance is transferred now,
3366
      e.g. 'disk/0'
3367
  @param ieio: Input/output type
3368
  @param ieioargs: Input/output arguments
3369

3370
  """
3371
  if mode == constants.IEM_IMPORT:
3372
    prefix = "import"
3373

    
3374
    if not (host is None and port is None):
3375
      _Fail("Can not specify host or port on import")
3376

    
3377
  elif mode == constants.IEM_EXPORT:
3378
    prefix = "export"
3379

    
3380
    if host is None or port is None:
3381
      _Fail("Host and port must be specified for an export")
3382

    
3383
  else:
3384
    _Fail("Invalid mode %r", mode)
3385

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

    
3389
  (cmd_env, cmd_prefix, cmd_suffix, exp_size) = \
3390
    _GetImportExportIoCommand(instance, mode, ieio, ieioargs)
3391

    
3392
  if opts.key_name is None:
3393
    # Use server.pem
3394
    key_path = pathutils.NODED_CERT_FILE
3395
    cert_path = pathutils.NODED_CERT_FILE
3396
    assert opts.ca_pem is None
3397
  else:
3398
    (_, key_path, cert_path) = _GetX509Filenames(pathutils.CRYPTO_KEYS_DIR,
3399
                                                 opts.key_name)
3400
    assert opts.ca_pem is not None
3401

    
3402
  for i in [key_path, cert_path]:
3403
    if not os.path.exists(i):
3404
      _Fail("File '%s' does not exist" % i)
3405

    
3406
  status_dir = _CreateImportExportStatusDir("%s-%s" % (prefix, component))
3407
  try:
3408
    status_file = utils.PathJoin(status_dir, _IES_STATUS_FILE)
3409
    pid_file = utils.PathJoin(status_dir, _IES_PID_FILE)
3410
    ca_file = utils.PathJoin(status_dir, _IES_CA_FILE)
3411

    
3412
    if opts.ca_pem is None:
3413
      # Use server.pem
3414
      ca = utils.ReadFile(pathutils.NODED_CERT_FILE)
3415
    else:
3416
      ca = opts.ca_pem
3417

    
3418
    # Write CA file
3419
    utils.WriteFile(ca_file, data=ca, mode=0400)
3420

    
3421
    cmd = [
3422
      pathutils.IMPORT_EXPORT_DAEMON,
3423
      status_file, mode,
3424
      "--key=%s" % key_path,
3425
      "--cert=%s" % cert_path,
3426
      "--ca=%s" % ca_file,
3427
      ]
3428

    
3429
    if host:
3430
      cmd.append("--host=%s" % host)
3431

    
3432
    if port:
3433
      cmd.append("--port=%s" % port)
3434

    
3435
    if opts.ipv6:
3436
      cmd.append("--ipv6")
3437
    else:
3438
      cmd.append("--ipv4")
3439

    
3440
    if opts.compress:
3441
      cmd.append("--compress=%s" % opts.compress)
3442

    
3443
    if opts.magic:
3444
      cmd.append("--magic=%s" % opts.magic)
3445

    
3446
    if exp_size is not None:
3447
      cmd.append("--expected-size=%s" % exp_size)
3448

    
3449
    if cmd_prefix:
3450
      cmd.append("--cmd-prefix=%s" % cmd_prefix)
3451

    
3452
    if cmd_suffix:
3453
      cmd.append("--cmd-suffix=%s" % cmd_suffix)
3454

    
3455
    if mode == constants.IEM_EXPORT:
3456
      # Retry connection a few times when connecting to remote peer
3457
      cmd.append("--connect-retries=%s" % constants.RIE_CONNECT_RETRIES)
3458
      cmd.append("--connect-timeout=%s" % constants.RIE_CONNECT_ATTEMPT_TIMEOUT)
3459
    elif opts.connect_timeout is not None:
3460
      assert mode == constants.IEM_IMPORT
3461
      # Overall timeout for establishing connection while listening
3462
      cmd.append("--connect-timeout=%s" % opts.connect_timeout)
3463

    
3464
    logfile = _InstanceLogName(prefix, instance.os, instance.name, component)
3465

    
3466
    # TODO: Once _InstanceLogName uses tempfile.mkstemp, StartDaemon has
3467
    # support for receiving a file descriptor for output
3468
    utils.StartDaemon(cmd, env=cmd_env, pidfile=pid_file,
3469
                      output=logfile)
3470

    
3471
    # The import/export name is simply the status directory name
3472
    return os.path.basename(status_dir)
3473

    
3474
  except Exception:
3475
    shutil.rmtree(status_dir, ignore_errors=True)
3476
    raise
3477

    
3478

    
3479
def GetImportExportStatus(names):
3480
  """Returns import/export daemon status.
3481

3482
  @type names: sequence
3483
  @param names: List of names
3484
  @rtype: List of dicts
3485
  @return: Returns a list of the state of each named import/export or None if a
3486
           status couldn't be read
3487

3488
  """
3489
  result = []
3490

    
3491
  for name in names:
3492
    status_file = utils.PathJoin(pathutils.IMPORT_EXPORT_DIR, name,
3493
                                 _IES_STATUS_FILE)
3494

    
3495
    try:
3496
      data = utils.ReadFile(status_file)
3497
    except EnvironmentError, err:
3498
      if err.errno != errno.ENOENT:
3499
        raise
3500
      data = None
3501

    
3502
    if not data:
3503
      result.append(None)
3504
      continue
3505

    
3506
    result.append(serializer.LoadJson(data))
3507

    
3508
  return result
3509

    
3510

    
3511
def AbortImportExport(name):
3512
  """Sends SIGTERM to a running import/export daemon.
3513

3514
  """
3515
  logging.info("Abort import/export %s", name)
3516

    
3517
  status_dir = utils.PathJoin(pathutils.IMPORT_EXPORT_DIR, name)
3518
  pid = utils.ReadLockedPidFile(utils.PathJoin(status_dir, _IES_PID_FILE))
3519

    
3520
  if pid:
3521
    logging.info("Import/export %s is running with PID %s, sending SIGTERM",
3522
                 name, pid)
3523
    utils.IgnoreProcessNotFound(os.kill, pid, signal.SIGTERM)
3524

    
3525

    
3526
def CleanupImportExport(name):
3527
  """Cleanup after an import or export.
3528

3529
  If the import/export daemon is still running it's killed. Afterwards the
3530
  whole status directory is removed.
3531

3532
  """
3533
  logging.info("Finalizing import/export %s", name)
3534

    
3535
  status_dir = utils.PathJoin(pathutils.IMPORT_EXPORT_DIR, name)
3536

    
3537
  pid = utils.ReadLockedPidFile(utils.PathJoin(status_dir, _IES_PID_FILE))
3538

    
3539
  if pid:
3540
    logging.info("Import/export %s is still running with PID %s",
3541
                 name, pid)
3542
    utils.KillProcess(pid, waitpid=False)
3543

    
3544
  shutil.rmtree(status_dir, ignore_errors=True)
3545

    
3546

    
3547
def _FindDisks(nodes_ip, disks):
3548
  """Sets the physical ID on disks and returns the block devices.
3549

3550
  """
3551
  # set the correct physical ID
3552
  my_name = netutils.Hostname.GetSysName()
3553
  for cf in disks:
3554
    cf.SetPhysicalID(my_name, nodes_ip)
3555

    
3556
  bdevs = []
3557

    
3558
  for cf in disks:
3559
    rd = _RecursiveFindBD(cf)
3560
    if rd is None:
3561
      _Fail("Can't find device %s", cf)
3562
    bdevs.append(rd)
3563
  return bdevs
3564

    
3565

    
3566
def DrbdDisconnectNet(nodes_ip, disks):
3567
  """Disconnects the network on a list of drbd devices.
3568

3569
  """
3570
  bdevs = _FindDisks(nodes_ip, disks)
3571

    
3572
  # disconnect disks
3573
  for rd in bdevs:
3574
    try:
3575
      rd.DisconnectNet()
3576
    except errors.BlockDeviceError, err:
3577
      _Fail("Can't change network configuration to standalone mode: %s",
3578
            err, exc=True)
3579

    
3580

    
3581
def DrbdAttachNet(nodes_ip, disks, instance_name, multimaster):
3582
  """Attaches the network on a list of drbd devices.
3583

3584
  """
3585
  bdevs = _FindDisks(nodes_ip, disks)
3586

    
3587
  if multimaster:
3588
    for idx, rd in enumerate(bdevs):
3589
      try:
3590
        _SymlinkBlockDev(instance_name, rd.dev_path, idx)
3591
      except EnvironmentError, err:
3592
        _Fail("Can't create symlink: %s", err)
3593
  # reconnect disks, switch to new master configuration and if
3594
  # needed primary mode
3595
  for rd in bdevs:
3596
    try:
3597
      rd.AttachNet(multimaster)
3598
    except errors.BlockDeviceError, err:
3599
      _Fail("Can't change network configuration: %s", err)
3600

    
3601
  # wait until the disks are connected; we need to retry the re-attach
3602
  # if the device becomes standalone, as this might happen if the one
3603
  # node disconnects and reconnects in a different mode before the
3604
  # other node reconnects; in this case, one or both of the nodes will
3605
  # decide it has wrong configuration and switch to standalone
3606

    
3607
  def _Attach():
3608
    all_connected = True
3609

    
3610
    for rd in bdevs:
3611
      stats = rd.GetProcStatus()
3612

    
3613
      all_connected = (all_connected and
3614
                       (stats.is_connected or stats.is_in_resync))
3615

    
3616
      if stats.is_standalone:
3617
        # peer had different config info and this node became
3618
        # standalone, even though this should not happen with the
3619
        # new staged way of changing disk configs
3620
        try:
3621
          rd.AttachNet(multimaster)
3622
        except errors.BlockDeviceError, err:
3623
          _Fail("Can't change network configuration: %s", err)
3624

    
3625
    if not all_connected:
3626
      raise utils.RetryAgain()
3627

    
3628
  try:
3629
    # Start with a delay of 100 miliseconds and go up to 5 seconds
3630
    utils.Retry(_Attach, (0.1, 1.5, 5.0), 2 * 60)
3631
  except utils.RetryTimeout:
3632
    _Fail("Timeout in disk reconnecting")
3633

    
3634
  if multimaster:
3635
    # change to primary mode
3636
    for rd in bdevs:
3637
      try:
3638
        rd.Open()
3639
      except errors.BlockDeviceError, err:
3640
        _Fail("Can't change to primary mode: %s", err)
3641

    
3642

    
3643
def DrbdWaitSync(nodes_ip, disks):
3644
  """Wait until DRBDs have synchronized.
3645

3646
  """
3647
  def _helper(rd):
3648
    stats = rd.GetProcStatus()
3649
    if not (stats.is_connected or stats.is_in_resync):
3650
      raise utils.RetryAgain()
3651
    return stats
3652

    
3653
  bdevs = _FindDisks(nodes_ip, disks)
3654

    
3655
  min_resync = 100
3656
  alldone = True
3657
  for rd in bdevs:
3658
    try:
3659
      # poll each second for 15 seconds
3660
      stats = utils.Retry(_helper, 1, 15, args=[rd])
3661
    except utils.RetryTimeout:
3662
      stats = rd.GetProcStatus()
3663
      # last check
3664
      if not (stats.is_connected or stats.is_in_resync):
3665
        _Fail("DRBD device %s is not in sync: stats=%s", rd, stats)
3666
    alldone = alldone and (not stats.is_in_resync)
3667
    if stats.sync_percent is not None:
3668
      min_resync = min(min_resync, stats.sync_percent)
3669

    
3670
  return (alldone, min_resync)
3671

    
3672

    
3673
def GetDrbdUsermodeHelper():
3674
  """Returns DRBD usermode helper currently configured.
3675

3676
  """
3677
  try:
3678
    return bdev.BaseDRBD.GetUsermodeHelper()
3679
  except errors.BlockDeviceError, err:
3680
    _Fail(str(err))
3681

    
3682

    
3683
def PowercycleNode(hypervisor_type):
3684
  """Hard-powercycle the node.
3685

3686
  Because we need to return first, and schedule the powercycle in the
3687
  background, we won't be able to report failures nicely.
3688

3689
  """
3690
  hyper = hypervisor.GetHypervisor(hypervisor_type)
3691
  try:
3692
    pid = os.fork()
3693
  except OSError:
3694
    # if we can't fork, we'll pretend that we're in the child process
3695
    pid = 0
3696
  if pid > 0:
3697
    return "Reboot scheduled in 5 seconds"
3698
  # ensure the child is running on ram
3699
  try:
3700
    utils.Mlockall()
3701
  except Exception: # pylint: disable=W0703
3702
    pass
3703
  time.sleep(5)
3704
  hyper.PowercycleNode()
3705

    
3706

    
3707
def _VerifyRestrictedCmdName(cmd):
3708
  """Verifies a restricted command name.
3709

3710
  @type cmd: string
3711
  @param cmd: Command name
3712
  @rtype: tuple; (boolean, string or None)
3713
  @return: The tuple's first element is the status; if C{False}, the second
3714
    element is an error message string, otherwise it's C{None}
3715

3716
  """
3717
  if not cmd.strip():
3718
    return (False, "Missing command name")
3719

    
3720
  if os.path.basename(cmd) != cmd:
3721
    return (False, "Invalid command name")
3722

    
3723
  if not constants.EXT_PLUGIN_MASK.match(cmd):
3724
    return (False, "Command name contains forbidden characters")
3725

    
3726
  return (True, None)
3727

    
3728

    
3729
def _CommonRestrictedCmdCheck(path, owner):
3730
  """Common checks for restricted command file system directories and files.
3731

3732
  @type path: string
3733
  @param path: Path to check
3734
  @param owner: C{None} or tuple containing UID and GID
3735
  @rtype: tuple; (boolean, string or C{os.stat} result)
3736
  @return: The tuple's first element is the status; if C{False}, the second
3737
    element is an error message string, otherwise it's the result of C{os.stat}
3738

3739
  """
3740
  if owner is None:
3741
    # Default to root as owner
3742
    owner = (0, 0)
3743

    
3744
  try:
3745
    st = os.stat(path)
3746
  except EnvironmentError, err:
3747
    return (False, "Can't stat(2) '%s': %s" % (path, err))
3748

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

    
3752
  if (st.st_uid, st.st_gid) != owner:
3753
    (owner_uid, owner_gid) = owner
3754
    return (False, "'%s' is not owned by %s:%s" % (path, owner_uid, owner_gid))
3755

    
3756
  return (True, st)
3757

    
3758

    
3759
def _VerifyRestrictedCmdDirectory(path, _owner=None):
3760
  """Verifies restricted command directory.
3761

3762
  @type path: string
3763
  @param path: Path to check
3764
  @rtype: tuple; (boolean, string or None)
3765
  @return: The tuple's first element is the status; if C{False}, the second
3766
    element is an error message string, otherwise it's C{None}
3767

3768
  """
3769
  (status, value) = _CommonRestrictedCmdCheck(path, _owner)
3770

    
3771
  if not status:
3772
    return (False, value)
3773

    
3774
  if not stat.S_ISDIR(value.st_mode):
3775
    return (False, "Path '%s' is not a directory" % path)
3776

    
3777
  return (True, None)
3778

    
3779

    
3780
def _VerifyRestrictedCmd(path, cmd, _owner=None):
3781
  """Verifies a whole restricted command and returns its executable filename.
3782

3783
  @type path: string
3784
  @param path: Directory containing restricted commands
3785
  @type cmd: string
3786
  @param cmd: Command name
3787
  @rtype: tuple; (boolean, string)
3788
  @return: The tuple's first element is the status; if C{False}, the second
3789
    element is an error message string, otherwise the second element is the
3790
    absolute path to the executable
3791

3792
  """
3793
  executable = utils.PathJoin(path, cmd)
3794

    
3795
  (status, msg) = _CommonRestrictedCmdCheck(executable, _owner)
3796

    
3797
  if not status:
3798
    return (False, msg)
3799

    
3800
  if not utils.IsExecutable(executable):
3801
    return (False, "access(2) thinks '%s' can't be executed" % executable)
3802

    
3803
  return (True, executable)
3804

    
3805

    
3806
def _PrepareRestrictedCmd(path, cmd,
3807
                          _verify_dir=_VerifyRestrictedCmdDirectory,
3808
                          _verify_name=_VerifyRestrictedCmdName,
3809
                          _verify_cmd=_VerifyRestrictedCmd):
3810
  """Performs a number of tests on a restricted command.
3811

3812
  @type path: string
3813
  @param path: Directory containing restricted commands
3814
  @type cmd: string
3815
  @param cmd: Command name
3816
  @return: Same as L{_VerifyRestrictedCmd}
3817

3818
  """
3819
  # Verify the directory first
3820
  (status, msg) = _verify_dir(path)
3821
  if status:
3822
    # Check command if everything was alright
3823
    (status, msg) = _verify_name(cmd)
3824

    
3825
  if not status:
3826
    return (False, msg)
3827

    
3828
  # Check actual executable
3829
  return _verify_cmd(path, cmd)
3830

    
3831

    
3832
def RunRestrictedCmd(cmd,
3833
                     _lock_timeout=_RCMD_LOCK_TIMEOUT,
3834
                     _lock_file=pathutils.RESTRICTED_COMMANDS_LOCK_FILE,
3835
                     _path=pathutils.RESTRICTED_COMMANDS_DIR,
3836
                     _sleep_fn=time.sleep,
3837
                     _prepare_fn=_PrepareRestrictedCmd,
3838
                     _runcmd_fn=utils.RunCmd,
3839
                     _enabled=constants.ENABLE_RESTRICTED_COMMANDS):
3840
  """Executes a restricted command after performing strict tests.
3841

3842
  @type cmd: string
3843
  @param cmd: Command name
3844
  @rtype: string
3845
  @return: Command output
3846
  @raise RPCFail: In case of an error
3847

3848
  """
3849
  logging.info("Preparing to run restricted command '%s'", cmd)
3850

    
3851
  if not _enabled:
3852
    _Fail("Restricted commands disabled at configure time")
3853

    
3854
  lock = None
3855
  try:
3856
    cmdresult = None
3857
    try:
3858
      lock = utils.FileLock.Open(_lock_file)
3859
      lock.Exclusive(blocking=True, timeout=_lock_timeout)
3860

    
3861
      (status, value) = _prepare_fn(_path, cmd)
3862

    
3863
      if status:
3864
        cmdresult = _runcmd_fn([value], env={}, reset_env=True,
3865
                               postfork_fn=lambda _: lock.Unlock())
3866
      else:
3867
        logging.error(value)
3868
    except Exception: # pylint: disable=W0703
3869
      # Keep original error in log
3870
      logging.exception("Caught exception")
3871

    
3872
    if cmdresult is None:
3873
      logging.info("Sleeping for %0.1f seconds before returning",
3874
                   _RCMD_INVALID_DELAY)
3875
      _sleep_fn(_RCMD_INVALID_DELAY)
3876

    
3877
      # Do not include original error message in returned error
3878
      _Fail("Executing command '%s' failed" % cmd)
3879
    elif cmdresult.failed or cmdresult.fail_reason:
3880
      _Fail("Restricted command '%s' failed: %s; output: %s",
3881
            cmd, cmdresult.fail_reason, cmdresult.output)
3882
    else:
3883
      return cmdresult.output
3884
  finally:
3885
    if lock is not None:
3886
      # Release lock at last
3887
      lock.Close()
3888
      lock = None
3889

    
3890

    
3891
def SetWatcherPause(until, _filename=pathutils.WATCHER_PAUSEFILE):
3892
  """Creates or removes the watcher pause file.
3893

3894
  @type until: None or number
3895
  @param until: Unix timestamp saying until when the watcher shouldn't run
3896

3897
  """
3898
  if until is None:
3899
    logging.info("Received request to no longer pause watcher")
3900
    utils.RemoveFile(_filename)
3901
  else:
3902
    logging.info("Received request to pause watcher until %s", until)
3903

    
3904
    if not ht.TNumber(until):
3905
      _Fail("Duration must be numeric")
3906

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

    
3909

    
3910
class HooksRunner(object):
3911
  """Hook runner.
3912

3913
  This class is instantiated on the node side (ganeti-noded) and not
3914
  on the master side.
3915

3916
  """
3917
  def __init__(self, hooks_base_dir=None):
3918
    """Constructor for hooks runner.
3919

3920
    @type hooks_base_dir: str or None
3921
    @param hooks_base_dir: if not None, this overrides the
3922
        L{pathutils.HOOKS_BASE_DIR} (useful for unittests)
3923

3924
    """
3925
    if hooks_base_dir is None:
3926
      hooks_base_dir = pathutils.HOOKS_BASE_DIR
3927
    # yeah, _BASE_DIR is not valid for attributes, we use it like a
3928
    # constant
3929
    self._BASE_DIR = hooks_base_dir # pylint: disable=C0103
3930

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

3934
    """
3935
    assert len(node_list) == 1
3936
    node = node_list[0]
3937
    _, myself = ssconf.GetMasterAndMyself()
3938
    assert node == myself
3939

    
3940
    results = self.RunHooks(hpath, phase, env)
3941

    
3942
    # Return values in the form expected by HooksMaster
3943
    return {node: (None, False, results)}
3944

    
3945
  def RunHooks(self, hpath, phase, env):
3946
    """Run the scripts in the hooks directory.
3947

3948
    @type hpath: str
3949
    @param hpath: the path to the hooks directory which
3950
        holds the scripts
3951
    @type phase: str
3952
    @param phase: either L{constants.HOOKS_PHASE_PRE} or
3953
        L{constants.HOOKS_PHASE_POST}
3954
    @type env: dict
3955
    @param env: dictionary with the environment for the hook
3956
    @rtype: list
3957
    @return: list of 3-element tuples:
3958
      - script path
3959
      - script result, either L{constants.HKR_SUCCESS} or
3960
        L{constants.HKR_FAIL}
3961
      - output of the script
3962

3963
    @raise errors.ProgrammerError: for invalid input
3964
        parameters
3965

3966
    """
3967
    if phase == constants.HOOKS_PHASE_PRE:
3968
      suffix = "pre"
3969
    elif phase == constants.HOOKS_PHASE_POST:
3970
      suffix = "post"
3971
    else:
3972
      _Fail("Unknown hooks phase '%s'", phase)
3973

    
3974
    subdir = "%s-%s.d" % (hpath, suffix)
3975
    dir_name = utils.PathJoin(self._BASE_DIR, subdir)
3976

    
3977
    results = []
3978

    
3979
    if not os.path.isdir(dir_name):
3980
      # for non-existing/non-dirs, we simply exit instead of logging a
3981
      # warning at every operation
3982
      return results
3983

    
3984
    runparts_results = utils.RunParts(dir_name, env=env, reset_env=True)
3985

    
3986
    for (relname, relstatus, runresult) in runparts_results:
3987
      if relstatus == constants.RUNPARTS_SKIP:
3988
        rrval = constants.HKR_SKIP
3989
        output = ""
3990
      elif relstatus == constants.RUNPARTS_ERR:
3991
        rrval = constants.HKR_FAIL
3992
        output = "Hook script execution error: %s" % runresult
3993
      elif relstatus == constants.RUNPARTS_RUN:
3994
        if runresult.failed:
3995
          rrval = constants.HKR_FAIL
3996
        else:
3997
          rrval = constants.HKR_SUCCESS
3998
        output = utils.SafeEncode(runresult.output.strip())
3999
      results.append(("%s/%s" % (subdir, relname), rrval, output))
4000

    
4001
    return results
4002

    
4003

    
4004
class IAllocatorRunner(object):
4005
  """IAllocator runner.
4006

4007
  This class is instantiated on the node side (ganeti-noded) and not on
4008
  the master side.
4009

4010
  """
4011
  @staticmethod
4012
  def Run(name, idata):
4013
    """Run an iallocator script.
4014

4015
    @type name: str
4016
    @param name: the iallocator script name
4017
    @type idata: str
4018
    @param idata: the allocator input data
4019

4020
    @rtype: tuple
4021
    @return: two element tuple of:
4022
       - status
4023
       - either error message or stdout of allocator (for success)
4024

4025
    """
4026
    alloc_script = utils.FindFile(name, constants.IALLOCATOR_SEARCH_PATH,
4027
                                  os.path.isfile)
4028
    if alloc_script is None:
4029
      _Fail("iallocator module '%s' not found in the search path", name)
4030

    
4031
    fd, fin_name = tempfile.mkstemp(prefix="ganeti-iallocator.")
4032
    try:
4033
      os.write(fd, idata)
4034
      os.close(fd)
4035
      result = utils.RunCmd([alloc_script, fin_name])
4036
      if result.failed:
4037
        _Fail("iallocator module '%s' failed: %s, output '%s'",
4038
              name, result.fail_reason, result.output)
4039
    finally:
4040
      os.unlink(fin_name)
4041

    
4042
    return result.stdout
4043

    
4044

    
4045
class DevCacheManager(object):
4046
  """Simple class for managing a cache of block device information.
4047

4048
  """
4049
  _DEV_PREFIX = "/dev/"
4050
  _ROOT_DIR = pathutils.BDEV_CACHE_DIR
4051

    
4052
  @classmethod
4053
  def _ConvertPath(cls, dev_path):
4054
    """Converts a /dev/name path to the cache file name.
4055

4056
    This replaces slashes with underscores and strips the /dev
4057
    prefix. It then returns the full path to the cache file.
4058

4059
    @type dev_path: str
4060
    @param dev_path: the C{/dev/} path name
4061
    @rtype: str
4062
    @return: the converted path name
4063

4064
    """
4065
    if dev_path.startswith(cls._DEV_PREFIX):
4066
      dev_path = dev_path[len(cls._DEV_PREFIX):]
4067
    dev_path = dev_path.replace("/", "_")
4068
    fpath = utils.PathJoin(cls._ROOT_DIR, "bdev_%s" % dev_path)
4069
    return fpath
4070

    
4071
  @classmethod
4072
  def UpdateCache(cls, dev_path, owner, on_primary, iv_name):
4073
    """Updates the cache information for a given device.
4074

4075
    @type dev_path: str
4076
    @param dev_path: the pathname of the device
4077
    @type owner: str
4078
    @param owner: the owner (instance name) of the device
4079
    @type on_primary: bool
4080
    @param on_primary: whether this is the primary
4081
        node nor not
4082
    @type iv_name: str
4083
    @param iv_name: the instance-visible name of the
4084
        device, as in objects.Disk.iv_name
4085

4086
    @rtype: None
4087

4088
    """
4089
    if dev_path is None:
4090
      logging.error("DevCacheManager.UpdateCache got a None dev_path")
4091
      return
4092
    fpath = cls._ConvertPath(dev_path)
4093
    if on_primary:
4094
      state = "primary"
4095
    else:
4096
      state = "secondary"
4097
    if iv_name is None:
4098
      iv_name = "not_visible"
4099
    fdata = "%s %s %s\n" % (str(owner), state, iv_name)
4100
    try:
4101
      utils.WriteFile(fpath, data=fdata)
4102
    except EnvironmentError, err:
4103
      logging.exception("Can't update bdev cache for %s: %s", dev_path, err)
4104

    
4105
  @classmethod
4106
  def RemoveCache(cls, dev_path):
4107
    """Remove data for a dev_path.
4108

4109
    This is just a wrapper over L{utils.io.RemoveFile} with a converted
4110
    path name and logging.
4111

4112
    @type dev_path: str
4113
    @param dev_path: the pathname of the device
4114

4115
    @rtype: None
4116

4117
    """
4118
    if dev_path is None:
4119
      logging.error("DevCacheManager.RemoveCache got a