Statistics
| Branch: | Tag: | Revision:

root / lib / backend.py @ ce9283c1

History | View | Annotate | Download (124.8 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.block 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):
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
  @rtype: None
1368

1369
  """
1370
  hv_name = instance.hypervisor
1371
  hyper = hypervisor.GetHypervisor(hv_name)
1372
  iname = instance.name
1373

    
1374
  if instance.name not in hyper.ListInstances():
1375
    logging.info("Instance %s not running, doing nothing", iname)
1376
    return
1377

    
1378
  class _TryShutdown:
1379
    def __init__(self):
1380
      self.tried_once = False
1381

    
1382
    def __call__(self):
1383
      if iname not in hyper.ListInstances():
1384
        return
1385

    
1386
      try:
1387
        hyper.StopInstance(instance, retry=self.tried_once)
1388
      except errors.HypervisorError, err:
1389
        if iname not in hyper.ListInstances():
1390
          # if the instance is no longer existing, consider this a
1391
          # success and go to cleanup
1392
          return
1393

    
1394
        _Fail("Failed to stop instance %s: %s", iname, err)
1395

    
1396
      self.tried_once = True
1397

    
1398
      raise utils.RetryAgain()
1399

    
1400
  try:
1401
    utils.Retry(_TryShutdown(), 5, timeout)
1402
  except utils.RetryTimeout:
1403
    # the shutdown did not succeed
1404
    logging.error("Shutdown of '%s' unsuccessful, forcing", iname)
1405

    
1406
    try:
1407
      hyper.StopInstance(instance, force=True)
1408
    except errors.HypervisorError, err:
1409
      if iname in hyper.ListInstances():
1410
        # only raise an error if the instance still exists, otherwise
1411
        # the error could simply be "instance ... unknown"!
1412
        _Fail("Failed to force stop instance %s: %s", iname, err)
1413

    
1414
    time.sleep(1)
1415

    
1416
    if iname in hyper.ListInstances():
1417
      _Fail("Could not shutdown instance %s even by destroy", iname)
1418

    
1419
  try:
1420
    hyper.CleanupInstance(instance.name)
1421
  except errors.HypervisorError, err:
1422
    logging.warning("Failed to execute post-shutdown cleanup step: %s", err)
1423

    
1424
  _RemoveBlockDevLinks(iname, instance.disks)
1425

    
1426

    
1427
def InstanceReboot(instance, reboot_type, shutdown_timeout, reason):
1428
  """Reboot an instance.
1429

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

1449
  """
1450
  running_instances = GetInstanceList([instance.hypervisor])
1451

    
1452
  if instance.name not in running_instances:
1453
    _Fail("Cannot reboot instance %s that is not running", instance.name)
1454

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

    
1472

    
1473
def InstanceBalloonMemory(instance, memory):
1474
  """Resize an instance's memory.
1475

1476
  @type instance: L{objects.Instance}
1477
  @param instance: the instance object
1478
  @type memory: int
1479
  @param memory: new memory amount in MB
1480
  @rtype: None
1481

1482
  """
1483
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1484
  running = hyper.ListInstances()
1485
  if instance.name not in running:
1486
    logging.info("Instance %s is not running, cannot balloon", instance.name)
1487
    return
1488
  try:
1489
    hyper.BalloonInstanceMemory(instance, memory)
1490
  except errors.HypervisorError, err:
1491
    _Fail("Failed to balloon instance memory: %s", err, exc=True)
1492

    
1493

    
1494
def MigrationInfo(instance):
1495
  """Gather information about an instance to be migrated.
1496

1497
  @type instance: L{objects.Instance}
1498
  @param instance: the instance definition
1499

1500
  """
1501
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1502
  try:
1503
    info = hyper.MigrationInfo(instance)
1504
  except errors.HypervisorError, err:
1505
    _Fail("Failed to fetch migration information: %s", err, exc=True)
1506
  return info
1507

    
1508

    
1509
def AcceptInstance(instance, info, target):
1510
  """Prepare the node to accept an instance.
1511

1512
  @type instance: L{objects.Instance}
1513
  @param instance: the instance definition
1514
  @type info: string/data (opaque)
1515
  @param info: migration information, from the source node
1516
  @type target: string
1517
  @param target: target host (usually ip), on this node
1518

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

    
1529
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1530
  try:
1531
    hyper.AcceptInstance(instance, info, target)
1532
  except errors.HypervisorError, err:
1533
    if instance.disk_template in constants.DTS_EXT_MIRROR:
1534
      _RemoveBlockDevLinks(instance.name, instance.disks)
1535
    _Fail("Failed to accept instance: %s", err, exc=True)
1536

    
1537

    
1538
def FinalizeMigrationDst(instance, info, success):
1539
  """Finalize any preparation to accept an instance.
1540

1541
  @type instance: L{objects.Instance}
1542
  @param instance: the instance definition
1543
  @type info: string/data (opaque)
1544
  @param info: migration information, from the source node
1545
  @type success: boolean
1546
  @param success: whether the migration was a success or a failure
1547

1548
  """
1549
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1550
  try:
1551
    hyper.FinalizeMigrationDst(instance, info, success)
1552
  except errors.HypervisorError, err:
1553
    _Fail("Failed to finalize migration on the target node: %s", err, exc=True)
1554

    
1555

    
1556
def MigrateInstance(instance, target, live):
1557
  """Migrates an instance to another node.
1558

1559
  @type instance: L{objects.Instance}
1560
  @param instance: the instance definition
1561
  @type target: string
1562
  @param target: the target node name
1563
  @type live: boolean
1564
  @param live: whether the migration should be done live or not (the
1565
      interpretation of this parameter is left to the hypervisor)
1566
  @raise RPCFail: if migration fails for some reason
1567

1568
  """
1569
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1570

    
1571
  try:
1572
    hyper.MigrateInstance(instance, target, live)
1573
  except errors.HypervisorError, err:
1574
    _Fail("Failed to migrate instance: %s", err, exc=True)
1575

    
1576

    
1577
def FinalizeMigrationSource(instance, success, live):
1578
  """Finalize the instance migration on the source node.
1579

1580
  @type instance: L{objects.Instance}
1581
  @param instance: the instance definition of the migrated instance
1582
  @type success: bool
1583
  @param success: whether the migration succeeded or not
1584
  @type live: bool
1585
  @param live: whether the user requested a live migration or not
1586
  @raise RPCFail: If the execution fails for some reason
1587

1588
  """
1589
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1590

    
1591
  try:
1592
    hyper.FinalizeMigrationSource(instance, success, live)
1593
  except Exception, err:  # pylint: disable=W0703
1594
    _Fail("Failed to finalize the migration on the source node: %s", err,
1595
          exc=True)
1596

    
1597

    
1598
def GetMigrationStatus(instance):
1599
  """Get the migration status
1600

1601
  @type instance: L{objects.Instance}
1602
  @param instance: the instance that is being migrated
1603
  @rtype: L{objects.MigrationStatus}
1604
  @return: the status of the current migration (one of
1605
           L{constants.HV_MIGRATION_VALID_STATUSES}), plus any additional
1606
           progress info that can be retrieved from the hypervisor
1607
  @raise RPCFail: If the migration status cannot be retrieved
1608

1609
  """
1610
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1611
  try:
1612
    return hyper.GetMigrationStatus(instance)
1613
  except Exception, err:  # pylint: disable=W0703
1614
    _Fail("Failed to get migration status: %s", err, exc=True)
1615

    
1616

    
1617
def BlockdevCreate(disk, size, owner, on_primary, info, excl_stor):
1618
  """Creates a block device for an instance.
1619

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

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

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

    
1659
  try:
1660
    device = bdev.Create(disk, clist, excl_stor)
1661
  except errors.BlockDeviceError, err:
1662
    _Fail("Can't create block device: %s", err)
1663

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

    
1677
  device.SetInfo(info)
1678

    
1679
  return device.unique_id
1680

    
1681

    
1682
def _WipeDevice(path, offset, size):
1683
  """This function actually wipes the device.
1684

1685
  @param path: The path to the device to wipe
1686
  @param offset: The offset in MiB in the file
1687
  @param size: The size in MiB to write
1688

1689
  """
1690
  # Internal sizes are always in Mebibytes; if the following "dd" command
1691
  # should use a different block size the offset and size given to this
1692
  # function must be adjusted accordingly before being passed to "dd".
1693
  block_size = 1024 * 1024
1694

    
1695
  cmd = [constants.DD_CMD, "if=/dev/zero", "seek=%d" % offset,
1696
         "bs=%s" % block_size, "oflag=direct", "of=%s" % path,
1697
         "count=%d" % size]
1698
  result = utils.RunCmd(cmd)
1699

    
1700
  if result.failed:
1701
    _Fail("Wipe command '%s' exited with error: %s; output: %s", result.cmd,
1702
          result.fail_reason, result.output)
1703

    
1704

    
1705
def BlockdevWipe(disk, offset, size):
1706
  """Wipes a block device.
1707

1708
  @type disk: L{objects.Disk}
1709
  @param disk: the disk object we want to wipe
1710
  @type offset: int
1711
  @param offset: The offset in MiB in the file
1712
  @type size: int
1713
  @param size: The size in MiB to write
1714

1715
  """
1716
  try:
1717
    rdev = _RecursiveFindBD(disk)
1718
  except errors.BlockDeviceError:
1719
    rdev = None
1720

    
1721
  if not rdev:
1722
    _Fail("Cannot execute wipe for device %s: device not found", disk.iv_name)
1723

    
1724
  # Do cross verify some of the parameters
1725
  if offset < 0:
1726
    _Fail("Negative offset")
1727
  if size < 0:
1728
    _Fail("Negative size")
1729
  if offset > rdev.size:
1730
    _Fail("Offset is bigger than device size")
1731
  if (offset + size) > rdev.size:
1732
    _Fail("The provided offset and size to wipe is bigger than device size")
1733

    
1734
  _WipeDevice(rdev.dev_path, offset, size)
1735

    
1736

    
1737
def BlockdevPauseResumeSync(disks, pause):
1738
  """Pause or resume the sync of the block device.
1739

1740
  @type disks: list of L{objects.Disk}
1741
  @param disks: the disks object we want to pause/resume
1742
  @type pause: bool
1743
  @param pause: Wheater to pause or resume
1744

1745
  """
1746
  success = []
1747
  for disk in disks:
1748
    try:
1749
      rdev = _RecursiveFindBD(disk)
1750
    except errors.BlockDeviceError:
1751
      rdev = None
1752

    
1753
    if not rdev:
1754
      success.append((False, ("Cannot change sync for device %s:"
1755
                              " device not found" % disk.iv_name)))
1756
      continue
1757

    
1758
    result = rdev.PauseResumeSync(pause)
1759

    
1760
    if result:
1761
      success.append((result, None))
1762
    else:
1763
      if pause:
1764
        msg = "Pause"
1765
      else:
1766
        msg = "Resume"
1767
      success.append((result, "%s for device %s failed" % (msg, disk.iv_name)))
1768

    
1769
  return success
1770

    
1771

    
1772
def BlockdevRemove(disk):
1773
  """Remove a block device.
1774

1775
  @note: This is intended to be called recursively.
1776

1777
  @type disk: L{objects.Disk}
1778
  @param disk: the disk object we should remove
1779
  @rtype: boolean
1780
  @return: the success of the operation
1781

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

    
1799
  if disk.children:
1800
    for child in disk.children:
1801
      try:
1802
        BlockdevRemove(child)
1803
      except RPCFail, err:
1804
        msgs.append(str(err))
1805

    
1806
  if msgs:
1807
    _Fail("; ".join(msgs))
1808

    
1809

    
1810
def _RecursiveAssembleBD(disk, owner, as_primary):
1811
  """Activate a block device for an instance.
1812

1813
  This is run on the primary and secondary nodes for an instance.
1814

1815
  @note: this function is called recursively.
1816

1817
  @type disk: L{objects.Disk}
1818
  @param disk: the disk we try to assemble
1819
  @type owner: str
1820
  @param owner: the name of the instance which owns the disk
1821
  @type as_primary: boolean
1822
  @param as_primary: if we should make the block device
1823
      read/write
1824

1825
  @return: the assembled device or None (in case no device
1826
      was assembled)
1827
  @raise errors.BlockDeviceError: in case there is an error
1828
      during the activation of the children or the device
1829
      itself
1830

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

    
1850
  if as_primary or disk.AssembleOnSecondary():
1851
    r_dev = bdev.Assemble(disk, children)
1852
    result = r_dev
1853
    if as_primary or disk.OpenOnSecondary():
1854
      r_dev.Open()
1855
    DevCacheManager.UpdateCache(r_dev.dev_path, owner,
1856
                                as_primary, disk.iv_name)
1857

    
1858
  else:
1859
    result = True
1860
  return result
1861

    
1862

    
1863
def BlockdevAssemble(disk, owner, as_primary, idx):
1864
  """Activate a block device for an instance.
1865

1866
  This is a wrapper over _RecursiveAssembleBD.
1867

1868
  @rtype: str or boolean
1869
  @return: a C{/dev/...} path for primary nodes, and
1870
      C{True} for secondary nodes
1871

1872
  """
1873
  try:
1874
    result = _RecursiveAssembleBD(disk, owner, as_primary)
1875
    if isinstance(result, bdev.BlockDev):
1876
      # pylint: disable=E1103
1877
      result = result.dev_path
1878
      if as_primary:
1879
        _SymlinkBlockDev(owner, result, idx)
1880
  except errors.BlockDeviceError, err:
1881
    _Fail("Error while assembling disk: %s", err, exc=True)
1882
  except OSError, err:
1883
    _Fail("Error while symlinking disk: %s", err, exc=True)
1884

    
1885
  return result
1886

    
1887

    
1888
def BlockdevShutdown(disk):
1889
  """Shut down a block device.
1890

1891
  First, if the device is assembled (Attach() is successful), then
1892
  the device is shutdown. Then the children of the device are
1893
  shutdown.
1894

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

1899
  @type disk: L{objects.Disk}
1900
  @param disk: the description of the disk we should
1901
      shutdown
1902
  @rtype: None
1903

1904
  """
1905
  msgs = []
1906
  r_dev = _RecursiveFindBD(disk)
1907
  if r_dev is not None:
1908
    r_path = r_dev.dev_path
1909
    try:
1910
      r_dev.Shutdown()
1911
      DevCacheManager.RemoveCache(r_path)
1912
    except errors.BlockDeviceError, err:
1913
      msgs.append(str(err))
1914

    
1915
  if disk.children:
1916
    for child in disk.children:
1917
      try:
1918
        BlockdevShutdown(child)
1919
      except RPCFail, err:
1920
        msgs.append(str(err))
1921

    
1922
  if msgs:
1923
    _Fail("; ".join(msgs))
1924

    
1925

    
1926
def BlockdevAddchildren(parent_cdev, new_cdevs):
1927
  """Extend a mirrored block device.
1928

1929
  @type parent_cdev: L{objects.Disk}
1930
  @param parent_cdev: the disk to which we should add children
1931
  @type new_cdevs: list of L{objects.Disk}
1932
  @param new_cdevs: the list of children which we should add
1933
  @rtype: None
1934

1935
  """
1936
  parent_bdev = _RecursiveFindBD(parent_cdev)
1937
  if parent_bdev is None:
1938
    _Fail("Can't find parent device '%s' in add children", parent_cdev)
1939
  new_bdevs = [_RecursiveFindBD(disk) for disk in new_cdevs]
1940
  if new_bdevs.count(None) > 0:
1941
    _Fail("Can't find new device(s) to add: %s:%s", new_bdevs, new_cdevs)
1942
  parent_bdev.AddChildren(new_bdevs)
1943

    
1944

    
1945
def BlockdevRemovechildren(parent_cdev, new_cdevs):
1946
  """Shrink a mirrored block device.
1947

1948
  @type parent_cdev: L{objects.Disk}
1949
  @param parent_cdev: the disk from which we should remove children
1950
  @type new_cdevs: list of L{objects.Disk}
1951
  @param new_cdevs: the list of children which we should remove
1952
  @rtype: None
1953

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

    
1973

    
1974
def BlockdevGetmirrorstatus(disks):
1975
  """Get the mirroring status of a list of devices.
1976

1977
  @type disks: list of L{objects.Disk}
1978
  @param disks: the list of disks which we should query
1979
  @rtype: disk
1980
  @return: List of L{objects.BlockDevStatus}, one for each disk
1981
  @raise errors.BlockDeviceError: if any of the disks cannot be
1982
      found
1983

1984
  """
1985
  stats = []
1986
  for dsk in disks:
1987
    rbd = _RecursiveFindBD(dsk)
1988
    if rbd is None:
1989
      _Fail("Can't find device %s", dsk)
1990

    
1991
    stats.append(rbd.CombinedSyncStatus())
1992

    
1993
  return stats
1994

    
1995

    
1996
def BlockdevGetmirrorstatusMulti(disks):
1997
  """Get the mirroring status of a list of devices.
1998

1999
  @type disks: list of L{objects.Disk}
2000
  @param disks: the list of disks which we should query
2001
  @rtype: disk
2002
  @return: List of tuples, (bool, status), one for each disk; bool denotes
2003
    success/failure, status is L{objects.BlockDevStatus} on success, string
2004
    otherwise
2005

2006
  """
2007
  result = []
2008
  for disk in disks:
2009
    try:
2010
      rbd = _RecursiveFindBD(disk)
2011
      if rbd is None:
2012
        result.append((False, "Can't find device %s" % disk))
2013
        continue
2014

    
2015
      status = rbd.CombinedSyncStatus()
2016
    except errors.BlockDeviceError, err:
2017
      logging.exception("Error while getting disk status")
2018
      result.append((False, str(err)))
2019
    else:
2020
      result.append((True, status))
2021

    
2022
  assert len(disks) == len(result)
2023

    
2024
  return result
2025

    
2026

    
2027
def _RecursiveFindBD(disk):
2028
  """Check if a device is activated.
2029

2030
  If so, return information about the real device.
2031

2032
  @type disk: L{objects.Disk}
2033
  @param disk: the disk object we need to find
2034

2035
  @return: None if the device can't be found,
2036
      otherwise the device instance
2037

2038
  """
2039
  children = []
2040
  if disk.children:
2041
    for chdisk in disk.children:
2042
      children.append(_RecursiveFindBD(chdisk))
2043

    
2044
  return bdev.FindDevice(disk, children)
2045

    
2046

    
2047
def _OpenRealBD(disk):
2048
  """Opens the underlying block device of a disk.
2049

2050
  @type disk: L{objects.Disk}
2051
  @param disk: the disk object we want to open
2052

2053
  """
2054
  real_disk = _RecursiveFindBD(disk)
2055
  if real_disk is None:
2056
    _Fail("Block device '%s' is not set up", disk)
2057

    
2058
  real_disk.Open()
2059

    
2060
  return real_disk
2061

    
2062

    
2063
def BlockdevFind(disk):
2064
  """Check if a device is activated.
2065

2066
  If it is, return information about the real device.
2067

2068
  @type disk: L{objects.Disk}
2069
  @param disk: the disk to find
2070
  @rtype: None or objects.BlockDevStatus
2071
  @return: None if the disk cannot be found, otherwise a the current
2072
           information
2073

2074
  """
2075
  try:
2076
    rbd = _RecursiveFindBD(disk)
2077
  except errors.BlockDeviceError, err:
2078
    _Fail("Failed to find device: %s", err, exc=True)
2079

    
2080
  if rbd is None:
2081
    return None
2082

    
2083
  return rbd.GetSyncStatus()
2084

    
2085

    
2086
def BlockdevGetsize(disks):
2087
  """Computes the size of the given disks.
2088

2089
  If a disk is not found, returns None instead.
2090

2091
  @type disks: list of L{objects.Disk}
2092
  @param disks: the list of disk to compute the size for
2093
  @rtype: list
2094
  @return: list with elements None if the disk cannot be found,
2095
      otherwise the size
2096

2097
  """
2098
  result = []
2099
  for cf in disks:
2100
    try:
2101
      rbd = _RecursiveFindBD(cf)
2102
    except errors.BlockDeviceError:
2103
      result.append(None)
2104
      continue
2105
    if rbd is None:
2106
      result.append(None)
2107
    else:
2108
      result.append(rbd.GetActualSize())
2109
  return result
2110

    
2111

    
2112
def BlockdevExport(disk, dest_node, dest_path, cluster_name):
2113
  """Export a block device to a remote node.
2114

2115
  @type disk: L{objects.Disk}
2116
  @param disk: the description of the disk to export
2117
  @type dest_node: str
2118
  @param dest_node: the destination node to export to
2119
  @type dest_path: str
2120
  @param dest_path: the destination path on the target node
2121
  @type cluster_name: str
2122
  @param cluster_name: the cluster name, needed for SSH hostalias
2123
  @rtype: None
2124

2125
  """
2126
  real_disk = _OpenRealBD(disk)
2127

    
2128
  # the block size on the read dd is 1MiB to match our units
2129
  expcmd = utils.BuildShellCmd("set -e; set -o pipefail; "
2130
                               "dd if=%s bs=1048576 count=%s",
2131
                               real_disk.dev_path, str(disk.size))
2132

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

    
2142
  remotecmd = _GetSshRunner(cluster_name).BuildCmd(dest_node,
2143
                                                   constants.SSH_LOGIN_USER,
2144
                                                   destcmd)
2145

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

    
2149
  result = utils.RunCmd(["bash", "-c", command])
2150

    
2151
  if result.failed:
2152
    _Fail("Disk copy command '%s' returned error: %s"
2153
          " output: %s", command, result.fail_reason, result.output)
2154

    
2155

    
2156
def UploadFile(file_name, data, mode, uid, gid, atime, mtime):
2157
  """Write a file to the filesystem.
2158

2159
  This allows the master to overwrite(!) a file. It will only perform
2160
  the operation if the file belongs to a list of configuration files.
2161

2162
  @type file_name: str
2163
  @param file_name: the target file name
2164
  @type data: str
2165
  @param data: the new contents of the file
2166
  @type mode: int
2167
  @param mode: the mode to give the file (can be None)
2168
  @type uid: string
2169
  @param uid: the owner of the file
2170
  @type gid: string
2171
  @param gid: the group of the file
2172
  @type atime: float
2173
  @param atime: the atime to set on the file (can be None)
2174
  @type mtime: float
2175
  @param mtime: the mtime to set on the file (can be None)
2176
  @rtype: None
2177

2178
  """
2179
  file_name = vcluster.LocalizeVirtualPath(file_name)
2180

    
2181
  if not os.path.isabs(file_name):
2182
    _Fail("Filename passed to UploadFile is not absolute: '%s'", file_name)
2183

    
2184
  if file_name not in _ALLOWED_UPLOAD_FILES:
2185
    _Fail("Filename passed to UploadFile not in allowed upload targets: '%s'",
2186
          file_name)
2187

    
2188
  raw_data = _Decompress(data)
2189

    
2190
  if not (isinstance(uid, basestring) and isinstance(gid, basestring)):
2191
    _Fail("Invalid username/groupname type")
2192

    
2193
  getents = runtime.GetEnts()
2194
  uid = getents.LookupUser(uid)
2195
  gid = getents.LookupGroup(gid)
2196

    
2197
  utils.SafeWriteFile(file_name, None,
2198
                      data=raw_data, mode=mode, uid=uid, gid=gid,
2199
                      atime=atime, mtime=mtime)
2200

    
2201

    
2202
def RunOob(oob_program, command, node, timeout):
2203
  """Executes oob_program with given command on given node.
2204

2205
  @param oob_program: The path to the executable oob_program
2206
  @param command: The command to invoke on oob_program
2207
  @param node: The node given as an argument to the program
2208
  @param timeout: Timeout after which we kill the oob program
2209

2210
  @return: stdout
2211
  @raise RPCFail: If execution fails for some reason
2212

2213
  """
2214
  result = utils.RunCmd([oob_program, command, node], timeout=timeout)
2215

    
2216
  if result.failed:
2217
    _Fail("'%s' failed with reason '%s'; output: %s", result.cmd,
2218
          result.fail_reason, result.output)
2219

    
2220
  return result.stdout
2221

    
2222

    
2223
def _OSOndiskAPIVersion(os_dir):
2224
  """Compute and return the API version of a given OS.
2225

2226
  This function will try to read the API version of the OS residing in
2227
  the 'os_dir' directory.
2228

2229
  @type os_dir: str
2230
  @param os_dir: the directory in which we should look for the OS
2231
  @rtype: tuple
2232
  @return: tuple (status, data) with status denoting the validity and
2233
      data holding either the vaid versions or an error message
2234

2235
  """
2236
  api_file = utils.PathJoin(os_dir, constants.OS_API_FILE)
2237

    
2238
  try:
2239
    st = os.stat(api_file)
2240
  except EnvironmentError, err:
2241
    return False, ("Required file '%s' not found under path %s: %s" %
2242
                   (constants.OS_API_FILE, os_dir, utils.ErrnoOrStr(err)))
2243

    
2244
  if not stat.S_ISREG(stat.S_IFMT(st.st_mode)):
2245
    return False, ("File '%s' in %s is not a regular file" %
2246
                   (constants.OS_API_FILE, os_dir))
2247

    
2248
  try:
2249
    api_versions = utils.ReadFile(api_file).splitlines()
2250
  except EnvironmentError, err:
2251
    return False, ("Error while reading the API version file at %s: %s" %
2252
                   (api_file, utils.ErrnoOrStr(err)))
2253

    
2254
  try:
2255
    api_versions = [int(version.strip()) for version in api_versions]
2256
  except (TypeError, ValueError), err:
2257
    return False, ("API version(s) can't be converted to integer: %s" %
2258
                   str(err))
2259

    
2260
  return True, api_versions
2261

    
2262

    
2263
def DiagnoseOS(top_dirs=None):
2264
  """Compute the validity for all OSes.
2265

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

2282
  """
2283
  if top_dirs is None:
2284
    top_dirs = pathutils.OS_SEARCH_PATH
2285

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

    
2308
  return result
2309

    
2310

    
2311
def _TryOSFromDisk(name, base_dir=None):
2312
  """Create an OS instance from disk.
2313

2314
  This function will return an OS instance if the given name is a
2315
  valid OS name.
2316

2317
  @type base_dir: string
2318
  @keyword base_dir: Base directory containing OS installations.
2319
                     Defaults to a search in all the OS_SEARCH_PATH dirs.
2320
  @rtype: tuple
2321
  @return: success and either the OS instance if we find a valid one,
2322
      or error message
2323

2324
  """
2325
  if base_dir is None:
2326
    os_dir = utils.FindFile(name, pathutils.OS_SEARCH_PATH, os.path.isdir)
2327
  else:
2328
    os_dir = utils.FindFile(name, [base_dir], os.path.isdir)
2329

    
2330
  if os_dir is None:
2331
    return False, "Directory for OS %s not found in search path" % name
2332

    
2333
  status, api_versions = _OSOndiskAPIVersion(os_dir)
2334
  if not status:
2335
    # push the error up
2336
    return status, api_versions
2337

    
2338
  if not constants.OS_API_VERSIONS.intersection(api_versions):
2339
    return False, ("API version mismatch for path '%s': found %s, want %s." %
2340
                   (os_dir, api_versions, constants.OS_API_VERSIONS))
2341

    
2342
  # OS Files dictionary, we will populate it with the absolute path
2343
  # names; if the value is True, then it is a required file, otherwise
2344
  # an optional one
2345
  os_files = dict.fromkeys(constants.OS_SCRIPTS, True)
2346

    
2347
  if max(api_versions) >= constants.OS_API_V15:
2348
    os_files[constants.OS_VARIANTS_FILE] = False
2349

    
2350
  if max(api_versions) >= constants.OS_API_V20:
2351
    os_files[constants.OS_PARAMETERS_FILE] = True
2352
  else:
2353
    del os_files[constants.OS_SCRIPT_VERIFY]
2354

    
2355
  for (filename, required) in os_files.items():
2356
    os_files[filename] = utils.PathJoin(os_dir, filename)
2357

    
2358
    try:
2359
      st = os.stat(os_files[filename])
2360
    except EnvironmentError, err:
2361
      if err.errno == errno.ENOENT and not required:
2362
        del os_files[filename]
2363
        continue
2364
      return False, ("File '%s' under path '%s' is missing (%s)" %
2365
                     (filename, os_dir, utils.ErrnoOrStr(err)))
2366

    
2367
    if not stat.S_ISREG(stat.S_IFMT(st.st_mode)):
2368
      return False, ("File '%s' under path '%s' is not a regular file" %
2369
                     (filename, os_dir))
2370

    
2371
    if filename in constants.OS_SCRIPTS:
2372
      if stat.S_IMODE(st.st_mode) & stat.S_IXUSR != stat.S_IXUSR:
2373
        return False, ("File '%s' under path '%s' is not executable" %
2374
                       (filename, os_dir))
2375

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

    
2388
  parameters = []
2389
  if constants.OS_PARAMETERS_FILE in os_files:
2390
    parameters_file = os_files[constants.OS_PARAMETERS_FILE]
2391
    try:
2392
      parameters = utils.ReadFile(parameters_file).splitlines()
2393
    except EnvironmentError, err:
2394
      return False, ("Error while reading the OS parameters file at %s: %s" %
2395
                     (parameters_file, utils.ErrnoOrStr(err)))
2396
    parameters = [v.split(None, 1) for v in parameters]
2397

    
2398
  os_obj = objects.OS(name=name, path=os_dir,
2399
                      create_script=os_files[constants.OS_SCRIPT_CREATE],
2400
                      export_script=os_files[constants.OS_SCRIPT_EXPORT],
2401
                      import_script=os_files[constants.OS_SCRIPT_IMPORT],
2402
                      rename_script=os_files[constants.OS_SCRIPT_RENAME],
2403
                      verify_script=os_files.get(constants.OS_SCRIPT_VERIFY,
2404
                                                 None),
2405
                      supported_variants=variants,
2406
                      supported_parameters=parameters,
2407
                      api_versions=api_versions)
2408
  return True, os_obj
2409

    
2410

    
2411
def OSFromDisk(name, base_dir=None):
2412
  """Create an OS instance from disk.
2413

2414
  This function will return an OS instance if the given name is a
2415
  valid OS name. Otherwise, it will raise an appropriate
2416
  L{RPCFail} exception, detailing why this is not a valid OS.
2417

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

2421
  @type base_dir: string
2422
  @keyword base_dir: Base directory containing OS installations.
2423
                     Defaults to a search in all the OS_SEARCH_PATH dirs.
2424
  @rtype: L{objects.OS}
2425
  @return: the OS instance if we find a valid one
2426
  @raise RPCFail: if we don't find a valid OS
2427

2428
  """
2429
  name_only = objects.OS.GetName(name)
2430
  status, payload = _TryOSFromDisk(name_only, base_dir)
2431

    
2432
  if not status:
2433
    _Fail(payload)
2434

    
2435
  return payload
2436

    
2437

    
2438
def OSCoreEnv(os_name, inst_os, os_params, debug=0):
2439
  """Calculate the basic environment for an os script.
2440

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

2454
  """
2455
  result = {}
2456
  api_version = \
2457
    max(constants.OS_API_VERSIONS.intersection(inst_os.api_versions))
2458
  result["OS_API_VERSION"] = "%d" % api_version
2459
  result["OS_NAME"] = inst_os.name
2460
  result["DEBUG_LEVEL"] = "%d" % debug
2461

    
2462
  # OS variants
2463
  if api_version >= constants.OS_API_V15 and inst_os.supported_variants:
2464
    variant = objects.OS.GetVariant(os_name)
2465
    if not variant:
2466
      variant = inst_os.supported_variants[0]
2467
  else:
2468
    variant = ""
2469
  result["OS_VARIANT"] = variant
2470

    
2471
  # OS params
2472
  for pname, pvalue in os_params.items():
2473
    result["OSP_%s" % pname.upper()] = pvalue
2474

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

    
2480
  return result
2481

    
2482

    
2483
def OSEnvironment(instance, inst_os, debug=0):
2484
  """Calculate the environment for an os script.
2485

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

2497
  """
2498
  result = OSCoreEnv(instance.os, inst_os, instance.osparams, debug=debug)
2499

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

    
2503
  result["HYPERVISOR"] = instance.hypervisor
2504
  result["DISK_COUNT"] = "%d" % len(instance.disks)
2505
  result["NIC_COUNT"] = "%d" % len(instance.nics)
2506
  result["INSTANCE_SECONDARY_NODES"] = \
2507
      ("%s" % " ".join(instance.secondary_nodes))
2508

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

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

    
2540
  # HV/BE params
2541
  for source, kind in [(instance.beparams, "BE"), (instance.hvparams, "HV")]:
2542
    for key, value in source.items():
2543
      result["INSTANCE_%s_%s" % (kind, key)] = str(value)
2544

    
2545
  return result
2546

    
2547

    
2548
def DiagnoseExtStorage(top_dirs=None):
2549
  """Compute the validity for all ExtStorage Providers.
2550

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

2566
  """
2567
  if top_dirs is None:
2568
    top_dirs = pathutils.ES_SEARCH_PATH
2569

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

    
2590
  return result
2591

    
2592

    
2593
def BlockdevGrow(disk, amount, dryrun, backingstore):
2594
  """Grow a stack of block devices.
2595

2596
  This function is called recursively, with the childrens being the
2597
  first ones to resize.
2598

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

2613
  """
2614
  r_dev = _RecursiveFindBD(disk)
2615
  if r_dev is None:
2616
    _Fail("Cannot find block device %s", disk)
2617

    
2618
  try:
2619
    r_dev.Grow(amount, dryrun, backingstore)
2620
  except errors.BlockDeviceError, err:
2621
    _Fail("Failed to grow block device: %s", err, exc=True)
2622

    
2623

    
2624
def BlockdevSnapshot(disk):
2625
  """Create a snapshot copy of a block device.
2626

2627
  This function is called recursively, and the snapshot is actually created
2628
  just for the leaf lvm backend device.
2629

2630
  @type disk: L{objects.Disk}
2631
  @param disk: the disk to be snapshotted
2632
  @rtype: string
2633
  @return: snapshot disk ID as (vg, lv)
2634

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

    
2653

    
2654
def BlockdevSetInfo(disk, info):
2655
  """Sets 'metadata' information on block devices.
2656

2657
  This function sets 'info' metadata on block devices. Initial
2658
  information is set at device creation; this function should be used
2659
  for example after renames.
2660

2661
  @type disk: L{objects.Disk}
2662
  @param disk: the disk to be grown
2663
  @type info: string
2664
  @param info: new 'info' metadata
2665
  @rtype: (status, result)
2666
  @return: a tuple with the status of the operation (True/False), and
2667
      the errors message if status is False
2668

2669
  """
2670
  r_dev = _RecursiveFindBD(disk)
2671
  if r_dev is None:
2672
    _Fail("Cannot find block device %s", disk)
2673

    
2674
  try:
2675
    r_dev.SetInfo(info)
2676
  except errors.BlockDeviceError, err:
2677
    _Fail("Failed to set information on block device: %s", err, exc=True)
2678

    
2679

    
2680
def FinalizeExport(instance, snap_disks):
2681
  """Write out the export configuration information.
2682

2683
  @type instance: L{objects.Instance}
2684
  @param instance: the instance which we export, used for
2685
      saving configuration
2686
  @type snap_disks: list of L{objects.Disk}
2687
  @param snap_disks: list of snapshot block devices, which
2688
      will be used to get the actual name of the dump file
2689

2690
  @rtype: None
2691

2692
  """
2693
  destdir = utils.PathJoin(pathutils.EXPORT_DIR, instance.name + ".new")
2694
  finaldestdir = utils.PathJoin(pathutils.EXPORT_DIR, instance.name)
2695

    
2696
  config = objects.SerializableConfigParser()
2697

    
2698
  config.add_section(constants.INISECT_EXP)
2699
  config.set(constants.INISECT_EXP, "version", "0")
2700
  config.set(constants.INISECT_EXP, "timestamp", "%d" % int(time.time()))
2701
  config.set(constants.INISECT_EXP, "source", instance.primary_node)
2702
  config.set(constants.INISECT_EXP, "os", instance.os)
2703
  config.set(constants.INISECT_EXP, "compression", "none")
2704

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

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

    
2734
  disk_total = 0
2735
  for disk_count, disk in enumerate(snap_disks):
2736
    if disk:
2737
      disk_total += 1
2738
      config.set(constants.INISECT_INS, "disk%d_ivname" % disk_count,
2739
                 ("%s" % disk.iv_name))
2740
      config.set(constants.INISECT_INS, "disk%d_dump" % disk_count,
2741
                 ("%s" % disk.physical_id[1]))
2742
      config.set(constants.INISECT_INS, "disk%d_size" % disk_count,
2743
                 ("%d" % disk.size))
2744

    
2745
  config.set(constants.INISECT_INS, "disk_count", "%d" % disk_total)
2746

    
2747
  # New-style hypervisor/backend parameters
2748

    
2749
  config.add_section(constants.INISECT_HYP)
2750
  for name, value in instance.hvparams.items():
2751
    if name not in constants.HVC_GLOBALS:
2752
      config.set(constants.INISECT_HYP, name, str(value))
2753

    
2754
  config.add_section(constants.INISECT_BEP)
2755
  for name, value in instance.beparams.items():
2756
    config.set(constants.INISECT_BEP, name, str(value))
2757

    
2758
  config.add_section(constants.INISECT_OSP)
2759
  for name, value in instance.osparams.items():
2760
    config.set(constants.INISECT_OSP, name, str(value))
2761

    
2762
  utils.WriteFile(utils.PathJoin(destdir, constants.EXPORT_CONF_FILE),
2763
                  data=config.Dumps())
2764
  shutil.rmtree(finaldestdir, ignore_errors=True)
2765
  shutil.move(destdir, finaldestdir)
2766

    
2767

    
2768
def ExportInfo(dest):
2769
  """Get export configuration information.
2770

2771
  @type dest: str
2772
  @param dest: directory containing the export
2773

2774
  @rtype: L{objects.SerializableConfigParser}
2775
  @return: a serializable config file containing the
2776
      export info
2777

2778
  """
2779
  cff = utils.PathJoin(dest, constants.EXPORT_CONF_FILE)
2780

    
2781
  config = objects.SerializableConfigParser()
2782
  config.read(cff)
2783

    
2784
  if (not config.has_section(constants.INISECT_EXP) or
2785
      not config.has_section(constants.INISECT_INS)):
2786
    _Fail("Export info file doesn't have the required fields")
2787

    
2788
  return config.Dumps()
2789

    
2790

    
2791
def ListExports():
2792
  """Return a list of exports currently available on this machine.
2793

2794
  @rtype: list
2795
  @return: list of the exports
2796

2797
  """
2798
  if os.path.isdir(pathutils.EXPORT_DIR):
2799
    return sorted(utils.ListVisibleFiles(pathutils.EXPORT_DIR))
2800
  else:
2801
    _Fail("No exports directory")
2802

    
2803

    
2804
def RemoveExport(export):
2805
  """Remove an existing export from the node.
2806

2807
  @type export: str
2808
  @param export: the name of the export to remove
2809
  @rtype: None
2810

2811
  """
2812
  target = utils.PathJoin(pathutils.EXPORT_DIR, export)
2813

    
2814
  try:
2815
    shutil.rmtree(target)
2816
  except EnvironmentError, err:
2817
    _Fail("Error while removing the export: %s", err, exc=True)
2818

    
2819

    
2820
def BlockdevRename(devlist):
2821
  """Rename a list of block devices.
2822

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

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

    
2860

    
2861
def _TransformFileStorageDir(fs_dir):
2862
  """Checks whether given file_storage_dir is valid.
2863

2864
  Checks wheter the given fs_dir is within the cluster-wide default
2865
  file_storage_dir or the shared_file_storage_dir, which are stored in
2866
  SimpleStore. Only paths under those directories are allowed.
2867

2868
  @type fs_dir: str
2869
  @param fs_dir: the path to check
2870

2871
  @return: the normalized path if valid, None otherwise
2872

2873
  """
2874
  if not (constants.ENABLE_FILE_STORAGE or
2875
          constants.ENABLE_SHARED_FILE_STORAGE):
2876
    _Fail("File storage disabled at configure time")
2877

    
2878
  bdev.CheckFileStoragePath(fs_dir)
2879

    
2880
  return os.path.normpath(fs_dir)
2881

    
2882

    
2883
def CreateFileStorageDir(file_storage_dir):
2884
  """Create file storage directory.
2885

2886
  @type file_storage_dir: str
2887
  @param file_storage_dir: directory to create
2888

2889
  @rtype: tuple
2890
  @return: tuple with first element a boolean indicating wheter dir
2891
      creation was successful or not
2892

2893
  """
2894
  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
2895
  if os.path.exists(file_storage_dir):
2896
    if not os.path.isdir(file_storage_dir):
2897
      _Fail("Specified storage dir '%s' is not a directory",
2898
            file_storage_dir)
2899
  else:
2900
    try:
2901
      os.makedirs(file_storage_dir, 0750)
2902
    except OSError, err:
2903
      _Fail("Cannot create file storage directory '%s': %s",
2904
            file_storage_dir, err, exc=True)
2905

    
2906

    
2907
def RemoveFileStorageDir(file_storage_dir):
2908
  """Remove file storage directory.
2909

2910
  Remove it only if it's empty. If not log an error and return.
2911

2912
  @type file_storage_dir: str
2913
  @param file_storage_dir: the directory we should cleanup
2914
  @rtype: tuple (success,)
2915
  @return: tuple of one element, C{success}, denoting
2916
      whether the operation was successful
2917

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

    
2931

    
2932
def RenameFileStorageDir(old_file_storage_dir, new_file_storage_dir):
2933
  """Rename the file storage directory.
2934

2935
  @type old_file_storage_dir: str
2936
  @param old_file_storage_dir: the current path
2937
  @type new_file_storage_dir: str
2938
  @param new_file_storage_dir: the name we should rename to
2939
  @rtype: tuple (success,)
2940
  @return: tuple of one element, C{success}, denoting
2941
      whether the operation was successful
2942

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

    
2961

    
2962
def _EnsureJobQueueFile(file_name):
2963
  """Checks whether the given filename is in the queue directory.
2964

2965
  @type file_name: str
2966
  @param file_name: the file name we should check
2967
  @rtype: None
2968
  @raises RPCFail: if the file is not valid
2969

2970
  """
2971
  if not utils.IsBelowDir(pathutils.QUEUE_DIR, file_name):
2972
    _Fail("Passed job queue file '%s' does not belong to"
2973
          " the queue directory '%s'", file_name, pathutils.QUEUE_DIR)
2974

    
2975

    
2976
def JobQueueUpdate(file_name, content):
2977
  """Updates a file in the queue directory.
2978

2979
  This is just a wrapper over L{utils.io.WriteFile}, with proper
2980
  checking.
2981

2982
  @type file_name: str
2983
  @param file_name: the job file name
2984
  @type content: str
2985
  @param content: the new job contents
2986
  @rtype: boolean
2987
  @return: the success of the operation
2988

2989
  """
2990
  file_name = vcluster.LocalizeVirtualPath(file_name)
2991

    
2992
  _EnsureJobQueueFile(file_name)
2993
  getents = runtime.GetEnts()
2994

    
2995
  # Write and replace the file atomically
2996
  utils.WriteFile(file_name, data=_Decompress(content), uid=getents.masterd_uid,
2997
                  gid=getents.daemons_gid, mode=constants.JOB_QUEUE_FILES_PERMS)
2998

    
2999

    
3000
def JobQueueRename(old, new):
3001
  """Renames a job queue file.
3002

3003
  This is just a wrapper over os.rename with proper checking.
3004

3005
  @type old: str
3006
  @param old: the old (actual) file name
3007
  @type new: str
3008
  @param new: the desired file name
3009
  @rtype: tuple
3010
  @return: the success of the operation and payload
3011

3012
  """
3013
  old = vcluster.LocalizeVirtualPath(old)
3014
  new = vcluster.LocalizeVirtualPath(new)
3015

    
3016
  _EnsureJobQueueFile(old)
3017
  _EnsureJobQueueFile(new)
3018

    
3019
  getents = runtime.GetEnts()
3020

    
3021
  utils.RenameFile(old, new, mkdir=True, mkdir_mode=0750,
3022
                   dir_uid=getents.masterd_uid, dir_gid=getents.daemons_gid)
3023

    
3024

    
3025
def BlockdevClose(instance_name, disks):
3026
  """Closes the given block devices.
3027

3028
  This means they will be switched to secondary mode (in case of
3029
  DRBD).
3030

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

3041
  """
3042
  bdevs = []
3043
  for cf in disks:
3044
    rd = _RecursiveFindBD(cf)
3045
    if rd is None:
3046
      _Fail("Can't find device %s", cf)
3047
    bdevs.append(rd)
3048

    
3049
  msg = []
3050
  for rd in bdevs:
3051
    try:
3052
      rd.Close()
3053
    except errors.BlockDeviceError, err:
3054
      msg.append(str(err))
3055
  if msg:
3056
    _Fail("Can't make devices secondary: %s", ",".join(msg))
3057
  else:
3058
    if instance_name:
3059
      _RemoveBlockDevLinks(instance_name, disks)
3060

    
3061

    
3062
def ValidateHVParams(hvname, hvparams):
3063
  """Validates the given hypervisor parameters.
3064

3065
  @type hvname: string
3066
  @param hvname: the hypervisor name
3067
  @type hvparams: dict
3068
  @param hvparams: the hypervisor parameters to be validated
3069
  @rtype: None
3070

3071
  """
3072
  try:
3073
    hv_type = hypervisor.GetHypervisor(hvname)
3074
    hv_type.ValidateParameters(hvparams)
3075
  except errors.HypervisorError, err:
3076
    _Fail(str(err), log=False)
3077

    
3078

    
3079
def _CheckOSPList(os_obj, parameters):
3080
  """Check whether a list of parameters is supported by the OS.
3081

3082
  @type os_obj: L{objects.OS}
3083
  @param os_obj: OS object to check
3084
  @type parameters: list
3085
  @param parameters: the list of parameters to check
3086

3087
  """
3088
  supported = [v[0] for v in os_obj.supported_parameters]
3089
  delta = frozenset(parameters).difference(supported)
3090
  if delta:
3091
    _Fail("The following parameters are not supported"
3092
          " by the OS %s: %s" % (os_obj.name, utils.CommaJoin(delta)))
3093

    
3094

    
3095
def ValidateOS(required, osname, checks, osparams):
3096
  """Validate the given OS' parameters.
3097

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

3111
  """
3112
  if not constants.OS_VALIDATE_CALLS.issuperset(checks):
3113
    _Fail("Unknown checks required for OS %s: %s", osname,
3114
          set(checks).difference(constants.OS_VALIDATE_CALLS))
3115

    
3116
  name_only = objects.OS.GetName(osname)
3117
  status, tbv = _TryOSFromDisk(name_only, None)
3118

    
3119
  if not status:
3120
    if required:
3121
      _Fail(tbv)
3122
    else:
3123
      return False
3124

    
3125
  if max(tbv.api_versions) < constants.OS_API_V20:
3126
    return True
3127

    
3128
  if constants.OS_VALIDATE_PARAMETERS in checks:
3129
    _CheckOSPList(tbv, osparams.keys())
3130

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

    
3140
  return True
3141

    
3142

    
3143
def DemoteFromMC():
3144
  """Demotes the current node from master candidate role.
3145

3146
  """
3147
  # try to ensure we're not the master by mistake
3148
  master, myself = ssconf.GetMasterAndMyself()
3149
  if master == myself:
3150
    _Fail("ssconf status shows I'm the master node, will not demote")
3151

    
3152
  result = utils.RunCmd([pathutils.DAEMON_UTIL, "check", constants.MASTERD])
3153
  if not result.failed:
3154
    _Fail("The master daemon is running, will not demote")
3155

    
3156
  try:
3157
    if os.path.isfile(pathutils.CLUSTER_CONF_FILE):
3158
      utils.CreateBackup(pathutils.CLUSTER_CONF_FILE)
3159
  except EnvironmentError, err:
3160
    if err.errno != errno.ENOENT:
3161
      _Fail("Error while backing up cluster file: %s", err, exc=True)
3162

    
3163
  utils.RemoveFile(pathutils.CLUSTER_CONF_FILE)
3164

    
3165

    
3166
def _GetX509Filenames(cryptodir, name):
3167
  """Returns the full paths for the private key and certificate.
3168

3169
  """
3170
  return (utils.PathJoin(cryptodir, name),
3171
          utils.PathJoin(cryptodir, name, _X509_KEY_FILE),
3172
          utils.PathJoin(cryptodir, name, _X509_CERT_FILE))
3173

    
3174

    
3175
def CreateX509Certificate(validity, cryptodir=pathutils.CRYPTO_KEYS_DIR):
3176
  """Creates a new X509 certificate for SSL/TLS.
3177

3178
  @type validity: int
3179
  @param validity: Validity in seconds
3180
  @rtype: tuple; (string, string)
3181
  @return: Certificate name and public part
3182

3183
  """
3184
  (key_pem, cert_pem) = \
3185
    utils.GenerateSelfSignedX509Cert(netutils.Hostname.GetSysName(),
3186
                                     min(validity, _MAX_SSL_CERT_VALIDITY))
3187

    
3188
  cert_dir = tempfile.mkdtemp(dir=cryptodir,
3189
                              prefix="x509-%s-" % utils.TimestampForFilename())
3190
  try:
3191
    name = os.path.basename(cert_dir)
3192
    assert len(name) > 5
3193

    
3194
    (_, key_file, cert_file) = _GetX509Filenames(cryptodir, name)
3195

    
3196
    utils.WriteFile(key_file, mode=0400, data=key_pem)
3197
    utils.WriteFile(cert_file, mode=0400, data=cert_pem)
3198

    
3199
    # Never return private key as it shouldn't leave the node
3200
    return (name, cert_pem)
3201
  except Exception:
3202
    shutil.rmtree(cert_dir, ignore_errors=True)
3203
    raise
3204

    
3205

    
3206
def RemoveX509Certificate(name, cryptodir=pathutils.CRYPTO_KEYS_DIR):
3207
  """Removes a X509 certificate.
3208

3209
  @type name: string
3210
  @param name: Certificate name
3211

3212
  """
3213
  (cert_dir, key_file, cert_file) = _GetX509Filenames(cryptodir, name)
3214

    
3215
  utils.RemoveFile(key_file)
3216
  utils.RemoveFile(cert_file)
3217

    
3218
  try:
3219
    os.rmdir(cert_dir)
3220
  except EnvironmentError, err:
3221
    _Fail("Cannot remove certificate directory '%s': %s",
3222
          cert_dir, err)
3223

    
3224

    
3225
def _GetImportExportIoCommand(instance, mode, ieio, ieargs):
3226
  """Returns the command for the requested input/output.
3227

3228
  @type instance: L{objects.Instance}
3229
  @param instance: The instance object
3230
  @param mode: Import/export mode
3231
  @param ieio: Input/output type
3232
  @param ieargs: Input/output arguments
3233

3234
  """
3235
  assert mode in (constants.IEM_IMPORT, constants.IEM_EXPORT)
3236

    
3237
  env = None
3238
  prefix = None
3239
  suffix = None
3240
  exp_size = None
3241

    
3242
  if ieio == constants.IEIO_FILE:
3243
    (filename, ) = ieargs
3244

    
3245
    if not utils.IsNormAbsPath(filename):
3246
      _Fail("Path '%s' is not normalized or absolute", filename)
3247

    
3248
    real_filename = os.path.realpath(filename)
3249
    directory = os.path.dirname(real_filename)
3250

    
3251
    if not utils.IsBelowDir(pathutils.EXPORT_DIR, real_filename):
3252
      _Fail("File '%s' is not under exports directory '%s': %s",
3253
            filename, pathutils.EXPORT_DIR, real_filename)
3254

    
3255
    # Create directory
3256
    utils.Makedirs(directory, mode=0750)
3257

    
3258
    quoted_filename = utils.ShellQuote(filename)
3259

    
3260
    if mode == constants.IEM_IMPORT:
3261
      suffix = "> %s" % quoted_filename
3262
    elif mode == constants.IEM_EXPORT:
3263
      suffix = "< %s" % quoted_filename
3264

    
3265
      # Retrieve file size
3266
      try:
3267
        st = os.stat(filename)
3268
      except EnvironmentError, err:
3269
        logging.error("Can't stat(2) %s: %s", filename, err)
3270
      else:
3271
        exp_size = utils.BytesToMebibyte(st.st_size)
3272

    
3273
  elif ieio == constants.IEIO_RAW_DISK:
3274
    (disk, ) = ieargs
3275

    
3276
    real_disk = _OpenRealBD(disk)
3277

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

    
3290
    elif mode == constants.IEM_EXPORT:
3291
      # the block size on the read dd is 1MiB to match our units
3292
      prefix = utils.BuildShellCmd("dd if=%s bs=%s count=%s |",
3293
                                   real_disk.dev_path,
3294
                                   str(1024 * 1024), # 1 MB
3295
                                   str(disk.size))
3296
      exp_size = disk.size
3297

    
3298
  elif ieio == constants.IEIO_SCRIPT:
3299
    (disk, disk_index, ) = ieargs
3300

    
3301
    assert isinstance(disk_index, (int, long))
3302

    
3303
    real_disk = _OpenRealBD(disk)
3304

    
3305
    inst_os = OSFromDisk(instance.os)
3306
    env = OSEnvironment(instance, inst_os)
3307

    
3308
    if mode == constants.IEM_IMPORT:
3309
      env["IMPORT_DEVICE"] = env["DISK_%d_PATH" % disk_index]
3310
      env["IMPORT_INDEX"] = str(disk_index)
3311
      script = inst_os.import_script
3312

    
3313
    elif mode == constants.IEM_EXPORT:
3314
      env["EXPORT_DEVICE"] = real_disk.dev_path
3315
      env["EXPORT_INDEX"] = str(disk_index)
3316
      script = inst_os.export_script
3317

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

    
3321
    if mode == constants.IEM_IMPORT:
3322
      suffix = "| %s" % script_cmd
3323

    
3324
    elif mode == constants.IEM_EXPORT:
3325
      prefix = "%s |" % script_cmd
3326

    
3327
    # Let script predict size
3328
    exp_size = constants.IE_CUSTOM_SIZE
3329

    
3330
  else:
3331
    _Fail("Invalid %s I/O mode %r", mode, ieio)
3332

    
3333
  return (env, prefix, suffix, exp_size)
3334

    
3335

    
3336
def _CreateImportExportStatusDir(prefix):
3337
  """Creates status directory for import/export.
3338

3339
  """
3340
  return tempfile.mkdtemp(dir=pathutils.IMPORT_EXPORT_DIR,
3341
                          prefix=("%s-%s-" %
3342
                                  (prefix, utils.TimestampForFilename())))
3343

    
3344

    
3345
def StartImportExportDaemon(mode, opts, host, port, instance, component,
3346
                            ieio, ieioargs):
3347
  """Starts an import or export daemon.
3348

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

3364
  """
3365
  if mode == constants.IEM_IMPORT:
3366
    prefix = "import"
3367

    
3368
    if not (host is None and port is None):
3369
      _Fail("Can not specify host or port on import")
3370

    
3371
  elif mode == constants.IEM_EXPORT:
3372
    prefix = "export"
3373

    
3374
    if host is None or port is None:
3375
      _Fail("Host and port must be specified for an export")
3376

    
3377
  else:
3378
    _Fail("Invalid mode %r", mode)
3379

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

    
3383
  (cmd_env, cmd_prefix, cmd_suffix, exp_size) = \
3384
    _GetImportExportIoCommand(instance, mode, ieio, ieioargs)
3385

    
3386
  if opts.key_name is None:
3387
    # Use server.pem
3388
    key_path = pathutils.NODED_CERT_FILE
3389
    cert_path = pathutils.NODED_CERT_FILE
3390
    assert opts.ca_pem is None
3391
  else:
3392
    (_, key_path, cert_path) = _GetX509Filenames(pathutils.CRYPTO_KEYS_DIR,
3393
                                                 opts.key_name)
3394
    assert opts.ca_pem is not None
3395

    
3396
  for i in [key_path, cert_path]:
3397
    if not os.path.exists(i):
3398
      _Fail("File '%s' does not exist" % i)
3399

    
3400
  status_dir = _CreateImportExportStatusDir("%s-%s" % (prefix, component))
3401
  try:
3402
    status_file = utils.PathJoin(status_dir, _IES_STATUS_FILE)
3403
    pid_file = utils.PathJoin(status_dir, _IES_PID_FILE)
3404
    ca_file = utils.PathJoin(status_dir, _IES_CA_FILE)
3405

    
3406
    if opts.ca_pem is None:
3407
      # Use server.pem
3408
      ca = utils.ReadFile(pathutils.NODED_CERT_FILE)
3409
    else:
3410
      ca = opts.ca_pem
3411

    
3412
    # Write CA file
3413
    utils.WriteFile(ca_file, data=ca, mode=0400)
3414

    
3415
    cmd = [
3416
      pathutils.IMPORT_EXPORT_DAEMON,
3417
      status_file, mode,
3418
      "--key=%s" % key_path,
3419
      "--cert=%s" % cert_path,
3420
      "--ca=%s" % ca_file,
3421
      ]
3422

    
3423
    if host:
3424
      cmd.append("--host=%s" % host)
3425

    
3426
    if port:
3427
      cmd.append("--port=%s" % port)
3428

    
3429
    if opts.ipv6:
3430
      cmd.append("--ipv6")
3431
    else:
3432
      cmd.append("--ipv4")
3433

    
3434
    if opts.compress:
3435
      cmd.append("--compress=%s" % opts.compress)
3436

    
3437
    if opts.magic:
3438
      cmd.append("--magic=%s" % opts.magic)
3439

    
3440
    if exp_size is not None:
3441
      cmd.append("--expected-size=%s" % exp_size)
3442

    
3443
    if cmd_prefix:
3444
      cmd.append("--cmd-prefix=%s" % cmd_prefix)
3445

    
3446
    if cmd_suffix:
3447
      cmd.append("--cmd-suffix=%s" % cmd_suffix)
3448

    
3449
    if mode == constants.IEM_EXPORT:
3450
      # Retry connection a few times when connecting to remote peer
3451
      cmd.append("--connect-retries=%s" % constants.RIE_CONNECT_RETRIES)
3452
      cmd.append("--connect-timeout=%s" % constants.RIE_CONNECT_ATTEMPT_TIMEOUT)
3453
    elif opts.connect_timeout is not None:
3454
      assert mode == constants.IEM_IMPORT
3455
      # Overall timeout for establishing connection while listening
3456
      cmd.append("--connect-timeout=%s" % opts.connect_timeout)
3457

    
3458
    logfile = _InstanceLogName(prefix, instance.os, instance.name, component)
3459

    
3460
    # TODO: Once _InstanceLogName uses tempfile.mkstemp, StartDaemon has
3461
    # support for receiving a file descriptor for output
3462
    utils.StartDaemon(cmd, env=cmd_env, pidfile=pid_file,
3463
                      output=logfile)
3464

    
3465
    # The import/export name is simply the status directory name
3466
    return os.path.basename(status_dir)
3467

    
3468
  except Exception:
3469
    shutil.rmtree(status_dir, ignore_errors=True)
3470
    raise
3471

    
3472

    
3473
def GetImportExportStatus(names):
3474
  """Returns import/export daemon status.
3475

3476
  @type names: sequence
3477
  @param names: List of names
3478
  @rtype: List of dicts
3479
  @return: Returns a list of the state of each named import/export or None if a
3480
           status couldn't be read
3481

3482
  """
3483
  result = []
3484

    
3485
  for name in names:
3486
    status_file = utils.PathJoin(pathutils.IMPORT_EXPORT_DIR, name,
3487
                                 _IES_STATUS_FILE)
3488

    
3489
    try:
3490
      data = utils.ReadFile(status_file)
3491
    except EnvironmentError, err:
3492
      if err.errno != errno.ENOENT:
3493
        raise
3494
      data = None
3495

    
3496
    if not data:
3497
      result.append(None)
3498
      continue
3499

    
3500
    result.append(serializer.LoadJson(data))
3501

    
3502
  return result
3503

    
3504

    
3505
def AbortImportExport(name):
3506
  """Sends SIGTERM to a running import/export daemon.
3507

3508
  """
3509
  logging.info("Abort import/export %s", name)
3510

    
3511
  status_dir = utils.PathJoin(pathutils.IMPORT_EXPORT_DIR, name)
3512
  pid = utils.ReadLockedPidFile(utils.PathJoin(status_dir, _IES_PID_FILE))
3513

    
3514
  if pid:
3515
    logging.info("Import/export %s is running with PID %s, sending SIGTERM",
3516
                 name, pid)
3517
    utils.IgnoreProcessNotFound(os.kill, pid, signal.SIGTERM)
3518

    
3519

    
3520
def CleanupImportExport(name):
3521
  """Cleanup after an import or export.
3522

3523
  If the import/export daemon is still running it's killed. Afterwards the
3524
  whole status directory is removed.
3525

3526
  """
3527
  logging.info("Finalizing import/export %s", name)
3528

    
3529
  status_dir = utils.PathJoin(pathutils.IMPORT_EXPORT_DIR, name)
3530

    
3531
  pid = utils.ReadLockedPidFile(utils.PathJoin(status_dir, _IES_PID_FILE))
3532

    
3533
  if pid:
3534
    logging.info("Import/export %s is still running with PID %s",
3535
                 name, pid)
3536
    utils.KillProcess(pid, waitpid=False)
3537

    
3538
  shutil.rmtree(status_dir, ignore_errors=True)
3539

    
3540

    
3541
def _FindDisks(nodes_ip, disks):
3542
  """Sets the physical ID on disks and returns the block devices.
3543

3544
  """
3545
  # set the correct physical ID
3546
  my_name = netutils.Hostname.GetSysName()
3547
  for cf in disks:
3548
    cf.SetPhysicalID(my_name, nodes_ip)
3549

    
3550
  bdevs = []
3551

    
3552
  for cf in disks:
3553
    rd = _RecursiveFindBD(cf)
3554
    if rd is None:
3555
      _Fail("Can't find device %s", cf)
3556
    bdevs.append(rd)
3557
  return bdevs
3558

    
3559

    
3560
def DrbdDisconnectNet(nodes_ip, disks):
3561
  """Disconnects the network on a list of drbd devices.
3562

3563
  """
3564
  bdevs = _FindDisks(nodes_ip, disks)
3565

    
3566
  # disconnect disks
3567
  for rd in bdevs:
3568
    try:
3569
      rd.DisconnectNet()
3570
    except errors.BlockDeviceError, err:
3571
      _Fail("Can't change network configuration to standalone mode: %s",
3572
            err, exc=True)
3573

    
3574

    
3575
def DrbdAttachNet(nodes_ip, disks, instance_name, multimaster):
3576
  """Attaches the network on a list of drbd devices.
3577

3578
  """
3579
  bdevs = _FindDisks(nodes_ip, disks)
3580

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

    
3595
  # wait until the disks are connected; we need to retry the re-attach
3596
  # if the device becomes standalone, as this might happen if the one
3597
  # node disconnects and reconnects in a different mode before the
3598
  # other node reconnects; in this case, one or both of the nodes will
3599
  # decide it has wrong configuration and switch to standalone
3600

    
3601
  def _Attach():
3602
    all_connected = True
3603

    
3604
    for rd in bdevs:
3605
      stats = rd.GetProcStatus()
3606

    
3607
      all_connected = (all_connected and
3608
                       (stats.is_connected or stats.is_in_resync))
3609

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

    
3619
    if not all_connected:
3620
      raise utils.RetryAgain()
3621

    
3622
  try:
3623
    # Start with a delay of 100 miliseconds and go up to 5 seconds
3624
    utils.Retry(_Attach, (0.1, 1.5, 5.0), 2 * 60)
3625
  except utils.RetryTimeout:
3626
    _Fail("Timeout in disk reconnecting")
3627

    
3628
  if multimaster:
3629
    # change to primary mode
3630
    for rd in bdevs:
3631
      try:
3632
        rd.Open()
3633
      except errors.BlockDeviceError, err:
3634
        _Fail("Can't change to primary mode: %s", err)
3635

    
3636

    
3637
def DrbdWaitSync(nodes_ip, disks):
3638
  """Wait until DRBDs have synchronized.
3639

3640
  """
3641
  def _helper(rd):
3642
    stats = rd.GetProcStatus()
3643
    if not (stats.is_connected or stats.is_in_resync):
3644
      raise utils.RetryAgain()
3645
    return stats
3646

    
3647
  bdevs = _FindDisks(nodes_ip, disks)
3648

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

    
3664
  return (alldone, min_resync)
3665

    
3666

    
3667
def GetDrbdUsermodeHelper():
3668
  """Returns DRBD usermode helper currently configured.
3669

3670
  """
3671
  try:
3672
    return bdev.BaseDRBD.GetUsermodeHelper()
3673
  except errors.BlockDeviceError, err:
3674
    _Fail(str(err))
3675

    
3676

    
3677
def PowercycleNode(hypervisor_type):
3678
  """Hard-powercycle the node.
3679

3680
  Because we need to return first, and schedule the powercycle in the
3681
  background, we won't be able to report failures nicely.
3682

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

    
3700

    
3701
def _VerifyRestrictedCmdName(cmd):
3702
  """Verifies a restricted command name.
3703

3704
  @type cmd: string
3705
  @param cmd: Command name
3706
  @rtype: tuple; (boolean, string or None)
3707
  @return: The tuple's first element is the status; if C{False}, the second
3708
    element is an error message string, otherwise it's C{None}
3709

3710
  """
3711
  if not cmd.strip():
3712
    return (False, "Missing command name")
3713

    
3714
  if os.path.basename(cmd) != cmd:
3715
    return (False, "Invalid command name")
3716

    
3717
  if not constants.EXT_PLUGIN_MASK.match(cmd):
3718
    return (False, "Command name contains forbidden characters")
3719

    
3720
  return (True, None)
3721

    
3722

    
3723
def _CommonRestrictedCmdCheck(path, owner):
3724
  """Common checks for restricted command file system directories and files.
3725

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

3733
  """
3734
  if owner is None:
3735
    # Default to root as owner
3736
    owner = (0, 0)
3737

    
3738
  try:
3739
    st = os.stat(path)
3740
  except EnvironmentError, err:
3741
    return (False, "Can't stat(2) '%s': %s" % (path, err))
3742

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

    
3746
  if (st.st_uid, st.st_gid) != owner:
3747
    (owner_uid, owner_gid) = owner
3748
    return (False, "'%s' is not owned by %s:%s" % (path, owner_uid, owner_gid))
3749

    
3750
  return (True, st)
3751

    
3752

    
3753
def _VerifyRestrictedCmdDirectory(path, _owner=None):
3754
  """Verifies restricted command directory.
3755

3756
  @type path: string
3757
  @param path: Path to check
3758
  @rtype: tuple; (boolean, string or None)
3759
  @return: The tuple's first element is the status; if C{False}, the second
3760
    element is an error message string, otherwise it's C{None}
3761

3762
  """
3763
  (status, value) = _CommonRestrictedCmdCheck(path, _owner)
3764

    
3765
  if not status:
3766
    return (False, value)
3767

    
3768
  if not stat.S_ISDIR(value.st_mode):
3769
    return (False, "Path '%s' is not a directory" % path)
3770

    
3771
  return (True, None)
3772

    
3773

    
3774
def _VerifyRestrictedCmd(path, cmd, _owner=None):
3775
  """Verifies a whole restricted command and returns its executable filename.
3776

3777
  @type path: string
3778
  @param path: Directory containing restricted commands
3779
  @type cmd: string
3780
  @param cmd: Command name
3781
  @rtype: tuple; (boolean, string)
3782
  @return: The tuple's first element is the status; if C{False}, the second
3783
    element is an error message string, otherwise the second element is the
3784
    absolute path to the executable
3785

3786
  """
3787
  executable = utils.PathJoin(path, cmd)
3788

    
3789
  (status, msg) = _CommonRestrictedCmdCheck(executable, _owner)
3790

    
3791
  if not status:
3792
    return (False, msg)
3793

    
3794
  if not utils.IsExecutable(executable):
3795
    return (False, "access(2) thinks '%s' can't be executed" % executable)
3796

    
3797
  return (True, executable)
3798

    
3799

    
3800
def _PrepareRestrictedCmd(path, cmd,
3801
                          _verify_dir=_VerifyRestrictedCmdDirectory,
3802
                          _verify_name=_VerifyRestrictedCmdName,
3803
                          _verify_cmd=_VerifyRestrictedCmd):
3804
  """Performs a number of tests on a restricted command.
3805

3806
  @type path: string
3807
  @param path: Directory containing restricted commands
3808
  @type cmd: string
3809
  @param cmd: Command name
3810
  @return: Same as L{_VerifyRestrictedCmd}
3811

3812
  """
3813
  # Verify the directory first
3814
  (status, msg) = _verify_dir(path)
3815
  if status:
3816
    # Check command if everything was alright
3817
    (status, msg) = _verify_name(cmd)
3818

    
3819
  if not status:
3820
    return (False, msg)
3821

    
3822
  # Check actual executable
3823
  return _verify_cmd(path, cmd)
3824

    
3825

    
3826
def RunRestrictedCmd(cmd,
3827
                     _lock_timeout=_RCMD_LOCK_TIMEOUT,
3828
                     _lock_file=pathutils.RESTRICTED_COMMANDS_LOCK_FILE,
3829
                     _path=pathutils.RESTRICTED_COMMANDS_DIR,
3830
                     _sleep_fn=time.sleep,
3831
                     _prepare_fn=_PrepareRestrictedCmd,
3832
                     _runcmd_fn=utils.RunCmd,
3833
                     _enabled=constants.ENABLE_RESTRICTED_COMMANDS):
3834
  """Executes a restricted command after performing strict tests.
3835

3836
  @type cmd: string
3837
  @param cmd: Command name
3838
  @rtype: string
3839
  @return: Command output
3840
  @raise RPCFail: In case of an error
3841

3842
  """
3843
  logging.info("Preparing to run restricted command '%s'", cmd)
3844

    
3845
  if not _enabled:
3846
    _Fail("Restricted commands disabled at configure time")
3847

    
3848
  lock = None
3849
  try:
3850
    cmdresult = None
3851
    try:
3852
      lock = utils.FileLock.Open(_lock_file)
3853
      lock.Exclusive(blocking=True, timeout=_lock_timeout)
3854

    
3855
      (status, value) = _prepare_fn(_path, cmd)
3856

    
3857
      if status:
3858
        cmdresult = _runcmd_fn([value], env={}, reset_env=True,
3859
                               postfork_fn=lambda _: lock.Unlock())
3860
      else:
3861
        logging.error(value)
3862
    except Exception: # pylint: disable=W0703
3863
      # Keep original error in log
3864
      logging.exception("Caught exception")
3865

    
3866
    if cmdresult is None:
3867
      logging.info("Sleeping for %0.1f seconds before returning",
3868
                   _RCMD_INVALID_DELAY)
3869
      _sleep_fn(_RCMD_INVALID_DELAY)
3870

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

    
3884

    
3885
def SetWatcherPause(until, _filename=pathutils.WATCHER_PAUSEFILE):
3886
  """Creates or removes the watcher pause file.
3887

3888
  @type until: None or number
3889
  @param until: Unix timestamp saying until when the watcher shouldn't run
3890

3891
  """
3892
  if until is None:
3893
    logging.info("Received request to no longer pause watcher")
3894
    utils.RemoveFile(_filename)
3895
  else:
3896
    logging.info("Received request to pause watcher until %s", until)
3897

    
3898
    if not ht.TNumber(until):
3899
      _Fail("Duration must be numeric")
3900

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

    
3903

    
3904
class HooksRunner(object):
3905
  """Hook runner.
3906

3907
  This class is instantiated on the node side (ganeti-noded) and not
3908
  on the master side.
3909

3910
  """
3911
  def __init__(self, hooks_base_dir=None):
3912
    """Constructor for hooks runner.
3913

3914
    @type hooks_base_dir: str or None
3915
    @param hooks_base_dir: if not None, this overrides the
3916
        L{pathutils.HOOKS_BASE_DIR} (useful for unittests)
3917

3918
    """
3919
    if hooks_base_dir is None:
3920
      hooks_base_dir = pathutils.HOOKS_BASE_DIR
3921
    # yeah, _BASE_DIR is not valid for attributes, we use it like a
3922
    # constant
3923
    self._BASE_DIR = hooks_base_dir # pylint: disable=C0103
3924

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

3928
    """
3929
    assert len(node_list) == 1
3930
    node = node_list[0]
3931
    _, myself = ssconf.GetMasterAndMyself()
3932
    assert node == myself
3933

    
3934
    results = self.RunHooks(hpath, phase, env)
3935

    
3936
    # Return values in the form expected by HooksMaster
3937
    return {node: (None, False, results)}
3938

    
3939
  def RunHooks(self, hpath, phase, env):
3940
    """Run the scripts in the hooks directory.
3941

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

3957
    @raise errors.ProgrammerError: for invalid input
3958
        parameters
3959

3960
    """
3961
    if phase == constants.HOOKS_PHASE_PRE:
3962
      suffix = "pre"
3963
    elif phase == constants.HOOKS_PHASE_POST:
3964
      suffix = "post"
3965
    else:
3966
      _Fail("Unknown hooks phase '%s'", phase)
3967

    
3968
    subdir = "%s-%s.d" % (hpath, suffix)
3969
    dir_name = utils.PathJoin(self._BASE_DIR, subdir)
3970

    
3971
    results = []
3972

    
3973
    if not os.path.isdir(dir_name):
3974
      # for non-existing/non-dirs, we simply exit instead of logging a
3975
      # warning at every operation
3976
      return results
3977

    
3978
    runparts_results = utils.RunParts(dir_name, env=env, reset_env=True)
3979

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

    
3995
    return results
3996

    
3997

    
3998
class IAllocatorRunner(object):
3999
  """IAllocator runner.
4000

4001
  This class is instantiated on the node side (ganeti-noded) and not on
4002
  the master side.
4003

4004
  """
4005
  @staticmethod
4006
  def Run(name, idata):
4007
    """Run an iallocator script.
4008

4009
    @type name: str
4010
    @param name: the iallocator script name
4011
    @type idata: str
4012
    @param idata: the allocator input data
4013

4014
    @rtype: tuple
4015
    @return: two element tuple of:
4016
       - status
4017
       - either error message or stdout of allocator (for success)
4018

4019
    """
4020
    alloc_script = utils.FindFile(name, constants.IALLOCATOR_SEARCH_PATH,
4021
                                  os.path.isfile)
4022
    if alloc_script is None:
4023
      _Fail("iallocator module '%s' not found in the search path", name)
4024

    
4025
    fd, fin_name = tempfile.mkstemp(prefix="ganeti-iallocator.")
4026
    try:
4027
      os.write(fd, idata)
4028
      os.close(fd)
4029
      result = utils.RunCmd([alloc_script, fin_name])
4030
      if result.failed:
4031
        _Fail("iallocator module '%s' failed: %s, output '%s'",
4032
              name, result.fail_reason, result.output)
4033
    finally:
4034
      os.unlink(fin_name)
4035

    
4036
    return result.stdout
4037

    
4038

    
4039
class DevCacheManager(object):
4040
  """Simple class for managing a cache of block device information.
4041

4042
  """
4043
  _DEV_PREFIX = "/dev/"
4044
  _ROOT_DIR = pathutils.BDEV_CACHE_DIR
4045

    
4046
  @classmethod
4047
  def _ConvertPath(cls, dev_path):
4048
    """Converts a /dev/name path to the cache file name.
4049

4050
    This replaces slashes with underscores and strips the /dev
4051
    prefix. It then returns the full path to the cache file.
4052

4053
    @type dev_path: str
4054
    @param dev_path: the C{/dev/} path name
4055
    @rtype: str
4056
    @return: the converted path name
4057

4058
    """
4059
    if dev_path.startswith(cls._DEV_PREFIX):
4060
      dev_path = dev_path[len(cls._DEV_PREFIX):]
4061
    dev_path = dev_path.replace("/", "_")
4062
    fpath = utils.PathJoin(cls._ROOT_DIR, "bdev_%s" % dev_path)
4063
    return fpath
4064

    
4065
  @classmethod
4066
  def UpdateCache(cls, dev_path, owner, on_primary, iv_name):
4067
    """Updates the cache information for a given device.
4068

4069
    @type dev_path: str
4070
    @param dev_path: the pathname of the device
4071
    @type owner: str
4072
    @param owner: the owner (instance name) of the device
4073
    @type on_primary: bool
4074
    @param on_primary: whether this is the primary
4075
        node nor not
4076
    @type iv_name: str
4077
    @param iv_name: the instance-visible name of the
4078
        device, as in objects.Disk.iv_name
4079

4080
    @rtype: None
4081

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

    
4099
  @classmethod
4100
  def RemoveCache(cls, dev_path):
4101
    """Remove data for a dev_path.
4102

4103
    This is just a wrapper over L{utils.io.RemoveFile} with a converted
4104
    path name and logging.
4105

4106
    @type dev_path: str
4107
    @param dev_path: the pathname of the device
4108

4109
    @rtype: None
4110

4111
    """
4112
    if dev_path is None:
4113
      logging.error("DevCacheManager.RemoveCache got a None dev_path")
4114
      return
4115
    fpath = cls._ConvertPath(dev_path)
4116
    try:
4117
      utils.RemoveFile(fpath)
4118
    except EnvironmentError, err:
4119
      logging.exception("Can't update bdev cache for %s: %s", dev_path, err)