Statistics
| Branch: | Tag: | Revision:

root / lib / backend.py @ c1912a48

History | View | Annotate | Download (121.1 kB)

1
#
2
#
3

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

    
21

    
22
"""Functions used by the node daemon
23

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

29
"""
30

    
31
# pylint: disable=E1103
32

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

    
37

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

    
52
from ganeti import errors
53
from ganeti import utils
54
from ganeti import ssh
55
from ganeti import hypervisor
56
from ganeti import constants
57
from ganeti import bdev
58
from ganeti import objects
59
from ganeti import ssconf
60
from ganeti import serializer
61
from ganeti import netutils
62
from ganeti import runtime
63
from ganeti import mcpu
64
from ganeti import compat
65
from ganeti import pathutils
66
from ganeti import vcluster
67
from ganeti import ht
68

    
69

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

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

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

    
91
#: Maximum file permissions for remote 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 remote commands
97
_RCMD_INVALID_DELAY = 10
98

    
99
#: How long to wait to acquire lock for remote 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 _Fail(msg, *args, **kwargs):
114
  """Log an error and the raise an RPCFail exception.
115

116
  This exception is then handled specially in the ganeti daemon and
117
  turned into a 'failed' return type. As such, this function is a
118
  useful shortcut for logging the error and returning it to the master
119
  daemon.
120

121
  @type msg: string
122
  @param msg: the text of the exception
123
  @raise RPCFail
124

125
  """
126
  if args:
127
    msg = msg % args
128
  if "log" not in kwargs or kwargs["log"]: # if we should log this error
129
    if "exc" in kwargs and kwargs["exc"]:
130
      logging.exception(msg)
131
    else:
132
      logging.error(msg)
133
  raise RPCFail(msg)
134

    
135

    
136
def _GetConfig():
137
  """Simple wrapper to return a SimpleStore.
138

139
  @rtype: L{ssconf.SimpleStore}
140
  @return: a SimpleStore instance
141

142
  """
143
  return ssconf.SimpleStore()
144

    
145

    
146
def _GetSshRunner(cluster_name):
147
  """Simple wrapper to return an SshRunner.
148

149
  @type cluster_name: str
150
  @param cluster_name: the cluster name, which is needed
151
      by the SshRunner constructor
152
  @rtype: L{ssh.SshRunner}
153
  @return: an SshRunner instance
154

155
  """
156
  return ssh.SshRunner(cluster_name)
157

    
158

    
159
def _Decompress(data):
160
  """Unpacks data compressed by the RPC client.
161

162
  @type data: list or tuple
163
  @param data: Data sent by RPC client
164
  @rtype: str
165
  @return: Decompressed data
166

167
  """
168
  assert isinstance(data, (list, tuple))
169
  assert len(data) == 2
170
  (encoding, content) = data
171
  if encoding == constants.RPC_ENCODING_NONE:
172
    return content
173
  elif encoding == constants.RPC_ENCODING_ZLIB_BASE64:
174
    return zlib.decompress(base64.b64decode(content))
175
  else:
176
    raise AssertionError("Unknown data encoding")
177

    
178

    
179
def _CleanDirectory(path, exclude=None):
180
  """Removes all regular files in a directory.
181

182
  @type path: str
183
  @param path: the directory to clean
184
  @type exclude: list
185
  @param exclude: list of files to be excluded, defaults
186
      to the empty list
187

188
  """
189
  if path not in _ALLOWED_CLEAN_DIRS:
190
    _Fail("Path passed to _CleanDirectory not in allowed clean targets: '%s'",
191
          path)
192

    
193
  if not os.path.isdir(path):
194
    return
195
  if exclude is None:
196
    exclude = []
197
  else:
198
    # Normalize excluded paths
199
    exclude = [os.path.normpath(i) for i in exclude]
200

    
201
  for rel_name in utils.ListVisibleFiles(path):
202
    full_name = utils.PathJoin(path, rel_name)
203
    if full_name in exclude:
204
      continue
205
    if os.path.isfile(full_name) and not os.path.islink(full_name):
206
      utils.RemoveFile(full_name)
207

    
208

    
209
def _BuildUploadFileList():
210
  """Build the list of allowed upload files.
211

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

214
  """
215
  allowed_files = set([
216
    pathutils.CLUSTER_CONF_FILE,
217
    pathutils.ETC_HOSTS,
218
    pathutils.SSH_KNOWN_HOSTS_FILE,
219
    pathutils.VNC_PASSWORD_FILE,
220
    pathutils.RAPI_CERT_FILE,
221
    pathutils.SPICE_CERT_FILE,
222
    pathutils.SPICE_CACERT_FILE,
223
    pathutils.RAPI_USERS_FILE,
224
    pathutils.CONFD_HMAC_KEY,
225
    pathutils.CLUSTER_DOMAIN_SECRET_FILE,
226
    ])
227

    
228
  for hv_name in constants.HYPER_TYPES:
229
    hv_class = hypervisor.GetHypervisorClass(hv_name)
230
    allowed_files.update(hv_class.GetAncillaryFiles()[0])
231

    
232
  assert pathutils.FILE_STORAGE_PATHS_FILE not in allowed_files, \
233
    "Allowed file storage paths should never be uploaded via RPC"
234

    
235
  return frozenset(allowed_files)
236

    
237

    
238
_ALLOWED_UPLOAD_FILES = _BuildUploadFileList()
239

    
240

    
241
def JobQueuePurge():
242
  """Removes job queue files and archived jobs.
243

244
  @rtype: tuple
245
  @return: True, None
246

247
  """
248
  _CleanDirectory(pathutils.QUEUE_DIR, exclude=[pathutils.JOB_QUEUE_LOCK_FILE])
249
  _CleanDirectory(pathutils.JOB_QUEUE_ARCHIVE_DIR)
250

    
251

    
252
def GetMasterInfo():
253
  """Returns master information.
254

255
  This is an utility function to compute master information, either
256
  for consumption here or from the node daemon.
257

258
  @rtype: tuple
259
  @return: master_netdev, master_ip, master_name, primary_ip_family,
260
    master_netmask
261
  @raise RPCFail: in case of errors
262

263
  """
264
  try:
265
    cfg = _GetConfig()
266
    master_netdev = cfg.GetMasterNetdev()
267
    master_ip = cfg.GetMasterIP()
268
    master_netmask = cfg.GetMasterNetmask()
269
    master_node = cfg.GetMasterNode()
270
    primary_ip_family = cfg.GetPrimaryIPFamily()
271
  except errors.ConfigurationError, err:
272
    _Fail("Cluster configuration incomplete: %s", err, exc=True)
273
  return (master_netdev, master_ip, master_node, primary_ip_family,
274
          master_netmask)
275

    
276

    
277
def RunLocalHooks(hook_opcode, hooks_path, env_builder_fn):
278
  """Decorator that runs hooks before and after the decorated function.
279

280
  @type hook_opcode: string
281
  @param hook_opcode: opcode of the hook
282
  @type hooks_path: string
283
  @param hooks_path: path of the hooks
284
  @type env_builder_fn: function
285
  @param env_builder_fn: function that returns a dictionary containing the
286
    environment variables for the hooks. Will get all the parameters of the
287
    decorated function.
288
  @raise RPCFail: in case of pre-hook failure
289

290
  """
291
  def decorator(fn):
292
    def wrapper(*args, **kwargs):
293
      _, myself = ssconf.GetMasterAndMyself()
294
      nodes = ([myself], [myself])  # these hooks run locally
295

    
296
      env_fn = compat.partial(env_builder_fn, *args, **kwargs)
297

    
298
      cfg = _GetConfig()
299
      hr = HooksRunner()
300
      hm = mcpu.HooksMaster(hook_opcode, hooks_path, nodes, hr.RunLocalHooks,
301
                            None, env_fn, logging.warning, cfg.GetClusterName(),
302
                            cfg.GetMasterNode())
303

    
304
      hm.RunPhase(constants.HOOKS_PHASE_PRE)
305
      result = fn(*args, **kwargs)
306
      hm.RunPhase(constants.HOOKS_PHASE_POST)
307

    
308
      return result
309
    return wrapper
310
  return decorator
311

    
312

    
313
def _BuildMasterIpEnv(master_params, use_external_mip_script=None):
314
  """Builds environment variables for master IP hooks.
315

316
  @type master_params: L{objects.MasterNetworkParameters}
317
  @param master_params: network parameters of the master
318
  @type use_external_mip_script: boolean
319
  @param use_external_mip_script: whether to use an external master IP
320
    address setup script (unused, but necessary per the implementation of the
321
    _RunLocalHooks decorator)
322

323
  """
324
  # pylint: disable=W0613
325
  ver = netutils.IPAddress.GetVersionFromAddressFamily(master_params.ip_family)
326
  env = {
327
    "MASTER_NETDEV": master_params.netdev,
328
    "MASTER_IP": master_params.ip,
329
    "MASTER_NETMASK": str(master_params.netmask),
330
    "CLUSTER_IP_VERSION": str(ver),
331
  }
332

    
333
  return env
334

    
335

    
336
def _RunMasterSetupScript(master_params, action, use_external_mip_script):
337
  """Execute the master IP address setup script.
338

339
  @type master_params: L{objects.MasterNetworkParameters}
340
  @param master_params: network parameters of the master
341
  @type action: string
342
  @param action: action to pass to the script. Must be one of
343
    L{backend._MASTER_START} or L{backend._MASTER_STOP}
344
  @type use_external_mip_script: boolean
345
  @param use_external_mip_script: whether to use an external master IP
346
    address setup script
347
  @raise backend.RPCFail: if there are errors during the execution of the
348
    script
349

350
  """
351
  env = _BuildMasterIpEnv(master_params)
352

    
353
  if use_external_mip_script:
354
    setup_script = pathutils.EXTERNAL_MASTER_SETUP_SCRIPT
355
  else:
356
    setup_script = pathutils.DEFAULT_MASTER_SETUP_SCRIPT
357

    
358
  result = utils.RunCmd([setup_script, action], env=env, reset_env=True)
359

    
360
  if result.failed:
361
    _Fail("Failed to %s the master IP. Script return value: %s" %
362
          (action, result.exit_code), log=True)
363

    
364

    
365
@RunLocalHooks(constants.FAKE_OP_MASTER_TURNUP, "master-ip-turnup",
366
               _BuildMasterIpEnv)
367
def ActivateMasterIp(master_params, use_external_mip_script):
368
  """Activate the IP address of the master daemon.
369

370
  @type master_params: L{objects.MasterNetworkParameters}
371
  @param master_params: network parameters of the master
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 RPCFail: in case of errors during the IP startup
376

377
  """
378
  _RunMasterSetupScript(master_params, _MASTER_START,
379
                        use_external_mip_script)
380

    
381

    
382
def StartMasterDaemons(no_voting):
383
  """Activate local node as master node.
384

385
  The function will start the master daemons (ganeti-masterd and ganeti-rapi).
386

387
  @type no_voting: boolean
388
  @param no_voting: whether to start ganeti-masterd without a node vote
389
      but still non-interactively
390
  @rtype: None
391

392
  """
393

    
394
  if no_voting:
395
    masterd_args = "--no-voting --yes-do-it"
396
  else:
397
    masterd_args = ""
398

    
399
  env = {
400
    "EXTRA_MASTERD_ARGS": masterd_args,
401
    }
402

    
403
  result = utils.RunCmd([pathutils.DAEMON_UTIL, "start-master"], env=env)
404
  if result.failed:
405
    msg = "Can't start Ganeti master: %s" % result.output
406
    logging.error(msg)
407
    _Fail(msg)
408

    
409

    
410
@RunLocalHooks(constants.FAKE_OP_MASTER_TURNDOWN, "master-ip-turndown",
411
               _BuildMasterIpEnv)
412
def DeactivateMasterIp(master_params, use_external_mip_script):
413
  """Deactivate the master IP on this node.
414

415
  @type master_params: L{objects.MasterNetworkParameters}
416
  @param master_params: network parameters of the master
417
  @type use_external_mip_script: boolean
418
  @param use_external_mip_script: whether to use an external master IP
419
    address setup script
420
  @raise RPCFail: in case of errors during the IP turndown
421

422
  """
423
  _RunMasterSetupScript(master_params, _MASTER_STOP,
424
                        use_external_mip_script)
425

    
426

    
427
def StopMasterDaemons():
428
  """Stop the master daemons on this node.
429

430
  Stop the master daemons (ganeti-masterd and ganeti-rapi) on this node.
431

432
  @rtype: None
433

434
  """
435
  # TODO: log and report back to the caller the error failures; we
436
  # need to decide in which case we fail the RPC for this
437

    
438
  result = utils.RunCmd([pathutils.DAEMON_UTIL, "stop-master"])
439
  if result.failed:
440
    logging.error("Could not stop Ganeti master, command %s had exitcode %s"
441
                  " and error %s",
442
                  result.cmd, result.exit_code, result.output)
443

    
444

    
445
def ChangeMasterNetmask(old_netmask, netmask, master_ip, master_netdev):
446
  """Change the netmask of the master IP.
447

448
  @param old_netmask: the old value of the netmask
449
  @param netmask: the new value of the netmask
450
  @param master_ip: the master IP
451
  @param master_netdev: the master network device
452

453
  """
454
  if old_netmask == netmask:
455
    return
456

    
457
  if not netutils.IPAddress.Own(master_ip):
458
    _Fail("The master IP address is not up, not attempting to change its"
459
          " netmask")
460

    
461
  result = utils.RunCmd([constants.IP_COMMAND_PATH, "address", "add",
462
                         "%s/%s" % (master_ip, netmask),
463
                         "dev", master_netdev, "label",
464
                         "%s:0" % master_netdev])
465
  if result.failed:
466
    _Fail("Could not set the new netmask on the master IP address")
467

    
468
  result = utils.RunCmd([constants.IP_COMMAND_PATH, "address", "del",
469
                         "%s/%s" % (master_ip, old_netmask),
470
                         "dev", master_netdev, "label",
471
                         "%s:0" % master_netdev])
472
  if result.failed:
473
    _Fail("Could not bring down the master IP address with the old netmask")
474

    
475

    
476
def EtcHostsModify(mode, host, ip):
477
  """Modify a host entry in /etc/hosts.
478

479
  @param mode: The mode to operate. Either add or remove entry
480
  @param host: The host to operate on
481
  @param ip: The ip associated with the entry
482

483
  """
484
  if mode == constants.ETC_HOSTS_ADD:
485
    if not ip:
486
      RPCFail("Mode 'add' needs 'ip' parameter, but parameter not"
487
              " present")
488
    utils.AddHostToEtcHosts(host, ip)
489
  elif mode == constants.ETC_HOSTS_REMOVE:
490
    if ip:
491
      RPCFail("Mode 'remove' does not allow 'ip' parameter, but"
492
              " parameter is present")
493
    utils.RemoveHostFromEtcHosts(host)
494
  else:
495
    RPCFail("Mode not supported")
496

    
497

    
498
def LeaveCluster(modify_ssh_setup):
499
  """Cleans up and remove the current node.
500

501
  This function cleans up and prepares the current node to be removed
502
  from the cluster.
503

504
  If processing is successful, then it raises an
505
  L{errors.QuitGanetiException} which is used as a special case to
506
  shutdown the node daemon.
507

508
  @param modify_ssh_setup: boolean
509

510
  """
511
  _CleanDirectory(pathutils.DATA_DIR)
512
  _CleanDirectory(pathutils.CRYPTO_KEYS_DIR)
513
  JobQueuePurge()
514

    
515
  if modify_ssh_setup:
516
    try:
517
      priv_key, pub_key, auth_keys = ssh.GetUserFiles(constants.SSH_LOGIN_USER)
518

    
519
      utils.RemoveAuthorizedKey(auth_keys, utils.ReadFile(pub_key))
520

    
521
      utils.RemoveFile(priv_key)
522
      utils.RemoveFile(pub_key)
523
    except errors.OpExecError:
524
      logging.exception("Error while processing ssh files")
525

    
526
  try:
527
    utils.RemoveFile(pathutils.CONFD_HMAC_KEY)
528
    utils.RemoveFile(pathutils.RAPI_CERT_FILE)
529
    utils.RemoveFile(pathutils.SPICE_CERT_FILE)
530
    utils.RemoveFile(pathutils.SPICE_CACERT_FILE)
531
    utils.RemoveFile(pathutils.NODED_CERT_FILE)
532
  except: # pylint: disable=W0702
533
    logging.exception("Error while removing cluster secrets")
534

    
535
  result = utils.RunCmd([pathutils.DAEMON_UTIL, "stop", constants.CONFD])
536
  if result.failed:
537
    logging.error("Command %s failed with exitcode %s and error %s",
538
                  result.cmd, result.exit_code, result.output)
539

    
540
  # Raise a custom exception (handled in ganeti-noded)
541
  raise errors.QuitGanetiException(True, "Shutdown scheduled")
542

    
543

    
544
def _GetVgInfo(name):
545
  """Retrieves information about a LVM volume group.
546

547
  """
548
  # TODO: GetVGInfo supports returning information for multiple VGs at once
549
  vginfo = bdev.LogicalVolume.GetVGInfo([name])
550
  if vginfo:
551
    vg_free = int(round(vginfo[0][0], 0))
552
    vg_size = int(round(vginfo[0][1], 0))
553
  else:
554
    vg_free = None
555
    vg_size = None
556

    
557
  return {
558
    "name": name,
559
    "vg_free": vg_free,
560
    "vg_size": vg_size,
561
    }
562

    
563

    
564
def _GetHvInfo(name):
565
  """Retrieves node information from a hypervisor.
566

567
  The information returned depends on the hypervisor. Common items:
568

569
    - vg_size is the size of the configured volume group in MiB
570
    - vg_free is the free size of the volume group in MiB
571
    - memory_dom0 is the memory allocated for domain0 in MiB
572
    - memory_free is the currently available (free) ram in MiB
573
    - memory_total is the total number of ram in MiB
574
    - hv_version: the hypervisor version, if available
575

576
  """
577
  return hypervisor.GetHypervisor(name).GetNodeInfo()
578

    
579

    
580
def _GetNamedNodeInfo(names, fn):
581
  """Calls C{fn} for all names in C{names} and returns a dictionary.
582

583
  @rtype: None or dict
584

585
  """
586
  if names is None:
587
    return None
588
  else:
589
    return map(fn, names)
590

    
591

    
592
def GetNodeInfo(vg_names, hv_names):
593
  """Gives back a hash with different information about the node.
594

595
  @type vg_names: list of string
596
  @param vg_names: Names of the volume groups to ask for disk space information
597
  @type hv_names: list of string
598
  @param hv_names: Names of the hypervisors to ask for node information
599
  @rtype: tuple; (string, None/dict, None/dict)
600
  @return: Tuple containing boot ID, volume group information and hypervisor
601
    information
602

603
  """
604
  bootid = utils.ReadFile(_BOOT_ID_PATH, size=128).rstrip("\n")
605
  vg_info = _GetNamedNodeInfo(vg_names, _GetVgInfo)
606
  hv_info = _GetNamedNodeInfo(hv_names, _GetHvInfo)
607

    
608
  return (bootid, vg_info, hv_info)
609

    
610

    
611
def VerifyNode(what, cluster_name):
612
  """Verify the status of the local node.
613

614
  Based on the input L{what} parameter, various checks are done on the
615
  local node.
616

617
  If the I{filelist} key is present, this list of
618
  files is checksummed and the file/checksum pairs are returned.
619

620
  If the I{nodelist} key is present, we check that we have
621
  connectivity via ssh with the target nodes (and check the hostname
622
  report).
623

624
  If the I{node-net-test} key is present, we check that we have
625
  connectivity to the given nodes via both primary IP and, if
626
  applicable, secondary IPs.
627

628
  @type what: C{dict}
629
  @param what: a dictionary of things to check:
630
      - filelist: list of files for which to compute checksums
631
      - nodelist: list of nodes we should check ssh communication with
632
      - node-net-test: list of nodes we should check node daemon port
633
        connectivity with
634
      - hypervisor: list with hypervisors to run the verify for
635
  @rtype: dict
636
  @return: a dictionary with the same keys as the input dict, and
637
      values representing the result of the checks
638

639
  """
640
  result = {}
641
  my_name = netutils.Hostname.GetSysName()
642
  port = netutils.GetDaemonPort(constants.NODED)
643
  vm_capable = my_name not in what.get(constants.NV_VMNODES, [])
644

    
645
  if constants.NV_HYPERVISOR in what and vm_capable:
646
    result[constants.NV_HYPERVISOR] = tmp = {}
647
    for hv_name in what[constants.NV_HYPERVISOR]:
648
      try:
649
        val = hypervisor.GetHypervisor(hv_name).Verify()
650
      except errors.HypervisorError, err:
651
        val = "Error while checking hypervisor: %s" % str(err)
652
      tmp[hv_name] = val
653

    
654
  if constants.NV_HVPARAMS in what and vm_capable:
655
    result[constants.NV_HVPARAMS] = tmp = []
656
    for source, hv_name, hvparms in what[constants.NV_HVPARAMS]:
657
      try:
658
        logging.info("Validating hv %s, %s", hv_name, hvparms)
659
        hypervisor.GetHypervisor(hv_name).ValidateParameters(hvparms)
660
      except errors.HypervisorError, err:
661
        tmp.append((source, hv_name, str(err)))
662

    
663
  if constants.NV_FILELIST in what:
664
    fingerprints = utils.FingerprintFiles(map(vcluster.LocalizeVirtualPath,
665
                                              what[constants.NV_FILELIST]))
666
    result[constants.NV_FILELIST] = \
667
      dict((vcluster.MakeVirtualPath(key), value)
668
           for (key, value) in fingerprints.items())
669

    
670
  if constants.NV_NODELIST in what:
671
    (nodes, bynode) = what[constants.NV_NODELIST]
672

    
673
    # Add nodes from other groups (different for each node)
674
    try:
675
      nodes.extend(bynode[my_name])
676
    except KeyError:
677
      pass
678

    
679
    # Use a random order
680
    random.shuffle(nodes)
681

    
682
    # Try to contact all nodes
683
    val = {}
684
    for node in nodes:
685
      success, message = _GetSshRunner(cluster_name).VerifyNodeHostname(node)
686
      if not success:
687
        val[node] = message
688

    
689
    result[constants.NV_NODELIST] = val
690

    
691
  if constants.NV_NODENETTEST in what:
692
    result[constants.NV_NODENETTEST] = tmp = {}
693
    my_pip = my_sip = None
694
    for name, pip, sip in what[constants.NV_NODENETTEST]:
695
      if name == my_name:
696
        my_pip = pip
697
        my_sip = sip
698
        break
699
    if not my_pip:
700
      tmp[my_name] = ("Can't find my own primary/secondary IP"
701
                      " in the node list")
702
    else:
703
      for name, pip, sip in what[constants.NV_NODENETTEST]:
704
        fail = []
705
        if not netutils.TcpPing(pip, port, source=my_pip):
706
          fail.append("primary")
707
        if sip != pip:
708
          if not netutils.TcpPing(sip, port, source=my_sip):
709
            fail.append("secondary")
710
        if fail:
711
          tmp[name] = ("failure using the %s interface(s)" %
712
                       " and ".join(fail))
713

    
714
  if constants.NV_MASTERIP in what:
715
    # FIXME: add checks on incoming data structures (here and in the
716
    # rest of the function)
717
    master_name, master_ip = what[constants.NV_MASTERIP]
718
    if master_name == my_name:
719
      source = constants.IP4_ADDRESS_LOCALHOST
720
    else:
721
      source = None
722
    result[constants.NV_MASTERIP] = netutils.TcpPing(master_ip, port,
723
                                                     source=source)
724

    
725
  if constants.NV_USERSCRIPTS in what:
726
    result[constants.NV_USERSCRIPTS] = \
727
      [script for script in what[constants.NV_USERSCRIPTS]
728
       if not utils.IsExecutable(script)]
729

    
730
  if constants.NV_OOB_PATHS in what:
731
    result[constants.NV_OOB_PATHS] = tmp = []
732
    for path in what[constants.NV_OOB_PATHS]:
733
      try:
734
        st = os.stat(path)
735
      except OSError, err:
736
        tmp.append("error stating out of band helper: %s" % err)
737
      else:
738
        if stat.S_ISREG(st.st_mode):
739
          if stat.S_IMODE(st.st_mode) & stat.S_IXUSR:
740
            tmp.append(None)
741
          else:
742
            tmp.append("out of band helper %s is not executable" % path)
743
        else:
744
          tmp.append("out of band helper %s is not a file" % path)
745

    
746
  if constants.NV_LVLIST in what and vm_capable:
747
    try:
748
      val = GetVolumeList(utils.ListVolumeGroups().keys())
749
    except RPCFail, err:
750
      val = str(err)
751
    result[constants.NV_LVLIST] = val
752

    
753
  if constants.NV_INSTANCELIST in what and vm_capable:
754
    # GetInstanceList can fail
755
    try:
756
      val = GetInstanceList(what[constants.NV_INSTANCELIST])
757
    except RPCFail, err:
758
      val = str(err)
759
    result[constants.NV_INSTANCELIST] = val
760

    
761
  if constants.NV_VGLIST in what and vm_capable:
762
    result[constants.NV_VGLIST] = utils.ListVolumeGroups()
763

    
764
  if constants.NV_PVLIST in what and vm_capable:
765
    val = bdev.LogicalVolume.GetPVInfo(what[constants.NV_PVLIST],
766
                                       filter_allocatable=False)
767
    result[constants.NV_PVLIST] = map(objects.LvmPvInfo.ToDict, val)
768

    
769
  if constants.NV_VERSION in what:
770
    result[constants.NV_VERSION] = (constants.PROTOCOL_VERSION,
771
                                    constants.RELEASE_VERSION)
772

    
773
  if constants.NV_HVINFO in what and vm_capable:
774
    hyper = hypervisor.GetHypervisor(what[constants.NV_HVINFO])
775
    result[constants.NV_HVINFO] = hyper.GetNodeInfo()
776

    
777
  if constants.NV_DRBDLIST in what and vm_capable:
778
    try:
779
      used_minors = bdev.DRBD8.GetUsedDevs().keys()
780
    except errors.BlockDeviceError, err:
781
      logging.warning("Can't get used minors list", exc_info=True)
782
      used_minors = str(err)
783
    result[constants.NV_DRBDLIST] = used_minors
784

    
785
  if constants.NV_DRBDHELPER in what and vm_capable:
786
    status = True
787
    try:
788
      payload = bdev.BaseDRBD.GetUsermodeHelper()
789
    except errors.BlockDeviceError, err:
790
      logging.error("Can't get DRBD usermode helper: %s", str(err))
791
      status = False
792
      payload = str(err)
793
    result[constants.NV_DRBDHELPER] = (status, payload)
794

    
795
  if constants.NV_NODESETUP in what:
796
    result[constants.NV_NODESETUP] = tmpr = []
797
    if not os.path.isdir("/sys/block") or not os.path.isdir("/sys/class/net"):
798
      tmpr.append("The sysfs filesytem doesn't seem to be mounted"
799
                  " under /sys, missing required directories /sys/block"
800
                  " and /sys/class/net")
801
    if (not os.path.isdir("/proc/sys") or
802
        not os.path.isfile("/proc/sysrq-trigger")):
803
      tmpr.append("The procfs filesystem doesn't seem to be mounted"
804
                  " under /proc, missing required directory /proc/sys and"
805
                  " the file /proc/sysrq-trigger")
806

    
807
  if constants.NV_TIME in what:
808
    result[constants.NV_TIME] = utils.SplitTime(time.time())
809

    
810
  if constants.NV_OSLIST in what and vm_capable:
811
    result[constants.NV_OSLIST] = DiagnoseOS()
812

    
813
  if constants.NV_BRIDGES in what and vm_capable:
814
    result[constants.NV_BRIDGES] = [bridge
815
                                    for bridge in what[constants.NV_BRIDGES]
816
                                    if not utils.BridgeExists(bridge)]
817

    
818
  if what.get(constants.NV_FILE_STORAGE_PATHS) == my_name:
819
    result[constants.NV_FILE_STORAGE_PATHS] = \
820
      bdev.ComputeWrongFileStoragePaths()
821

    
822
  return result
823

    
824

    
825
def GetBlockDevSizes(devices):
826
  """Return the size of the given block devices
827

828
  @type devices: list
829
  @param devices: list of block device nodes to query
830
  @rtype: dict
831
  @return:
832
    dictionary of all block devices under /dev (key). The value is their
833
    size in MiB.
834

835
    {'/dev/disk/by-uuid/123456-12321231-312312-312': 124}
836

837
  """
838
  DEV_PREFIX = "/dev/"
839
  blockdevs = {}
840

    
841
  for devpath in devices:
842
    if not utils.IsBelowDir(DEV_PREFIX, devpath):
843
      continue
844

    
845
    try:
846
      st = os.stat(devpath)
847
    except EnvironmentError, err:
848
      logging.warning("Error stat()'ing device %s: %s", devpath, str(err))
849
      continue
850

    
851
    if stat.S_ISBLK(st.st_mode):
852
      result = utils.RunCmd(["blockdev", "--getsize64", devpath])
853
      if result.failed:
854
        # We don't want to fail, just do not list this device as available
855
        logging.warning("Cannot get size for block device %s", devpath)
856
        continue
857

    
858
      size = int(result.stdout) / (1024 * 1024)
859
      blockdevs[devpath] = size
860
  return blockdevs
861

    
862

    
863
def GetVolumeList(vg_names):
864
  """Compute list of logical volumes and their size.
865

866
  @type vg_names: list
867
  @param vg_names: the volume groups whose LVs we should list, or
868
      empty for all volume groups
869
  @rtype: dict
870
  @return:
871
      dictionary of all partions (key) with value being a tuple of
872
      their size (in MiB), inactive and online status::
873

874
        {'xenvg/test1': ('20.06', True, True)}
875

876
      in case of errors, a string is returned with the error
877
      details.
878

879
  """
880
  lvs = {}
881
  sep = "|"
882
  if not vg_names:
883
    vg_names = []
884
  result = utils.RunCmd(["lvs", "--noheadings", "--units=m", "--nosuffix",
885
                         "--separator=%s" % sep,
886
                         "-ovg_name,lv_name,lv_size,lv_attr"] + vg_names)
887
  if result.failed:
888
    _Fail("Failed to list logical volumes, lvs output: %s", result.output)
889

    
890
  for line in result.stdout.splitlines():
891
    line = line.strip()
892
    match = _LVSLINE_REGEX.match(line)
893
    if not match:
894
      logging.error("Invalid line returned from lvs output: '%s'", line)
895
      continue
896
    vg_name, name, size, attr = match.groups()
897
    inactive = attr[4] == "-"
898
    online = attr[5] == "o"
899
    virtual = attr[0] == "v"
900
    if virtual:
901
      # we don't want to report such volumes as existing, since they
902
      # don't really hold data
903
      continue
904
    lvs[vg_name + "/" + name] = (size, inactive, online)
905

    
906
  return lvs
907

    
908

    
909
def ListVolumeGroups():
910
  """List the volume groups and their size.
911

912
  @rtype: dict
913
  @return: dictionary with keys volume name and values the
914
      size of the volume
915

916
  """
917
  return utils.ListVolumeGroups()
918

    
919

    
920
def NodeVolumes():
921
  """List all volumes on this node.
922

923
  @rtype: list
924
  @return:
925
    A list of dictionaries, each having four keys:
926
      - name: the logical volume name,
927
      - size: the size of the logical volume
928
      - dev: the physical device on which the LV lives
929
      - vg: the volume group to which it belongs
930

931
    In case of errors, we return an empty list and log the
932
    error.
933

934
    Note that since a logical volume can live on multiple physical
935
    volumes, the resulting list might include a logical volume
936
    multiple times.
937

938
  """
939
  result = utils.RunCmd(["lvs", "--noheadings", "--units=m", "--nosuffix",
940
                         "--separator=|",
941
                         "--options=lv_name,lv_size,devices,vg_name"])
942
  if result.failed:
943
    _Fail("Failed to list logical volumes, lvs output: %s",
944
          result.output)
945

    
946
  def parse_dev(dev):
947
    return dev.split("(")[0]
948

    
949
  def handle_dev(dev):
950
    return [parse_dev(x) for x in dev.split(",")]
951

    
952
  def map_line(line):
953
    line = [v.strip() for v in line]
954
    return [{"name": line[0], "size": line[1],
955
             "dev": dev, "vg": line[3]} for dev in handle_dev(line[2])]
956

    
957
  all_devs = []
958
  for line in result.stdout.splitlines():
959
    if line.count("|") >= 3:
960
      all_devs.extend(map_line(line.split("|")))
961
    else:
962
      logging.warning("Strange line in the output from lvs: '%s'", line)
963
  return all_devs
964

    
965

    
966
def BridgesExist(bridges_list):
967
  """Check if a list of bridges exist on the current node.
968

969
  @rtype: boolean
970
  @return: C{True} if all of them exist, C{False} otherwise
971

972
  """
973
  missing = []
974
  for bridge in bridges_list:
975
    if not utils.BridgeExists(bridge):
976
      missing.append(bridge)
977

    
978
  if missing:
979
    _Fail("Missing bridges %s", utils.CommaJoin(missing))
980

    
981

    
982
def GetInstanceList(hypervisor_list):
983
  """Provides a list of instances.
984

985
  @type hypervisor_list: list
986
  @param hypervisor_list: the list of hypervisors to query information
987

988
  @rtype: list
989
  @return: a list of all running instances on the current node
990
    - instance1.example.com
991
    - instance2.example.com
992

993
  """
994
  results = []
995
  for hname in hypervisor_list:
996
    try:
997
      names = hypervisor.GetHypervisor(hname).ListInstances()
998
      results.extend(names)
999
    except errors.HypervisorError, err:
1000
      _Fail("Error enumerating instances (hypervisor %s): %s",
1001
            hname, err, exc=True)
1002

    
1003
  return results
1004

    
1005

    
1006
def GetInstanceInfo(instance, hname):
1007
  """Gives back the information about an instance as a dictionary.
1008

1009
  @type instance: string
1010
  @param instance: the instance name
1011
  @type hname: string
1012
  @param hname: the hypervisor type of the instance
1013

1014
  @rtype: dict
1015
  @return: dictionary with the following keys:
1016
      - memory: memory size of instance (int)
1017
      - state: xen state of instance (string)
1018
      - time: cpu time of instance (float)
1019
      - vcpus: the number of vcpus (int)
1020

1021
  """
1022
  output = {}
1023

    
1024
  iinfo = hypervisor.GetHypervisor(hname).GetInstanceInfo(instance)
1025
  if iinfo is not None:
1026
    output["memory"] = iinfo[2]
1027
    output["vcpus"] = iinfo[3]
1028
    output["state"] = iinfo[4]
1029
    output["time"] = iinfo[5]
1030

    
1031
  return output
1032

    
1033

    
1034
def GetInstanceMigratable(instance):
1035
  """Gives whether an instance can be migrated.
1036

1037
  @type instance: L{objects.Instance}
1038
  @param instance: object representing the instance to be checked.
1039

1040
  @rtype: tuple
1041
  @return: tuple of (result, description) where:
1042
      - result: whether the instance can be migrated or not
1043
      - description: a description of the issue, if relevant
1044

1045
  """
1046
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1047
  iname = instance.name
1048
  if iname not in hyper.ListInstances():
1049
    _Fail("Instance %s is not running", iname)
1050

    
1051
  for idx in range(len(instance.disks)):
1052
    link_name = _GetBlockDevSymlinkPath(iname, idx)
1053
    if not os.path.islink(link_name):
1054
      logging.warning("Instance %s is missing symlink %s for disk %d",
1055
                      iname, link_name, idx)
1056

    
1057

    
1058
def GetAllInstancesInfo(hypervisor_list):
1059
  """Gather data about all instances.
1060

1061
  This is the equivalent of L{GetInstanceInfo}, except that it
1062
  computes data for all instances at once, thus being faster if one
1063
  needs data about more than one instance.
1064

1065
  @type hypervisor_list: list
1066
  @param hypervisor_list: list of hypervisors to query for instance data
1067

1068
  @rtype: dict
1069
  @return: dictionary of instance: data, with data having 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
1074

1075
  """
1076
  output = {}
1077

    
1078
  for hname in hypervisor_list:
1079
    iinfo = hypervisor.GetHypervisor(hname).GetAllInstancesInfo()
1080
    if iinfo:
1081
      for name, _, memory, vcpus, state, times in iinfo:
1082
        value = {
1083
          "memory": memory,
1084
          "vcpus": vcpus,
1085
          "state": state,
1086
          "time": times,
1087
          }
1088
        if name in output:
1089
          # we only check static parameters, like memory and vcpus,
1090
          # and not state and time which can change between the
1091
          # invocations of the different hypervisors
1092
          for key in "memory", "vcpus":
1093
            if value[key] != output[name][key]:
1094
              _Fail("Instance %s is running twice"
1095
                    " with different parameters", name)
1096
        output[name] = value
1097

    
1098
  return output
1099

    
1100

    
1101
def _InstanceLogName(kind, os_name, instance, component):
1102
  """Compute the OS log filename for a given instance and operation.
1103

1104
  The instance name and os name are passed in as strings since not all
1105
  operations have these as part of an instance object.
1106

1107
  @type kind: string
1108
  @param kind: the operation type (e.g. add, import, etc.)
1109
  @type os_name: string
1110
  @param os_name: the os name
1111
  @type instance: string
1112
  @param instance: the name of the instance being imported/added/etc.
1113
  @type component: string or None
1114
  @param component: the name of the component of the instance being
1115
      transferred
1116

1117
  """
1118
  # TODO: Use tempfile.mkstemp to create unique filename
1119
  if component:
1120
    assert "/" not in component
1121
    c_msg = "-%s" % component
1122
  else:
1123
    c_msg = ""
1124
  base = ("%s-%s-%s%s-%s.log" %
1125
          (kind, os_name, instance, c_msg, utils.TimestampForFilename()))
1126
  return utils.PathJoin(pathutils.LOG_OS_DIR, base)
1127

    
1128

    
1129
def InstanceOsAdd(instance, reinstall, debug):
1130
  """Add an OS to an instance.
1131

1132
  @type instance: L{objects.Instance}
1133
  @param instance: Instance whose OS is to be installed
1134
  @type reinstall: boolean
1135
  @param reinstall: whether this is an instance reinstall
1136
  @type debug: integer
1137
  @param debug: debug level, passed to the OS scripts
1138
  @rtype: None
1139

1140
  """
1141
  inst_os = OSFromDisk(instance.os)
1142

    
1143
  create_env = OSEnvironment(instance, inst_os, debug)
1144
  if reinstall:
1145
    create_env["INSTANCE_REINSTALL"] = "1"
1146

    
1147
  logfile = _InstanceLogName("add", instance.os, instance.name, None)
1148

    
1149
  result = utils.RunCmd([inst_os.create_script], env=create_env,
1150
                        cwd=inst_os.path, output=logfile, reset_env=True)
1151
  if result.failed:
1152
    logging.error("os create command '%s' returned error: %s, logfile: %s,"
1153
                  " output: %s", result.cmd, result.fail_reason, logfile,
1154
                  result.output)
1155
    lines = [utils.SafeEncode(val)
1156
             for val in utils.TailFile(logfile, lines=20)]
1157
    _Fail("OS create script failed (%s), last lines in the"
1158
          " log file:\n%s", result.fail_reason, "\n".join(lines), log=False)
1159

    
1160

    
1161
def RunRenameInstance(instance, old_name, debug):
1162
  """Run the OS rename script for an instance.
1163

1164
  @type instance: L{objects.Instance}
1165
  @param instance: Instance whose OS is to be installed
1166
  @type old_name: string
1167
  @param old_name: previous instance name
1168
  @type debug: integer
1169
  @param debug: debug level, passed to the OS scripts
1170
  @rtype: boolean
1171
  @return: the success of the operation
1172

1173
  """
1174
  inst_os = OSFromDisk(instance.os)
1175

    
1176
  rename_env = OSEnvironment(instance, inst_os, debug)
1177
  rename_env["OLD_INSTANCE_NAME"] = old_name
1178

    
1179
  logfile = _InstanceLogName("rename", instance.os,
1180
                             "%s-%s" % (old_name, instance.name), None)
1181

    
1182
  result = utils.RunCmd([inst_os.rename_script], env=rename_env,
1183
                        cwd=inst_os.path, output=logfile, reset_env=True)
1184

    
1185
  if result.failed:
1186
    logging.error("os create command '%s' returned error: %s output: %s",
1187
                  result.cmd, result.fail_reason, result.output)
1188
    lines = [utils.SafeEncode(val)
1189
             for val in utils.TailFile(logfile, lines=20)]
1190
    _Fail("OS rename script failed (%s), last lines in the"
1191
          " log file:\n%s", result.fail_reason, "\n".join(lines), log=False)
1192

    
1193

    
1194
def _GetBlockDevSymlinkPath(instance_name, idx):
1195
  return utils.PathJoin(pathutils.DISK_LINKS_DIR, "%s%s%d" %
1196
                        (instance_name, constants.DISK_SEPARATOR, idx))
1197

    
1198

    
1199
def _SymlinkBlockDev(instance_name, device_path, idx):
1200
  """Set up symlinks to a instance's block device.
1201

1202
  This is an auxiliary function run when an instance is start (on the primary
1203
  node) or when an instance is migrated (on the target node).
1204

1205

1206
  @param instance_name: the name of the target instance
1207
  @param device_path: path of the physical block device, on the node
1208
  @param idx: the disk index
1209
  @return: absolute path to the disk's symlink
1210

1211
  """
1212
  link_name = _GetBlockDevSymlinkPath(instance_name, idx)
1213
  try:
1214
    os.symlink(device_path, link_name)
1215
  except OSError, err:
1216
    if err.errno == errno.EEXIST:
1217
      if (not os.path.islink(link_name) or
1218
          os.readlink(link_name) != device_path):
1219
        os.remove(link_name)
1220
        os.symlink(device_path, link_name)
1221
    else:
1222
      raise
1223

    
1224
  return link_name
1225

    
1226

    
1227
def _RemoveBlockDevLinks(instance_name, disks):
1228
  """Remove the block device symlinks belonging to the given instance.
1229

1230
  """
1231
  for idx, _ in enumerate(disks):
1232
    link_name = _GetBlockDevSymlinkPath(instance_name, idx)
1233
    if os.path.islink(link_name):
1234
      try:
1235
        os.remove(link_name)
1236
      except OSError:
1237
        logging.exception("Can't remove symlink '%s'", link_name)
1238

    
1239

    
1240
def _GatherAndLinkBlockDevs(instance):
1241
  """Set up an instance's block device(s).
1242

1243
  This is run on the primary node at instance startup. The block
1244
  devices must be already assembled.
1245

1246
  @type instance: L{objects.Instance}
1247
  @param instance: the instance whose disks we shoul assemble
1248
  @rtype: list
1249
  @return: list of (disk_object, device_path)
1250

1251
  """
1252
  block_devices = []
1253
  for idx, disk in enumerate(instance.disks):
1254
    device = _RecursiveFindBD(disk)
1255
    if device is None:
1256
      raise errors.BlockDeviceError("Block device '%s' is not set up." %
1257
                                    str(disk))
1258
    device.Open()
1259
    try:
1260
      link_name = _SymlinkBlockDev(instance.name, device.dev_path, idx)
1261
    except OSError, e:
1262
      raise errors.BlockDeviceError("Cannot create block device symlink: %s" %
1263
                                    e.strerror)
1264

    
1265
    block_devices.append((disk, link_name))
1266

    
1267
  return block_devices
1268

    
1269

    
1270
def StartInstance(instance, startup_paused):
1271
  """Start an instance.
1272

1273
  @type instance: L{objects.Instance}
1274
  @param instance: the instance object
1275
  @type startup_paused: bool
1276
  @param instance: pause instance at startup?
1277
  @rtype: None
1278

1279
  """
1280
  running_instances = GetInstanceList([instance.hypervisor])
1281

    
1282
  if instance.name in running_instances:
1283
    logging.info("Instance %s already running, not starting", instance.name)
1284
    return
1285

    
1286
  try:
1287
    block_devices = _GatherAndLinkBlockDevs(instance)
1288
    hyper = hypervisor.GetHypervisor(instance.hypervisor)
1289
    hyper.StartInstance(instance, block_devices, startup_paused)
1290
  except errors.BlockDeviceError, err:
1291
    _Fail("Block device error: %s", err, exc=True)
1292
  except errors.HypervisorError, err:
1293
    _RemoveBlockDevLinks(instance.name, instance.disks)
1294
    _Fail("Hypervisor error: %s", err, exc=True)
1295

    
1296

    
1297
def InstanceShutdown(instance, timeout):
1298
  """Shut an instance down.
1299

1300
  @note: this functions uses polling with a hardcoded timeout.
1301

1302
  @type instance: L{objects.Instance}
1303
  @param instance: the instance object
1304
  @type timeout: integer
1305
  @param timeout: maximum timeout for soft shutdown
1306
  @rtype: None
1307

1308
  """
1309
  hv_name = instance.hypervisor
1310
  hyper = hypervisor.GetHypervisor(hv_name)
1311
  iname = instance.name
1312

    
1313
  if instance.name not in hyper.ListInstances():
1314
    logging.info("Instance %s not running, doing nothing", iname)
1315
    return
1316

    
1317
  class _TryShutdown:
1318
    def __init__(self):
1319
      self.tried_once = False
1320

    
1321
    def __call__(self):
1322
      if iname not in hyper.ListInstances():
1323
        return
1324

    
1325
      try:
1326
        hyper.StopInstance(instance, retry=self.tried_once)
1327
      except errors.HypervisorError, err:
1328
        if iname not in hyper.ListInstances():
1329
          # if the instance is no longer existing, consider this a
1330
          # success and go to cleanup
1331
          return
1332

    
1333
        _Fail("Failed to stop instance %s: %s", iname, err)
1334

    
1335
      self.tried_once = True
1336

    
1337
      raise utils.RetryAgain()
1338

    
1339
  try:
1340
    utils.Retry(_TryShutdown(), 5, timeout)
1341
  except utils.RetryTimeout:
1342
    # the shutdown did not succeed
1343
    logging.error("Shutdown of '%s' unsuccessful, forcing", iname)
1344

    
1345
    try:
1346
      hyper.StopInstance(instance, force=True)
1347
    except errors.HypervisorError, err:
1348
      if iname in hyper.ListInstances():
1349
        # only raise an error if the instance still exists, otherwise
1350
        # the error could simply be "instance ... unknown"!
1351
        _Fail("Failed to force stop instance %s: %s", iname, err)
1352

    
1353
    time.sleep(1)
1354

    
1355
    if iname in hyper.ListInstances():
1356
      _Fail("Could not shutdown instance %s even by destroy", iname)
1357

    
1358
  try:
1359
    hyper.CleanupInstance(instance.name)
1360
  except errors.HypervisorError, err:
1361
    logging.warning("Failed to execute post-shutdown cleanup step: %s", err)
1362

    
1363
  _RemoveBlockDevLinks(iname, instance.disks)
1364

    
1365

    
1366
def InstanceReboot(instance, reboot_type, shutdown_timeout):
1367
  """Reboot an instance.
1368

1369
  @type instance: L{objects.Instance}
1370
  @param instance: the instance object to reboot
1371
  @type reboot_type: str
1372
  @param reboot_type: the type of reboot, one the following
1373
    constants:
1374
      - L{constants.INSTANCE_REBOOT_SOFT}: only reboot the
1375
        instance OS, do not recreate the VM
1376
      - L{constants.INSTANCE_REBOOT_HARD}: tear down and
1377
        restart the VM (at the hypervisor level)
1378
      - the other reboot type (L{constants.INSTANCE_REBOOT_FULL}) is
1379
        not accepted here, since that mode is handled differently, in
1380
        cmdlib, and translates into full stop and start of the
1381
        instance (instead of a call_instance_reboot RPC)
1382
  @type shutdown_timeout: integer
1383
  @param shutdown_timeout: maximum timeout for soft shutdown
1384
  @rtype: None
1385

1386
  """
1387
  running_instances = GetInstanceList([instance.hypervisor])
1388

    
1389
  if instance.name not in running_instances:
1390
    _Fail("Cannot reboot instance %s that is not running", instance.name)
1391

    
1392
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1393
  if reboot_type == constants.INSTANCE_REBOOT_SOFT:
1394
    try:
1395
      hyper.RebootInstance(instance)
1396
    except errors.HypervisorError, err:
1397
      _Fail("Failed to soft reboot instance %s: %s", instance.name, err)
1398
  elif reboot_type == constants.INSTANCE_REBOOT_HARD:
1399
    try:
1400
      InstanceShutdown(instance, shutdown_timeout)
1401
      return StartInstance(instance, False)
1402
    except errors.HypervisorError, err:
1403
      _Fail("Failed to hard reboot instance %s: %s", instance.name, err)
1404
  else:
1405
    _Fail("Invalid reboot_type received: %s", reboot_type)
1406

    
1407

    
1408
def InstanceBalloonMemory(instance, memory):
1409
  """Resize an instance's memory.
1410

1411
  @type instance: L{objects.Instance}
1412
  @param instance: the instance object
1413
  @type memory: int
1414
  @param memory: new memory amount in MB
1415
  @rtype: None
1416

1417
  """
1418
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1419
  running = hyper.ListInstances()
1420
  if instance.name not in running:
1421
    logging.info("Instance %s is not running, cannot balloon", instance.name)
1422
    return
1423
  try:
1424
    hyper.BalloonInstanceMemory(instance, memory)
1425
  except errors.HypervisorError, err:
1426
    _Fail("Failed to balloon instance memory: %s", err, exc=True)
1427

    
1428

    
1429
def MigrationInfo(instance):
1430
  """Gather information about an instance to be migrated.
1431

1432
  @type instance: L{objects.Instance}
1433
  @param instance: the instance definition
1434

1435
  """
1436
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1437
  try:
1438
    info = hyper.MigrationInfo(instance)
1439
  except errors.HypervisorError, err:
1440
    _Fail("Failed to fetch migration information: %s", err, exc=True)
1441
  return info
1442

    
1443

    
1444
def AcceptInstance(instance, info, target):
1445
  """Prepare the node to accept an instance.
1446

1447
  @type instance: L{objects.Instance}
1448
  @param instance: the instance definition
1449
  @type info: string/data (opaque)
1450
  @param info: migration information, from the source node
1451
  @type target: string
1452
  @param target: target host (usually ip), on this node
1453

1454
  """
1455
  # TODO: why is this required only for DTS_EXT_MIRROR?
1456
  if instance.disk_template in constants.DTS_EXT_MIRROR:
1457
    # Create the symlinks, as the disks are not active
1458
    # in any way
1459
    try:
1460
      _GatherAndLinkBlockDevs(instance)
1461
    except errors.BlockDeviceError, err:
1462
      _Fail("Block device error: %s", err, exc=True)
1463

    
1464
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1465
  try:
1466
    hyper.AcceptInstance(instance, info, target)
1467
  except errors.HypervisorError, err:
1468
    if instance.disk_template in constants.DTS_EXT_MIRROR:
1469
      _RemoveBlockDevLinks(instance.name, instance.disks)
1470
    _Fail("Failed to accept instance: %s", err, exc=True)
1471

    
1472

    
1473
def FinalizeMigrationDst(instance, info, success):
1474
  """Finalize any preparation to accept an instance.
1475

1476
  @type instance: L{objects.Instance}
1477
  @param instance: the instance definition
1478
  @type info: string/data (opaque)
1479
  @param info: migration information, from the source node
1480
  @type success: boolean
1481
  @param success: whether the migration was a success or a failure
1482

1483
  """
1484
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1485
  try:
1486
    hyper.FinalizeMigrationDst(instance, info, success)
1487
  except errors.HypervisorError, err:
1488
    _Fail("Failed to finalize migration on the target node: %s", err, exc=True)
1489

    
1490

    
1491
def MigrateInstance(instance, target, live):
1492
  """Migrates an instance to another node.
1493

1494
  @type instance: L{objects.Instance}
1495
  @param instance: the instance definition
1496
  @type target: string
1497
  @param target: the target node name
1498
  @type live: boolean
1499
  @param live: whether the migration should be done live or not (the
1500
      interpretation of this parameter is left to the hypervisor)
1501
  @raise RPCFail: if migration fails for some reason
1502

1503
  """
1504
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1505

    
1506
  try:
1507
    hyper.MigrateInstance(instance, target, live)
1508
  except errors.HypervisorError, err:
1509
    _Fail("Failed to migrate instance: %s", err, exc=True)
1510

    
1511

    
1512
def FinalizeMigrationSource(instance, success, live):
1513
  """Finalize the instance migration on the source node.
1514

1515
  @type instance: L{objects.Instance}
1516
  @param instance: the instance definition of the migrated instance
1517
  @type success: bool
1518
  @param success: whether the migration succeeded or not
1519
  @type live: bool
1520
  @param live: whether the user requested a live migration or not
1521
  @raise RPCFail: If the execution fails for some reason
1522

1523
  """
1524
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1525

    
1526
  try:
1527
    hyper.FinalizeMigrationSource(instance, success, live)
1528
  except Exception, err:  # pylint: disable=W0703
1529
    _Fail("Failed to finalize the migration on the source node: %s", err,
1530
          exc=True)
1531

    
1532

    
1533
def GetMigrationStatus(instance):
1534
  """Get the migration status
1535

1536
  @type instance: L{objects.Instance}
1537
  @param instance: the instance that is being migrated
1538
  @rtype: L{objects.MigrationStatus}
1539
  @return: the status of the current migration (one of
1540
           L{constants.HV_MIGRATION_VALID_STATUSES}), plus any additional
1541
           progress info that can be retrieved from the hypervisor
1542
  @raise RPCFail: If the migration status cannot be retrieved
1543

1544
  """
1545
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1546
  try:
1547
    return hyper.GetMigrationStatus(instance)
1548
  except Exception, err:  # pylint: disable=W0703
1549
    _Fail("Failed to get migration status: %s", err, exc=True)
1550

    
1551

    
1552
def BlockdevCreate(disk, size, owner, on_primary, info, excl_stor):
1553
  """Creates a block device for an instance.
1554

1555
  @type disk: L{objects.Disk}
1556
  @param disk: the object describing the disk we should create
1557
  @type size: int
1558
  @param size: the size of the physical underlying device, in MiB
1559
  @type owner: str
1560
  @param owner: the name of the instance for which disk is created,
1561
      used for device cache data
1562
  @type on_primary: boolean
1563
  @param on_primary:  indicates if it is the primary node or not
1564
  @type info: string
1565
  @param info: string that will be sent to the physical device
1566
      creation, used for example to set (LVM) tags on LVs
1567
  @type excl_stor: boolean
1568
  @param excl_stor: Whether exclusive_storage is active
1569

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

1574
  """
1575
  # TODO: remove the obsolete "size" argument
1576
  # pylint: disable=W0613
1577
  clist = []
1578
  if disk.children:
1579
    for child in disk.children:
1580
      try:
1581
        crdev = _RecursiveAssembleBD(child, owner, on_primary)
1582
      except errors.BlockDeviceError, err:
1583
        _Fail("Can't assemble device %s: %s", child, err)
1584
      if on_primary or disk.AssembleOnSecondary():
1585
        # we need the children open in case the device itself has to
1586
        # be assembled
1587
        try:
1588
          # pylint: disable=E1103
1589
          crdev.Open()
1590
        except errors.BlockDeviceError, err:
1591
          _Fail("Can't make child '%s' read-write: %s", child, err)
1592
      clist.append(crdev)
1593

    
1594
  try:
1595
    device = bdev.Create(disk, clist, excl_stor)
1596
  except errors.BlockDeviceError, err:
1597
    _Fail("Can't create block device: %s", err)
1598

    
1599
  if on_primary or disk.AssembleOnSecondary():
1600
    try:
1601
      device.Assemble()
1602
    except errors.BlockDeviceError, err:
1603
      _Fail("Can't assemble device after creation, unusual event: %s", err)
1604
    if on_primary or disk.OpenOnSecondary():
1605
      try:
1606
        device.Open(force=True)
1607
      except errors.BlockDeviceError, err:
1608
        _Fail("Can't make device r/w after creation, unusual event: %s", err)
1609
    DevCacheManager.UpdateCache(device.dev_path, owner,
1610
                                on_primary, disk.iv_name)
1611

    
1612
  device.SetInfo(info)
1613

    
1614
  return device.unique_id
1615

    
1616

    
1617
def _WipeDevice(path, offset, size):
1618
  """This function actually wipes the device.
1619

1620
  @param path: The path to the device to wipe
1621
  @param offset: The offset in MiB in the file
1622
  @param size: The size in MiB to write
1623

1624
  """
1625
  # Internal sizes are always in Mebibytes; if the following "dd" command
1626
  # should use a different block size the offset and size given to this
1627
  # function must be adjusted accordingly before being passed to "dd".
1628
  block_size = 1024 * 1024
1629

    
1630
  cmd = [constants.DD_CMD, "if=/dev/zero", "seek=%d" % offset,
1631
         "bs=%s" % block_size, "oflag=direct", "of=%s" % path,
1632
         "count=%d" % size]
1633
  result = utils.RunCmd(cmd)
1634

    
1635
  if result.failed:
1636
    _Fail("Wipe command '%s' exited with error: %s; output: %s", result.cmd,
1637
          result.fail_reason, result.output)
1638

    
1639

    
1640
def BlockdevWipe(disk, offset, size):
1641
  """Wipes a block device.
1642

1643
  @type disk: L{objects.Disk}
1644
  @param disk: the disk object we want to wipe
1645
  @type offset: int
1646
  @param offset: The offset in MiB in the file
1647
  @type size: int
1648
  @param size: The size in MiB to write
1649

1650
  """
1651
  try:
1652
    rdev = _RecursiveFindBD(disk)
1653
  except errors.BlockDeviceError:
1654
    rdev = None
1655

    
1656
  if not rdev:
1657
    _Fail("Cannot execute wipe for device %s: device not found", disk.iv_name)
1658

    
1659
  # Do cross verify some of the parameters
1660
  if offset < 0:
1661
    _Fail("Negative offset")
1662
  if size < 0:
1663
    _Fail("Negative size")
1664
  if offset > rdev.size:
1665
    _Fail("Offset is bigger than device size")
1666
  if (offset + size) > rdev.size:
1667
    _Fail("The provided offset and size to wipe is bigger than device size")
1668

    
1669
  _WipeDevice(rdev.dev_path, offset, size)
1670

    
1671

    
1672
def BlockdevPauseResumeSync(disks, pause):
1673
  """Pause or resume the sync of the block device.
1674

1675
  @type disks: list of L{objects.Disk}
1676
  @param disks: the disks object we want to pause/resume
1677
  @type pause: bool
1678
  @param pause: Wheater to pause or resume
1679

1680
  """
1681
  success = []
1682
  for disk in disks:
1683
    try:
1684
      rdev = _RecursiveFindBD(disk)
1685
    except errors.BlockDeviceError:
1686
      rdev = None
1687

    
1688
    if not rdev:
1689
      success.append((False, ("Cannot change sync for device %s:"
1690
                              " device not found" % disk.iv_name)))
1691
      continue
1692

    
1693
    result = rdev.PauseResumeSync(pause)
1694

    
1695
    if result:
1696
      success.append((result, None))
1697
    else:
1698
      if pause:
1699
        msg = "Pause"
1700
      else:
1701
        msg = "Resume"
1702
      success.append((result, "%s for device %s failed" % (msg, disk.iv_name)))
1703

    
1704
  return success
1705

    
1706

    
1707
def BlockdevRemove(disk):
1708
  """Remove a block device.
1709

1710
  @note: This is intended to be called recursively.
1711

1712
  @type disk: L{objects.Disk}
1713
  @param disk: the disk object we should remove
1714
  @rtype: boolean
1715
  @return: the success of the operation
1716

1717
  """
1718
  msgs = []
1719
  try:
1720
    rdev = _RecursiveFindBD(disk)
1721
  except errors.BlockDeviceError, err:
1722
    # probably can't attach
1723
    logging.info("Can't attach to device %s in remove", disk)
1724
    rdev = None
1725
  if rdev is not None:
1726
    r_path = rdev.dev_path
1727
    try:
1728
      rdev.Remove()
1729
    except errors.BlockDeviceError, err:
1730
      msgs.append(str(err))
1731
    if not msgs:
1732
      DevCacheManager.RemoveCache(r_path)
1733

    
1734
  if disk.children:
1735
    for child in disk.children:
1736
      try:
1737
        BlockdevRemove(child)
1738
      except RPCFail, err:
1739
        msgs.append(str(err))
1740

    
1741
  if msgs:
1742
    _Fail("; ".join(msgs))
1743

    
1744

    
1745
def _RecursiveAssembleBD(disk, owner, as_primary):
1746
  """Activate a block device for an instance.
1747

1748
  This is run on the primary and secondary nodes for an instance.
1749

1750
  @note: this function is called recursively.
1751

1752
  @type disk: L{objects.Disk}
1753
  @param disk: the disk we try to assemble
1754
  @type owner: str
1755
  @param owner: the name of the instance which owns the disk
1756
  @type as_primary: boolean
1757
  @param as_primary: if we should make the block device
1758
      read/write
1759

1760
  @return: the assembled device or None (in case no device
1761
      was assembled)
1762
  @raise errors.BlockDeviceError: in case there is an error
1763
      during the activation of the children or the device
1764
      itself
1765

1766
  """
1767
  children = []
1768
  if disk.children:
1769
    mcn = disk.ChildrenNeeded()
1770
    if mcn == -1:
1771
      mcn = 0 # max number of Nones allowed
1772
    else:
1773
      mcn = len(disk.children) - mcn # max number of Nones
1774
    for chld_disk in disk.children:
1775
      try:
1776
        cdev = _RecursiveAssembleBD(chld_disk, owner, as_primary)
1777
      except errors.BlockDeviceError, err:
1778
        if children.count(None) >= mcn:
1779
          raise
1780
        cdev = None
1781
        logging.error("Error in child activation (but continuing): %s",
1782
                      str(err))
1783
      children.append(cdev)
1784

    
1785
  if as_primary or disk.AssembleOnSecondary():
1786
    r_dev = bdev.Assemble(disk, children)
1787
    result = r_dev
1788
    if as_primary or disk.OpenOnSecondary():
1789
      r_dev.Open()
1790
    DevCacheManager.UpdateCache(r_dev.dev_path, owner,
1791
                                as_primary, disk.iv_name)
1792

    
1793
  else:
1794
    result = True
1795
  return result
1796

    
1797

    
1798
def BlockdevAssemble(disk, owner, as_primary, idx):
1799
  """Activate a block device for an instance.
1800

1801
  This is a wrapper over _RecursiveAssembleBD.
1802

1803
  @rtype: str or boolean
1804
  @return: a C{/dev/...} path for primary nodes, and
1805
      C{True} for secondary nodes
1806

1807
  """
1808
  try:
1809
    result = _RecursiveAssembleBD(disk, owner, as_primary)
1810
    if isinstance(result, bdev.BlockDev):
1811
      # pylint: disable=E1103
1812
      result = result.dev_path
1813
      if as_primary:
1814
        _SymlinkBlockDev(owner, result, idx)
1815
  except errors.BlockDeviceError, err:
1816
    _Fail("Error while assembling disk: %s", err, exc=True)
1817
  except OSError, err:
1818
    _Fail("Error while symlinking disk: %s", err, exc=True)
1819

    
1820
  return result
1821

    
1822

    
1823
def BlockdevShutdown(disk):
1824
  """Shut down a block device.
1825

1826
  First, if the device is assembled (Attach() is successful), then
1827
  the device is shutdown. Then the children of the device are
1828
  shutdown.
1829

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

1834
  @type disk: L{objects.Disk}
1835
  @param disk: the description of the disk we should
1836
      shutdown
1837
  @rtype: None
1838

1839
  """
1840
  msgs = []
1841
  r_dev = _RecursiveFindBD(disk)
1842
  if r_dev is not None:
1843
    r_path = r_dev.dev_path
1844
    try:
1845
      r_dev.Shutdown()
1846
      DevCacheManager.RemoveCache(r_path)
1847
    except errors.BlockDeviceError, err:
1848
      msgs.append(str(err))
1849

    
1850
  if disk.children:
1851
    for child in disk.children:
1852
      try:
1853
        BlockdevShutdown(child)
1854
      except RPCFail, err:
1855
        msgs.append(str(err))
1856

    
1857
  if msgs:
1858
    _Fail("; ".join(msgs))
1859

    
1860

    
1861
def BlockdevAddchildren(parent_cdev, new_cdevs):
1862
  """Extend a mirrored block device.
1863

1864
  @type parent_cdev: L{objects.Disk}
1865
  @param parent_cdev: the disk to which we should add children
1866
  @type new_cdevs: list of L{objects.Disk}
1867
  @param new_cdevs: the list of children which we should add
1868
  @rtype: None
1869

1870
  """
1871
  parent_bdev = _RecursiveFindBD(parent_cdev)
1872
  if parent_bdev is None:
1873
    _Fail("Can't find parent device '%s' in add children", parent_cdev)
1874
  new_bdevs = [_RecursiveFindBD(disk) for disk in new_cdevs]
1875
  if new_bdevs.count(None) > 0:
1876
    _Fail("Can't find new device(s) to add: %s:%s", new_bdevs, new_cdevs)
1877
  parent_bdev.AddChildren(new_bdevs)
1878

    
1879

    
1880
def BlockdevRemovechildren(parent_cdev, new_cdevs):
1881
  """Shrink a mirrored block device.
1882

1883
  @type parent_cdev: L{objects.Disk}
1884
  @param parent_cdev: the disk from which we should remove children
1885
  @type new_cdevs: list of L{objects.Disk}
1886
  @param new_cdevs: the list of children which we should remove
1887
  @rtype: None
1888

1889
  """
1890
  parent_bdev = _RecursiveFindBD(parent_cdev)
1891
  if parent_bdev is None:
1892
    _Fail("Can't find parent device '%s' in remove children", parent_cdev)
1893
  devs = []
1894
  for disk in new_cdevs:
1895
    rpath = disk.StaticDevPath()
1896
    if rpath is None:
1897
      bd = _RecursiveFindBD(disk)
1898
      if bd is None:
1899
        _Fail("Can't find device %s while removing children", disk)
1900
      else:
1901
        devs.append(bd.dev_path)
1902
    else:
1903
      if not utils.IsNormAbsPath(rpath):
1904
        _Fail("Strange path returned from StaticDevPath: '%s'", rpath)
1905
      devs.append(rpath)
1906
  parent_bdev.RemoveChildren(devs)
1907

    
1908

    
1909
def BlockdevGetmirrorstatus(disks):
1910
  """Get the mirroring status of a list of devices.
1911

1912
  @type disks: list of L{objects.Disk}
1913
  @param disks: the list of disks which we should query
1914
  @rtype: disk
1915
  @return: List of L{objects.BlockDevStatus}, one for each disk
1916
  @raise errors.BlockDeviceError: if any of the disks cannot be
1917
      found
1918

1919
  """
1920
  stats = []
1921
  for dsk in disks:
1922
    rbd = _RecursiveFindBD(dsk)
1923
    if rbd is None:
1924
      _Fail("Can't find device %s", dsk)
1925

    
1926
    stats.append(rbd.CombinedSyncStatus())
1927

    
1928
  return stats
1929

    
1930

    
1931
def BlockdevGetmirrorstatusMulti(disks):
1932
  """Get the mirroring status of a list of devices.
1933

1934
  @type disks: list of L{objects.Disk}
1935
  @param disks: the list of disks which we should query
1936
  @rtype: disk
1937
  @return: List of tuples, (bool, status), one for each disk; bool denotes
1938
    success/failure, status is L{objects.BlockDevStatus} on success, string
1939
    otherwise
1940

1941
  """
1942
  result = []
1943
  for disk in disks:
1944
    try:
1945
      rbd = _RecursiveFindBD(disk)
1946
      if rbd is None:
1947
        result.append((False, "Can't find device %s" % disk))
1948
        continue
1949

    
1950
      status = rbd.CombinedSyncStatus()
1951
    except errors.BlockDeviceError, err:
1952
      logging.exception("Error while getting disk status")
1953
      result.append((False, str(err)))
1954
    else:
1955
      result.append((True, status))
1956

    
1957
  assert len(disks) == len(result)
1958

    
1959
  return result
1960

    
1961

    
1962
def _RecursiveFindBD(disk):
1963
  """Check if a device is activated.
1964

1965
  If so, return information about the real device.
1966

1967
  @type disk: L{objects.Disk}
1968
  @param disk: the disk object we need to find
1969

1970
  @return: None if the device can't be found,
1971
      otherwise the device instance
1972

1973
  """
1974
  children = []
1975
  if disk.children:
1976
    for chdisk in disk.children:
1977
      children.append(_RecursiveFindBD(chdisk))
1978

    
1979
  return bdev.FindDevice(disk, children)
1980

    
1981

    
1982
def _OpenRealBD(disk):
1983
  """Opens the underlying block device of a disk.
1984

1985
  @type disk: L{objects.Disk}
1986
  @param disk: the disk object we want to open
1987

1988
  """
1989
  real_disk = _RecursiveFindBD(disk)
1990
  if real_disk is None:
1991
    _Fail("Block device '%s' is not set up", disk)
1992

    
1993
  real_disk.Open()
1994

    
1995
  return real_disk
1996

    
1997

    
1998
def BlockdevFind(disk):
1999
  """Check if a device is activated.
2000

2001
  If it is, return information about the real device.
2002

2003
  @type disk: L{objects.Disk}
2004
  @param disk: the disk to find
2005
  @rtype: None or objects.BlockDevStatus
2006
  @return: None if the disk cannot be found, otherwise a the current
2007
           information
2008

2009
  """
2010
  try:
2011
    rbd = _RecursiveFindBD(disk)
2012
  except errors.BlockDeviceError, err:
2013
    _Fail("Failed to find device: %s", err, exc=True)
2014

    
2015
  if rbd is None:
2016
    return None
2017

    
2018
  return rbd.GetSyncStatus()
2019

    
2020

    
2021
def BlockdevGetsize(disks):
2022
  """Computes the size of the given disks.
2023

2024
  If a disk is not found, returns None instead.
2025

2026
  @type disks: list of L{objects.Disk}
2027
  @param disks: the list of disk to compute the size for
2028
  @rtype: list
2029
  @return: list with elements None if the disk cannot be found,
2030
      otherwise the size
2031

2032
  """
2033
  result = []
2034
  for cf in disks:
2035
    try:
2036
      rbd = _RecursiveFindBD(cf)
2037
    except errors.BlockDeviceError:
2038
      result.append(None)
2039
      continue
2040
    if rbd is None:
2041
      result.append(None)
2042
    else:
2043
      result.append(rbd.GetActualSize())
2044
  return result
2045

    
2046

    
2047
def BlockdevExport(disk, dest_node, dest_path, cluster_name):
2048
  """Export a block device to a remote node.
2049

2050
  @type disk: L{objects.Disk}
2051
  @param disk: the description of the disk to export
2052
  @type dest_node: str
2053
  @param dest_node: the destination node to export to
2054
  @type dest_path: str
2055
  @param dest_path: the destination path on the target node
2056
  @type cluster_name: str
2057
  @param cluster_name: the cluster name, needed for SSH hostalias
2058
  @rtype: None
2059

2060
  """
2061
  real_disk = _OpenRealBD(disk)
2062

    
2063
  # the block size on the read dd is 1MiB to match our units
2064
  expcmd = utils.BuildShellCmd("set -e; set -o pipefail; "
2065
                               "dd if=%s bs=1048576 count=%s",
2066
                               real_disk.dev_path, str(disk.size))
2067

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

    
2077
  remotecmd = _GetSshRunner(cluster_name).BuildCmd(dest_node,
2078
                                                   constants.SSH_LOGIN_USER,
2079
                                                   destcmd)
2080

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

    
2084
  result = utils.RunCmd(["bash", "-c", command])
2085

    
2086
  if result.failed:
2087
    _Fail("Disk copy command '%s' returned error: %s"
2088
          " output: %s", command, result.fail_reason, result.output)
2089

    
2090

    
2091
def UploadFile(file_name, data, mode, uid, gid, atime, mtime):
2092
  """Write a file to the filesystem.
2093

2094
  This allows the master to overwrite(!) a file. It will only perform
2095
  the operation if the file belongs to a list of configuration files.
2096

2097
  @type file_name: str
2098
  @param file_name: the target file name
2099
  @type data: str
2100
  @param data: the new contents of the file
2101
  @type mode: int
2102
  @param mode: the mode to give the file (can be None)
2103
  @type uid: string
2104
  @param uid: the owner of the file
2105
  @type gid: string
2106
  @param gid: the group of the file
2107
  @type atime: float
2108
  @param atime: the atime to set on the file (can be None)
2109
  @type mtime: float
2110
  @param mtime: the mtime to set on the file (can be None)
2111
  @rtype: None
2112

2113
  """
2114
  file_name = vcluster.LocalizeVirtualPath(file_name)
2115

    
2116
  if not os.path.isabs(file_name):
2117
    _Fail("Filename passed to UploadFile is not absolute: '%s'", file_name)
2118

    
2119
  if file_name not in _ALLOWED_UPLOAD_FILES:
2120
    _Fail("Filename passed to UploadFile not in allowed upload targets: '%s'",
2121
          file_name)
2122

    
2123
  raw_data = _Decompress(data)
2124

    
2125
  if not (isinstance(uid, basestring) and isinstance(gid, basestring)):
2126
    _Fail("Invalid username/groupname type")
2127

    
2128
  getents = runtime.GetEnts()
2129
  uid = getents.LookupUser(uid)
2130
  gid = getents.LookupGroup(gid)
2131

    
2132
  utils.SafeWriteFile(file_name, None,
2133
                      data=raw_data, mode=mode, uid=uid, gid=gid,
2134
                      atime=atime, mtime=mtime)
2135

    
2136

    
2137
def RunOob(oob_program, command, node, timeout):
2138
  """Executes oob_program with given command on given node.
2139

2140
  @param oob_program: The path to the executable oob_program
2141
  @param command: The command to invoke on oob_program
2142
  @param node: The node given as an argument to the program
2143
  @param timeout: Timeout after which we kill the oob program
2144

2145
  @return: stdout
2146
  @raise RPCFail: If execution fails for some reason
2147

2148
  """
2149
  result = utils.RunCmd([oob_program, command, node], timeout=timeout)
2150

    
2151
  if result.failed:
2152
    _Fail("'%s' failed with reason '%s'; output: %s", result.cmd,
2153
          result.fail_reason, result.output)
2154

    
2155
  return result.stdout
2156

    
2157

    
2158
def _OSOndiskAPIVersion(os_dir):
2159
  """Compute and return the API version of a given OS.
2160

2161
  This function will try to read the API version of the OS residing in
2162
  the 'os_dir' directory.
2163

2164
  @type os_dir: str
2165
  @param os_dir: the directory in which we should look for the OS
2166
  @rtype: tuple
2167
  @return: tuple (status, data) with status denoting the validity and
2168
      data holding either the vaid versions or an error message
2169

2170
  """
2171
  api_file = utils.PathJoin(os_dir, constants.OS_API_FILE)
2172

    
2173
  try:
2174
    st = os.stat(api_file)
2175
  except EnvironmentError, err:
2176
    return False, ("Required file '%s' not found under path %s: %s" %
2177
                   (constants.OS_API_FILE, os_dir, utils.ErrnoOrStr(err)))
2178

    
2179
  if not stat.S_ISREG(stat.S_IFMT(st.st_mode)):
2180
    return False, ("File '%s' in %s is not a regular file" %
2181
                   (constants.OS_API_FILE, os_dir))
2182

    
2183
  try:
2184
    api_versions = utils.ReadFile(api_file).splitlines()
2185
  except EnvironmentError, err:
2186
    return False, ("Error while reading the API version file at %s: %s" %
2187
                   (api_file, utils.ErrnoOrStr(err)))
2188

    
2189
  try:
2190
    api_versions = [int(version.strip()) for version in api_versions]
2191
  except (TypeError, ValueError), err:
2192
    return False, ("API version(s) can't be converted to integer: %s" %
2193
                   str(err))
2194

    
2195
  return True, api_versions
2196

    
2197

    
2198
def DiagnoseOS(top_dirs=None):
2199
  """Compute the validity for all OSes.
2200

2201
  @type top_dirs: list
2202
  @param top_dirs: the list of directories in which to
2203
      search (if not given defaults to
2204
      L{pathutils.OS_SEARCH_PATH})
2205
  @rtype: list of L{objects.OS}
2206
  @return: a list of tuples (name, path, status, diagnose, variants,
2207
      parameters, api_version) for all (potential) OSes under all
2208
      search paths, where:
2209
          - name is the (potential) OS name
2210
          - path is the full path to the OS
2211
          - status True/False is the validity of the OS
2212
          - diagnose is the error message for an invalid OS, otherwise empty
2213
          - variants is a list of supported OS variants, if any
2214
          - parameters is a list of (name, help) parameters, if any
2215
          - api_version is a list of support OS API versions
2216

2217
  """
2218
  if top_dirs is None:
2219
    top_dirs = pathutils.OS_SEARCH_PATH
2220

    
2221
  result = []
2222
  for dir_name in top_dirs:
2223
    if os.path.isdir(dir_name):
2224
      try:
2225
        f_names = utils.ListVisibleFiles(dir_name)
2226
      except EnvironmentError, err:
2227
        logging.exception("Can't list the OS directory %s: %s", dir_name, err)
2228
        break
2229
      for name in f_names:
2230
        os_path = utils.PathJoin(dir_name, name)
2231
        status, os_inst = _TryOSFromDisk(name, base_dir=dir_name)
2232
        if status:
2233
          diagnose = ""
2234
          variants = os_inst.supported_variants
2235
          parameters = os_inst.supported_parameters
2236
          api_versions = os_inst.api_versions
2237
        else:
2238
          diagnose = os_inst
2239
          variants = parameters = api_versions = []
2240
        result.append((name, os_path, status, diagnose, variants,
2241
                       parameters, api_versions))
2242

    
2243
  return result
2244

    
2245

    
2246
def _TryOSFromDisk(name, base_dir=None):
2247
  """Create an OS instance from disk.
2248

2249
  This function will return an OS instance if the given name is a
2250
  valid OS name.
2251

2252
  @type base_dir: string
2253
  @keyword base_dir: Base directory containing OS installations.
2254
                     Defaults to a search in all the OS_SEARCH_PATH dirs.
2255
  @rtype: tuple
2256
  @return: success and either the OS instance if we find a valid one,
2257
      or error message
2258

2259
  """
2260
  if base_dir is None:
2261
    os_dir = utils.FindFile(name, pathutils.OS_SEARCH_PATH, os.path.isdir)
2262
  else:
2263
    os_dir = utils.FindFile(name, [base_dir], os.path.isdir)
2264

    
2265
  if os_dir is None:
2266
    return False, "Directory for OS %s not found in search path" % name
2267

    
2268
  status, api_versions = _OSOndiskAPIVersion(os_dir)
2269
  if not status:
2270
    # push the error up
2271
    return status, api_versions
2272

    
2273
  if not constants.OS_API_VERSIONS.intersection(api_versions):
2274
    return False, ("API version mismatch for path '%s': found %s, want %s." %
2275
                   (os_dir, api_versions, constants.OS_API_VERSIONS))
2276

    
2277
  # OS Files dictionary, we will populate it with the absolute path
2278
  # names; if the value is True, then it is a required file, otherwise
2279
  # an optional one
2280
  os_files = dict.fromkeys(constants.OS_SCRIPTS, True)
2281

    
2282
  if max(api_versions) >= constants.OS_API_V15:
2283
    os_files[constants.OS_VARIANTS_FILE] = False
2284

    
2285
  if max(api_versions) >= constants.OS_API_V20:
2286
    os_files[constants.OS_PARAMETERS_FILE] = True
2287
  else:
2288
    del os_files[constants.OS_SCRIPT_VERIFY]
2289

    
2290
  for (filename, required) in os_files.items():
2291
    os_files[filename] = utils.PathJoin(os_dir, filename)
2292

    
2293
    try:
2294
      st = os.stat(os_files[filename])
2295
    except EnvironmentError, err:
2296
      if err.errno == errno.ENOENT and not required:
2297
        del os_files[filename]
2298
        continue
2299
      return False, ("File '%s' under path '%s' is missing (%s)" %
2300
                     (filename, os_dir, utils.ErrnoOrStr(err)))
2301

    
2302
    if not stat.S_ISREG(stat.S_IFMT(st.st_mode)):
2303
      return False, ("File '%s' under path '%s' is not a regular file" %
2304
                     (filename, os_dir))
2305

    
2306
    if filename in constants.OS_SCRIPTS:
2307
      if stat.S_IMODE(st.st_mode) & stat.S_IXUSR != stat.S_IXUSR:
2308
        return False, ("File '%s' under path '%s' is not executable" %
2309
                       (filename, os_dir))
2310

    
2311
  variants = []
2312
  if constants.OS_VARIANTS_FILE in os_files:
2313
    variants_file = os_files[constants.OS_VARIANTS_FILE]
2314
    try:
2315
      variants = \
2316
        utils.FilterEmptyLinesAndComments(utils.ReadFile(variants_file))
2317
    except EnvironmentError, err:
2318
      # we accept missing files, but not other errors
2319
      if err.errno != errno.ENOENT:
2320
        return False, ("Error while reading the OS variants file at %s: %s" %
2321
                       (variants_file, utils.ErrnoOrStr(err)))
2322

    
2323
  parameters = []
2324
  if constants.OS_PARAMETERS_FILE in os_files:
2325
    parameters_file = os_files[constants.OS_PARAMETERS_FILE]
2326
    try:
2327
      parameters = utils.ReadFile(parameters_file).splitlines()
2328
    except EnvironmentError, err:
2329
      return False, ("Error while reading the OS parameters file at %s: %s" %
2330
                     (parameters_file, utils.ErrnoOrStr(err)))
2331
    parameters = [v.split(None, 1) for v in parameters]
2332

    
2333
  os_obj = objects.OS(name=name, path=os_dir,
2334
                      create_script=os_files[constants.OS_SCRIPT_CREATE],
2335
                      export_script=os_files[constants.OS_SCRIPT_EXPORT],
2336
                      import_script=os_files[constants.OS_SCRIPT_IMPORT],
2337
                      rename_script=os_files[constants.OS_SCRIPT_RENAME],
2338
                      verify_script=os_files.get(constants.OS_SCRIPT_VERIFY,
2339
                                                 None),
2340
                      supported_variants=variants,
2341
                      supported_parameters=parameters,
2342
                      api_versions=api_versions)
2343
  return True, os_obj
2344

    
2345

    
2346
def OSFromDisk(name, base_dir=None):
2347
  """Create an OS instance from disk.
2348

2349
  This function will return an OS instance if the given name is a
2350
  valid OS name. Otherwise, it will raise an appropriate
2351
  L{RPCFail} exception, detailing why this is not a valid OS.
2352

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

2356
  @type base_dir: string
2357
  @keyword base_dir: Base directory containing OS installations.
2358
                     Defaults to a search in all the OS_SEARCH_PATH dirs.
2359
  @rtype: L{objects.OS}
2360
  @return: the OS instance if we find a valid one
2361
  @raise RPCFail: if we don't find a valid OS
2362

2363
  """
2364
  name_only = objects.OS.GetName(name)
2365
  status, payload = _TryOSFromDisk(name_only, base_dir)
2366

    
2367
  if not status:
2368
    _Fail(payload)
2369

    
2370
  return payload
2371

    
2372

    
2373
def OSCoreEnv(os_name, inst_os, os_params, debug=0):
2374
  """Calculate the basic environment for an os script.
2375

2376
  @type os_name: str
2377
  @param os_name: full operating system name (including variant)
2378
  @type inst_os: L{objects.OS}
2379
  @param inst_os: operating system for which the environment is being built
2380
  @type os_params: dict
2381
  @param os_params: the OS parameters
2382
  @type debug: integer
2383
  @param debug: debug level (0 or 1, for OS Api 10)
2384
  @rtype: dict
2385
  @return: dict of environment variables
2386
  @raise errors.BlockDeviceError: if the block device
2387
      cannot be found
2388

2389
  """
2390
  result = {}
2391
  api_version = \
2392
    max(constants.OS_API_VERSIONS.intersection(inst_os.api_versions))
2393
  result["OS_API_VERSION"] = "%d" % api_version
2394
  result["OS_NAME"] = inst_os.name
2395
  result["DEBUG_LEVEL"] = "%d" % debug
2396

    
2397
  # OS variants
2398
  if api_version >= constants.OS_API_V15 and inst_os.supported_variants:
2399
    variant = objects.OS.GetVariant(os_name)
2400
    if not variant:
2401
      variant = inst_os.supported_variants[0]
2402
  else:
2403
    variant = ""
2404
  result["OS_VARIANT"] = variant
2405

    
2406
  # OS params
2407
  for pname, pvalue in os_params.items():
2408
    result["OSP_%s" % pname.upper()] = pvalue
2409

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

    
2415
  return result
2416

    
2417

    
2418
def OSEnvironment(instance, inst_os, debug=0):
2419
  """Calculate the environment for an os script.
2420

2421
  @type instance: L{objects.Instance}
2422
  @param instance: target instance for the os script run
2423
  @type inst_os: L{objects.OS}
2424
  @param inst_os: operating system for which the environment is being built
2425
  @type debug: integer
2426
  @param debug: debug level (0 or 1, for OS Api 10)
2427
  @rtype: dict
2428
  @return: dict of environment variables
2429
  @raise errors.BlockDeviceError: if the block device
2430
      cannot be found
2431

2432
  """
2433
  result = OSCoreEnv(instance.os, inst_os, instance.osparams, debug=debug)
2434

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

    
2438
  result["HYPERVISOR"] = instance.hypervisor
2439
  result["DISK_COUNT"] = "%d" % len(instance.disks)
2440
  result["NIC_COUNT"] = "%d" % len(instance.nics)
2441
  result["INSTANCE_SECONDARY_NODES"] = \
2442
      ("%s" % " ".join(instance.secondary_nodes))
2443

    
2444
  # Disks
2445
  for idx, disk in enumerate(instance.disks):
2446
    real_disk = _OpenRealBD(disk)
2447
    result["DISK_%d_PATH" % idx] = real_disk.dev_path
2448
    result["DISK_%d_ACCESS" % idx] = disk.mode
2449
    if constants.HV_DISK_TYPE in instance.hvparams:
2450
      result["DISK_%d_FRONTEND_TYPE" % idx] = \
2451
        instance.hvparams[constants.HV_DISK_TYPE]
2452
    if disk.dev_type in constants.LDS_BLOCK:
2453
      result["DISK_%d_BACKEND_TYPE" % idx] = "block"
2454
    elif disk.dev_type == constants.LD_FILE:
2455
      result["DISK_%d_BACKEND_TYPE" % idx] = \
2456
        "file:%s" % disk.physical_id[0]
2457

    
2458
  # NICs
2459
  for idx, nic in enumerate(instance.nics):
2460
    result["NIC_%d_MAC" % idx] = nic.mac
2461
    if nic.ip:
2462
      result["NIC_%d_IP" % idx] = nic.ip
2463
    result["NIC_%d_MODE" % idx] = nic.nicparams[constants.NIC_MODE]
2464
    if nic.nicparams[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED:
2465
      result["NIC_%d_BRIDGE" % idx] = nic.nicparams[constants.NIC_LINK]
2466
    if nic.nicparams[constants.NIC_LINK]:
2467
      result["NIC_%d_LINK" % idx] = nic.nicparams[constants.NIC_LINK]
2468
    if nic.network:
2469
      result["NIC_%d_NETWORK" % idx] = nic.network
2470
    if constants.HV_NIC_TYPE in instance.hvparams:
2471
      result["NIC_%d_FRONTEND_TYPE" % idx] = \
2472
        instance.hvparams[constants.HV_NIC_TYPE]
2473

    
2474
  # HV/BE params
2475
  for source, kind in [(instance.beparams, "BE"), (instance.hvparams, "HV")]:
2476
    for key, value in source.items():
2477
      result["INSTANCE_%s_%s" % (kind, key)] = str(value)
2478

    
2479
  return result
2480

    
2481

    
2482
def BlockdevGrow(disk, amount, dryrun, backingstore):
2483
  """Grow a stack of block devices.
2484

2485
  This function is called recursively, with the childrens being the
2486
  first ones to resize.
2487

2488
  @type disk: L{objects.Disk}
2489
  @param disk: the disk to be grown
2490
  @type amount: integer
2491
  @param amount: the amount (in mebibytes) to grow with
2492
  @type dryrun: boolean
2493
  @param dryrun: whether to execute the operation in simulation mode
2494
      only, without actually increasing the size
2495
  @param backingstore: whether to execute the operation on backing storage
2496
      only, or on "logical" storage only; e.g. DRBD is logical storage,
2497
      whereas LVM, file, RBD are backing storage
2498
  @rtype: (status, result)
2499
  @return: a tuple with the status of the operation (True/False), and
2500
      the errors message if status is False
2501

2502
  """
2503
  r_dev = _RecursiveFindBD(disk)
2504
  if r_dev is None:
2505
    _Fail("Cannot find block device %s", disk)
2506

    
2507
  try:
2508
    r_dev.Grow(amount, dryrun, backingstore)
2509
  except errors.BlockDeviceError, err:
2510
    _Fail("Failed to grow block device: %s", err, exc=True)
2511

    
2512

    
2513
def BlockdevSnapshot(disk):
2514
  """Create a snapshot copy of a block device.
2515

2516
  This function is called recursively, and the snapshot is actually created
2517
  just for the leaf lvm backend device.
2518

2519
  @type disk: L{objects.Disk}
2520
  @param disk: the disk to be snapshotted
2521
  @rtype: string
2522
  @return: snapshot disk ID as (vg, lv)
2523

2524
  """
2525
  if disk.dev_type == constants.LD_DRBD8:
2526
    if not disk.children:
2527
      _Fail("DRBD device '%s' without backing storage cannot be snapshotted",
2528
            disk.unique_id)
2529
    return BlockdevSnapshot(disk.children[0])
2530
  elif disk.dev_type == constants.LD_LV:
2531
    r_dev = _RecursiveFindBD(disk)
2532
    if r_dev is not None:
2533
      # FIXME: choose a saner value for the snapshot size
2534
      # let's stay on the safe side and ask for the full size, for now
2535
      return r_dev.Snapshot(disk.size)
2536
    else:
2537
      _Fail("Cannot find block device %s", disk)
2538
  else:
2539
    _Fail("Cannot snapshot non-lvm block device '%s' of type '%s'",
2540
          disk.unique_id, disk.dev_type)
2541

    
2542

    
2543
def BlockdevSetInfo(disk, info):
2544
  """Sets 'metadata' information on block devices.
2545

2546
  This function sets 'info' metadata on block devices. Initial
2547
  information is set at device creation; this function should be used
2548
  for example after renames.
2549

2550
  @type disk: L{objects.Disk}
2551
  @param disk: the disk to be grown
2552
  @type info: string
2553
  @param info: new 'info' metadata
2554
  @rtype: (status, result)
2555
  @return: a tuple with the status of the operation (True/False), and
2556
      the errors message if status is False
2557

2558
  """
2559
  r_dev = _RecursiveFindBD(disk)
2560
  if r_dev is None:
2561
    _Fail("Cannot find block device %s", disk)
2562

    
2563
  try:
2564
    r_dev.SetInfo(info)
2565
  except errors.BlockDeviceError, err:
2566
    _Fail("Failed to set information on block device: %s", err, exc=True)
2567

    
2568

    
2569
def FinalizeExport(instance, snap_disks):
2570
  """Write out the export configuration information.
2571

2572
  @type instance: L{objects.Instance}
2573
  @param instance: the instance which we export, used for
2574
      saving configuration
2575
  @type snap_disks: list of L{objects.Disk}
2576
  @param snap_disks: list of snapshot block devices, which
2577
      will be used to get the actual name of the dump file
2578

2579
  @rtype: None
2580

2581
  """
2582
  destdir = utils.PathJoin(pathutils.EXPORT_DIR, instance.name + ".new")
2583
  finaldestdir = utils.PathJoin(pathutils.EXPORT_DIR, instance.name)
2584

    
2585
  config = objects.SerializableConfigParser()
2586

    
2587
  config.add_section(constants.INISECT_EXP)
2588
  config.set(constants.INISECT_EXP, "version", "0")
2589
  config.set(constants.INISECT_EXP, "timestamp", "%d" % int(time.time()))
2590
  config.set(constants.INISECT_EXP, "source", instance.primary_node)
2591
  config.set(constants.INISECT_EXP, "os", instance.os)
2592
  config.set(constants.INISECT_EXP, "compression", "none")
2593

    
2594
  config.add_section(constants.INISECT_INS)
2595
  config.set(constants.INISECT_INS, "name", instance.name)
2596
  config.set(constants.INISECT_INS, "maxmem", "%d" %
2597
             instance.beparams[constants.BE_MAXMEM])
2598
  config.set(constants.INISECT_INS, "minmem", "%d" %
2599
             instance.beparams[constants.BE_MINMEM])
2600
  # "memory" is deprecated, but useful for exporting to old ganeti versions
2601
  config.set(constants.INISECT_INS, "memory", "%d" %
2602
             instance.beparams[constants.BE_MAXMEM])
2603
  config.set(constants.INISECT_INS, "vcpus", "%d" %
2604
             instance.beparams[constants.BE_VCPUS])
2605
  config.set(constants.INISECT_INS, "disk_template", instance.disk_template)
2606
  config.set(constants.INISECT_INS, "hypervisor", instance.hypervisor)
2607
  config.set(constants.INISECT_INS, "tags", " ".join(instance.GetTags()))
2608

    
2609
  nic_total = 0
2610
  for nic_count, nic in enumerate(instance.nics):
2611
    nic_total += 1
2612
    config.set(constants.INISECT_INS, "nic%d_mac" %
2613
               nic_count, "%s" % nic.mac)
2614
    config.set(constants.INISECT_INS, "nic%d_ip" % nic_count, "%s" % nic.ip)
2615
    config.set(constants.INISECT_INS, "nic%d_network" % nic_count,
2616
               "%s" % nic.network)
2617
    for param in constants.NICS_PARAMETER_TYPES:
2618
      config.set(constants.INISECT_INS, "nic%d_%s" % (nic_count, param),
2619
                 "%s" % nic.nicparams.get(param, None))
2620
  # TODO: redundant: on load can read nics until it doesn't exist
2621
  config.set(constants.INISECT_INS, "nic_count", "%d" % nic_total)
2622

    
2623
  disk_total = 0
2624
  for disk_count, disk in enumerate(snap_disks):
2625
    if disk:
2626
      disk_total += 1
2627
      config.set(constants.INISECT_INS, "disk%d_ivname" % disk_count,
2628
                 ("%s" % disk.iv_name))
2629
      config.set(constants.INISECT_INS, "disk%d_dump" % disk_count,
2630
                 ("%s" % disk.physical_id[1]))
2631
      config.set(constants.INISECT_INS, "disk%d_size" % disk_count,
2632
                 ("%d" % disk.size))
2633

    
2634
  config.set(constants.INISECT_INS, "disk_count", "%d" % disk_total)
2635

    
2636
  # New-style hypervisor/backend parameters
2637

    
2638
  config.add_section(constants.INISECT_HYP)
2639
  for name, value in instance.hvparams.items():
2640
    if name not in constants.HVC_GLOBALS:
2641
      config.set(constants.INISECT_HYP, name, str(value))
2642

    
2643
  config.add_section(constants.INISECT_BEP)
2644
  for name, value in instance.beparams.items():
2645
    config.set(constants.INISECT_BEP, name, str(value))
2646

    
2647
  config.add_section(constants.INISECT_OSP)
2648
  for name, value in instance.osparams.items():
2649
    config.set(constants.INISECT_OSP, name, str(value))
2650

    
2651
  utils.WriteFile(utils.PathJoin(destdir, constants.EXPORT_CONF_FILE),
2652
                  data=config.Dumps())
2653
  shutil.rmtree(finaldestdir, ignore_errors=True)
2654
  shutil.move(destdir, finaldestdir)
2655

    
2656

    
2657
def ExportInfo(dest):
2658
  """Get export configuration information.
2659

2660
  @type dest: str
2661
  @param dest: directory containing the export
2662

2663
  @rtype: L{objects.SerializableConfigParser}
2664
  @return: a serializable config file containing the
2665
      export info
2666

2667
  """
2668
  cff = utils.PathJoin(dest, constants.EXPORT_CONF_FILE)
2669

    
2670
  config = objects.SerializableConfigParser()
2671
  config.read(cff)
2672

    
2673
  if (not config.has_section(constants.INISECT_EXP) or
2674
      not config.has_section(constants.INISECT_INS)):
2675
    _Fail("Export info file doesn't have the required fields")
2676

    
2677
  return config.Dumps()
2678

    
2679

    
2680
def ListExports():
2681
  """Return a list of exports currently available on this machine.
2682

2683
  @rtype: list
2684
  @return: list of the exports
2685

2686
  """
2687
  if os.path.isdir(pathutils.EXPORT_DIR):
2688
    return sorted(utils.ListVisibleFiles(pathutils.EXPORT_DIR))
2689
  else:
2690
    _Fail("No exports directory")
2691

    
2692

    
2693
def RemoveExport(export):
2694
  """Remove an existing export from the node.
2695

2696
  @type export: str
2697
  @param export: the name of the export to remove
2698
  @rtype: None
2699

2700
  """
2701
  target = utils.PathJoin(pathutils.EXPORT_DIR, export)
2702

    
2703
  try:
2704
    shutil.rmtree(target)
2705
  except EnvironmentError, err:
2706
    _Fail("Error while removing the export: %s", err, exc=True)
2707

    
2708

    
2709
def BlockdevRename(devlist):
2710
  """Rename a list of block devices.
2711

2712
  @type devlist: list of tuples
2713
  @param devlist: list of tuples of the form  (disk,
2714
      new_logical_id, new_physical_id); disk is an
2715
      L{objects.Disk} object describing the current disk,
2716
      and new logical_id/physical_id is the name we
2717
      rename it to
2718
  @rtype: boolean
2719
  @return: True if all renames succeeded, False otherwise
2720

2721
  """
2722
  msgs = []
2723
  result = True
2724
  for disk, unique_id in devlist:
2725
    dev = _RecursiveFindBD(disk)
2726
    if dev is None:
2727
      msgs.append("Can't find device %s in rename" % str(disk))
2728
      result = False
2729
      continue
2730
    try:
2731
      old_rpath = dev.dev_path
2732
      dev.Rename(unique_id)
2733
      new_rpath = dev.dev_path
2734
      if old_rpath != new_rpath:
2735
        DevCacheManager.RemoveCache(old_rpath)
2736
        # FIXME: we should add the new cache information here, like:
2737
        # DevCacheManager.UpdateCache(new_rpath, owner, ...)
2738
        # but we don't have the owner here - maybe parse from existing
2739
        # cache? for now, we only lose lvm data when we rename, which
2740
        # is less critical than DRBD or MD
2741
    except errors.BlockDeviceError, err:
2742
      msgs.append("Can't rename device '%s' to '%s': %s" %
2743
                  (dev, unique_id, err))
2744
      logging.exception("Can't rename device '%s' to '%s'", dev, unique_id)
2745
      result = False
2746
  if not result:
2747
    _Fail("; ".join(msgs))
2748

    
2749

    
2750
def _TransformFileStorageDir(fs_dir):
2751
  """Checks whether given file_storage_dir is valid.
2752

2753
  Checks wheter the given fs_dir is within the cluster-wide default
2754
  file_storage_dir or the shared_file_storage_dir, which are stored in
2755
  SimpleStore. Only paths under those directories are allowed.
2756

2757
  @type fs_dir: str
2758
  @param fs_dir: the path to check
2759

2760
  @return: the normalized path if valid, None otherwise
2761

2762
  """
2763
  if not (constants.ENABLE_FILE_STORAGE or
2764
          constants.ENABLE_SHARED_FILE_STORAGE):
2765
    _Fail("File storage disabled at configure time")
2766

    
2767
  bdev.CheckFileStoragePath(fs_dir)
2768

    
2769
  return os.path.normpath(fs_dir)
2770

    
2771

    
2772
def CreateFileStorageDir(file_storage_dir):
2773
  """Create file storage directory.
2774

2775
  @type file_storage_dir: str
2776
  @param file_storage_dir: directory to create
2777

2778
  @rtype: tuple
2779
  @return: tuple with first element a boolean indicating wheter dir
2780
      creation was successful or not
2781

2782
  """
2783
  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
2784
  if os.path.exists(file_storage_dir):
2785
    if not os.path.isdir(file_storage_dir):
2786
      _Fail("Specified storage dir '%s' is not a directory",
2787
            file_storage_dir)
2788
  else:
2789
    try:
2790
      os.makedirs(file_storage_dir, 0750)
2791
    except OSError, err:
2792
      _Fail("Cannot create file storage directory '%s': %s",
2793
            file_storage_dir, err, exc=True)
2794

    
2795

    
2796
def RemoveFileStorageDir(file_storage_dir):
2797
  """Remove file storage directory.
2798

2799
  Remove it only if it's empty. If not log an error and return.
2800

2801
  @type file_storage_dir: str
2802
  @param file_storage_dir: the directory we should cleanup
2803
  @rtype: tuple (success,)
2804
  @return: tuple of one element, C{success}, denoting
2805
      whether the operation was successful
2806

2807
  """
2808
  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
2809
  if os.path.exists(file_storage_dir):
2810
    if not os.path.isdir(file_storage_dir):
2811
      _Fail("Specified Storage directory '%s' is not a directory",
2812
            file_storage_dir)
2813
    # deletes dir only if empty, otherwise we want to fail the rpc call
2814
    try:
2815
      os.rmdir(file_storage_dir)
2816
    except OSError, err:
2817
      _Fail("Cannot remove file storage directory '%s': %s",
2818
            file_storage_dir, err)
2819

    
2820

    
2821
def RenameFileStorageDir(old_file_storage_dir, new_file_storage_dir):
2822
  """Rename the file storage directory.
2823

2824
  @type old_file_storage_dir: str
2825
  @param old_file_storage_dir: the current path
2826
  @type new_file_storage_dir: str
2827
  @param new_file_storage_dir: the name we should rename to
2828
  @rtype: tuple (success,)
2829
  @return: tuple of one element, C{success}, denoting
2830
      whether the operation was successful
2831

2832
  """
2833
  old_file_storage_dir = _TransformFileStorageDir(old_file_storage_dir)
2834
  new_file_storage_dir = _TransformFileStorageDir(new_file_storage_dir)
2835
  if not os.path.exists(new_file_storage_dir):
2836
    if os.path.isdir(old_file_storage_dir):
2837
      try:
2838
        os.rename(old_file_storage_dir, new_file_storage_dir)
2839
      except OSError, err:
2840
        _Fail("Cannot rename '%s' to '%s': %s",
2841
              old_file_storage_dir, new_file_storage_dir, err)
2842
    else:
2843
      _Fail("Specified storage dir '%s' is not a directory",
2844
            old_file_storage_dir)
2845
  else:
2846
    if os.path.exists(old_file_storage_dir):
2847
      _Fail("Cannot rename '%s' to '%s': both locations exist",
2848
            old_file_storage_dir, new_file_storage_dir)
2849

    
2850

    
2851
def _EnsureJobQueueFile(file_name):
2852
  """Checks whether the given filename is in the queue directory.
2853

2854
  @type file_name: str
2855
  @param file_name: the file name we should check
2856
  @rtype: None
2857
  @raises RPCFail: if the file is not valid
2858

2859
  """
2860
  if not utils.IsBelowDir(pathutils.QUEUE_DIR, file_name):
2861
    _Fail("Passed job queue file '%s' does not belong to"
2862
          " the queue directory '%s'", file_name, pathutils.QUEUE_DIR)
2863

    
2864

    
2865
def JobQueueUpdate(file_name, content):
2866
  """Updates a file in the queue directory.
2867

2868
  This is just a wrapper over L{utils.io.WriteFile}, with proper
2869
  checking.
2870

2871
  @type file_name: str
2872
  @param file_name: the job file name
2873
  @type content: str
2874
  @param content: the new job contents
2875
  @rtype: boolean
2876
  @return: the success of the operation
2877

2878
  """
2879
  file_name = vcluster.LocalizeVirtualPath(file_name)
2880

    
2881
  _EnsureJobQueueFile(file_name)
2882
  getents = runtime.GetEnts()
2883

    
2884
  # Write and replace the file atomically
2885
  utils.WriteFile(file_name, data=_Decompress(content), uid=getents.masterd_uid,
2886
                  gid=getents.masterd_gid)
2887

    
2888

    
2889
def JobQueueRename(old, new):
2890
  """Renames a job queue file.
2891

2892
  This is just a wrapper over os.rename with proper checking.
2893

2894
  @type old: str
2895
  @param old: the old (actual) file name
2896
  @type new: str
2897
  @param new: the desired file name
2898
  @rtype: tuple
2899
  @return: the success of the operation and payload
2900

2901
  """
2902
  old = vcluster.LocalizeVirtualPath(old)
2903
  new = vcluster.LocalizeVirtualPath(new)
2904

    
2905
  _EnsureJobQueueFile(old)
2906
  _EnsureJobQueueFile(new)
2907

    
2908
  getents = runtime.GetEnts()
2909

    
2910
  utils.RenameFile(old, new, mkdir=True, mkdir_mode=0700,
2911
                   dir_uid=getents.masterd_uid, dir_gid=getents.masterd_gid)
2912

    
2913

    
2914
def BlockdevClose(instance_name, disks):
2915
  """Closes the given block devices.
2916

2917
  This means they will be switched to secondary mode (in case of
2918
  DRBD).
2919

2920
  @param instance_name: if the argument is not empty, the symlinks
2921
      of this instance will be removed
2922
  @type disks: list of L{objects.Disk}
2923
  @param disks: the list of disks to be closed
2924
  @rtype: tuple (success, message)
2925
  @return: a tuple of success and message, where success
2926
      indicates the succes of the operation, and message
2927
      which will contain the error details in case we
2928
      failed
2929

2930
  """
2931
  bdevs = []
2932
  for cf in disks:
2933
    rd = _RecursiveFindBD(cf)
2934
    if rd is None:
2935
      _Fail("Can't find device %s", cf)
2936
    bdevs.append(rd)
2937

    
2938
  msg = []
2939
  for rd in bdevs:
2940
    try:
2941
      rd.Close()
2942
    except errors.BlockDeviceError, err:
2943
      msg.append(str(err))
2944
  if msg:
2945
    _Fail("Can't make devices secondary: %s", ",".join(msg))
2946
  else:
2947
    if instance_name:
2948
      _RemoveBlockDevLinks(instance_name, disks)
2949

    
2950

    
2951
def ValidateHVParams(hvname, hvparams):
2952
  """Validates the given hypervisor parameters.
2953

2954
  @type hvname: string
2955
  @param hvname: the hypervisor name
2956
  @type hvparams: dict
2957
  @param hvparams: the hypervisor parameters to be validated
2958
  @rtype: None
2959

2960
  """
2961
  try:
2962
    hv_type = hypervisor.GetHypervisor(hvname)
2963
    hv_type.ValidateParameters(hvparams)
2964
  except errors.HypervisorError, err:
2965
    _Fail(str(err), log=False)
2966

    
2967

    
2968
def _CheckOSPList(os_obj, parameters):
2969
  """Check whether a list of parameters is supported by the OS.
2970

2971
  @type os_obj: L{objects.OS}
2972
  @param os_obj: OS object to check
2973
  @type parameters: list
2974
  @param parameters: the list of parameters to check
2975

2976
  """
2977
  supported = [v[0] for v in os_obj.supported_parameters]
2978
  delta = frozenset(parameters).difference(supported)
2979
  if delta:
2980
    _Fail("The following parameters are not supported"
2981
          " by the OS %s: %s" % (os_obj.name, utils.CommaJoin(delta)))
2982

    
2983

    
2984
def ValidateOS(required, osname, checks, osparams):
2985
  """Validate the given OS' parameters.
2986

2987
  @type required: boolean
2988
  @param required: whether absence of the OS should translate into
2989
      failure or not
2990
  @type osname: string
2991
  @param osname: the OS to be validated
2992
  @type checks: list
2993
  @param checks: list of the checks to run (currently only 'parameters')
2994
  @type osparams: dict
2995
  @param osparams: dictionary with OS parameters
2996
  @rtype: boolean
2997
  @return: True if the validation passed, or False if the OS was not
2998
      found and L{required} was false
2999

3000
  """
3001
  if not constants.OS_VALIDATE_CALLS.issuperset(checks):
3002
    _Fail("Unknown checks required for OS %s: %s", osname,
3003
          set(checks).difference(constants.OS_VALIDATE_CALLS))
3004

    
3005
  name_only = objects.OS.GetName(osname)
3006
  status, tbv = _TryOSFromDisk(name_only, None)
3007

    
3008
  if not status:
3009
    if required:
3010
      _Fail(tbv)
3011
    else:
3012
      return False
3013

    
3014
  if max(tbv.api_versions) < constants.OS_API_V20:
3015
    return True
3016

    
3017
  if constants.OS_VALIDATE_PARAMETERS in checks:
3018
    _CheckOSPList(tbv, osparams.keys())
3019

    
3020
  validate_env = OSCoreEnv(osname, tbv, osparams)
3021
  result = utils.RunCmd([tbv.verify_script] + checks, env=validate_env,
3022
                        cwd=tbv.path, reset_env=True)
3023
  if result.failed:
3024
    logging.error("os validate command '%s' returned error: %s output: %s",
3025
                  result.cmd, result.fail_reason, result.output)
3026
    _Fail("OS validation script failed (%s), output: %s",
3027
          result.fail_reason, result.output, log=False)
3028

    
3029
  return True
3030

    
3031

    
3032
def DemoteFromMC():
3033
  """Demotes the current node from master candidate role.
3034

3035
  """
3036
  # try to ensure we're not the master by mistake
3037
  master, myself = ssconf.GetMasterAndMyself()
3038
  if master == myself:
3039
    _Fail("ssconf status shows I'm the master node, will not demote")
3040

    
3041
  result = utils.RunCmd([pathutils.DAEMON_UTIL, "check", constants.MASTERD])
3042
  if not result.failed:
3043
    _Fail("The master daemon is running, will not demote")
3044

    
3045
  try:
3046
    if os.path.isfile(pathutils.CLUSTER_CONF_FILE):
3047
      utils.CreateBackup(pathutils.CLUSTER_CONF_FILE)
3048
  except EnvironmentError, err:
3049
    if err.errno != errno.ENOENT:
3050
      _Fail("Error while backing up cluster file: %s", err, exc=True)
3051

    
3052
  utils.RemoveFile(pathutils.CLUSTER_CONF_FILE)
3053

    
3054

    
3055
def _GetX509Filenames(cryptodir, name):
3056
  """Returns the full paths for the private key and certificate.
3057

3058
  """
3059
  return (utils.PathJoin(cryptodir, name),
3060
          utils.PathJoin(cryptodir, name, _X509_KEY_FILE),
3061
          utils.PathJoin(cryptodir, name, _X509_CERT_FILE))
3062

    
3063

    
3064
def CreateX509Certificate(validity, cryptodir=pathutils.CRYPTO_KEYS_DIR):
3065
  """Creates a new X509 certificate for SSL/TLS.
3066

3067
  @type validity: int
3068
  @param validity: Validity in seconds
3069
  @rtype: tuple; (string, string)
3070
  @return: Certificate name and public part
3071

3072
  """
3073
  (key_pem, cert_pem) = \
3074
    utils.GenerateSelfSignedX509Cert(netutils.Hostname.GetSysName(),
3075
                                     min(validity, _MAX_SSL_CERT_VALIDITY))
3076

    
3077
  cert_dir = tempfile.mkdtemp(dir=cryptodir,
3078
                              prefix="x509-%s-" % utils.TimestampForFilename())
3079
  try:
3080
    name = os.path.basename(cert_dir)
3081
    assert len(name) > 5
3082

    
3083
    (_, key_file, cert_file) = _GetX509Filenames(cryptodir, name)
3084

    
3085
    utils.WriteFile(key_file, mode=0400, data=key_pem)
3086
    utils.WriteFile(cert_file, mode=0400, data=cert_pem)
3087

    
3088
    # Never return private key as it shouldn't leave the node
3089
    return (name, cert_pem)
3090
  except Exception:
3091
    shutil.rmtree(cert_dir, ignore_errors=True)
3092
    raise
3093

    
3094

    
3095
def RemoveX509Certificate(name, cryptodir=pathutils.CRYPTO_KEYS_DIR):
3096
  """Removes a X509 certificate.
3097

3098
  @type name: string
3099
  @param name: Certificate name
3100

3101
  """
3102
  (cert_dir, key_file, cert_file) = _GetX509Filenames(cryptodir, name)
3103

    
3104
  utils.RemoveFile(key_file)
3105
  utils.RemoveFile(cert_file)
3106

    
3107
  try:
3108
    os.rmdir(cert_dir)
3109
  except EnvironmentError, err:
3110
    _Fail("Cannot remove certificate directory '%s': %s",
3111
          cert_dir, err)
3112

    
3113

    
3114
def _GetImportExportIoCommand(instance, mode, ieio, ieargs):
3115
  """Returns the command for the requested input/output.
3116

3117
  @type instance: L{objects.Instance}
3118
  @param instance: The instance object
3119
  @param mode: Import/export mode
3120
  @param ieio: Input/output type
3121
  @param ieargs: Input/output arguments
3122

3123
  """
3124
  assert mode in (constants.IEM_IMPORT, constants.IEM_EXPORT)
3125

    
3126
  env = None
3127
  prefix = None
3128
  suffix = None
3129
  exp_size = None
3130

    
3131
  if ieio == constants.IEIO_FILE:
3132
    (filename, ) = ieargs
3133

    
3134
    if not utils.IsNormAbsPath(filename):
3135
      _Fail("Path '%s' is not normalized or absolute", filename)
3136

    
3137
    real_filename = os.path.realpath(filename)
3138
    directory = os.path.dirname(real_filename)
3139

    
3140
    if not utils.IsBelowDir(pathutils.EXPORT_DIR, real_filename):
3141
      _Fail("File '%s' is not under exports directory '%s': %s",
3142
            filename, pathutils.EXPORT_DIR, real_filename)
3143

    
3144
    # Create directory
3145
    utils.Makedirs(directory, mode=0750)
3146

    
3147
    quoted_filename = utils.ShellQuote(filename)
3148

    
3149
    if mode == constants.IEM_IMPORT:
3150
      suffix = "> %s" % quoted_filename
3151
    elif mode == constants.IEM_EXPORT:
3152
      suffix = "< %s" % quoted_filename
3153

    
3154
      # Retrieve file size
3155
      try:
3156
        st = os.stat(filename)
3157
      except EnvironmentError, err:
3158
        logging.error("Can't stat(2) %s: %s", filename, err)
3159
      else:
3160
        exp_size = utils.BytesToMebibyte(st.st_size)
3161

    
3162
  elif ieio == constants.IEIO_RAW_DISK:
3163
    (disk, ) = ieargs
3164

    
3165
    real_disk = _OpenRealBD(disk)
3166

    
3167
    if mode == constants.IEM_IMPORT:
3168
      # we set here a smaller block size as, due to transport buffering, more
3169
      # than 64-128k will mostly ignored; we use nocreat to fail if the device
3170
      # is not already there or we pass a wrong path; we use notrunc to no
3171
      # attempt truncate on an LV device; we use oflag=dsync to not buffer too
3172
      # much memory; this means that at best, we flush every 64k, which will
3173
      # not be very fast
3174
      suffix = utils.BuildShellCmd(("| dd of=%s conv=nocreat,notrunc"
3175
                                    " bs=%s oflag=dsync"),
3176
                                    real_disk.dev_path,
3177
                                    str(64 * 1024))
3178

    
3179
    elif mode == constants.IEM_EXPORT:
3180
      # the block size on the read dd is 1MiB to match our units
3181
      prefix = utils.BuildShellCmd("dd if=%s bs=%s count=%s |",
3182
                                   real_disk.dev_path,
3183
                                   str(1024 * 1024), # 1 MB
3184
                                   str(disk.size))
3185
      exp_size = disk.size
3186

    
3187
  elif ieio == constants.IEIO_SCRIPT:
3188
    (disk, disk_index, ) = ieargs
3189

    
3190
    assert isinstance(disk_index, (int, long))
3191

    
3192
    real_disk = _OpenRealBD(disk)
3193

    
3194
    inst_os = OSFromDisk(instance.os)
3195
    env = OSEnvironment(instance, inst_os)
3196

    
3197
    if mode == constants.IEM_IMPORT:
3198
      env["IMPORT_DEVICE"] = env["DISK_%d_PATH" % disk_index]
3199
      env["IMPORT_INDEX"] = str(disk_index)
3200
      script = inst_os.import_script
3201

    
3202
    elif mode == constants.IEM_EXPORT:
3203
      env["EXPORT_DEVICE"] = real_disk.dev_path
3204
      env["EXPORT_INDEX"] = str(disk_index)
3205
      script = inst_os.export_script
3206

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

    
3210
    if mode == constants.IEM_IMPORT:
3211
      suffix = "| %s" % script_cmd
3212

    
3213
    elif mode == constants.IEM_EXPORT:
3214
      prefix = "%s |" % script_cmd
3215

    
3216
    # Let script predict size
3217
    exp_size = constants.IE_CUSTOM_SIZE
3218

    
3219
  else:
3220
    _Fail("Invalid %s I/O mode %r", mode, ieio)
3221

    
3222
  return (env, prefix, suffix, exp_size)
3223

    
3224

    
3225
def _CreateImportExportStatusDir(prefix):
3226
  """Creates status directory for import/export.
3227

3228
  """
3229
  return tempfile.mkdtemp(dir=pathutils.IMPORT_EXPORT_DIR,
3230
                          prefix=("%s-%s-" %
3231
                                  (prefix, utils.TimestampForFilename())))
3232

    
3233

    
3234
def StartImportExportDaemon(mode, opts, host, port, instance, component,
3235
                            ieio, ieioargs):
3236
  """Starts an import or export daemon.
3237

3238
  @param mode: Import/output mode
3239
  @type opts: L{objects.ImportExportOptions}
3240
  @param opts: Daemon options
3241
  @type host: string
3242
  @param host: Remote host for export (None for import)
3243
  @type port: int
3244
  @param port: Remote port for export (None for import)
3245
  @type instance: L{objects.Instance}
3246
  @param instance: Instance object
3247
  @type component: string
3248
  @param component: which part of the instance is transferred now,
3249
      e.g. 'disk/0'
3250
  @param ieio: Input/output type
3251
  @param ieioargs: Input/output arguments
3252

3253
  """
3254
  if mode == constants.IEM_IMPORT:
3255
    prefix = "import"
3256

    
3257
    if not (host is None and port is None):
3258
      _Fail("Can not specify host or port on import")
3259

    
3260
  elif mode == constants.IEM_EXPORT:
3261
    prefix = "export"
3262

    
3263
    if host is None or port is None:
3264
      _Fail("Host and port must be specified for an export")
3265

    
3266
  else:
3267
    _Fail("Invalid mode %r", mode)
3268

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

    
3272
  (cmd_env, cmd_prefix, cmd_suffix, exp_size) = \
3273
    _GetImportExportIoCommand(instance, mode, ieio, ieioargs)
3274

    
3275
  if opts.key_name is None:
3276
    # Use server.pem
3277
    key_path = pathutils.NODED_CERT_FILE
3278
    cert_path = pathutils.NODED_CERT_FILE
3279
    assert opts.ca_pem is None
3280
  else:
3281
    (_, key_path, cert_path) = _GetX509Filenames(pathutils.CRYPTO_KEYS_DIR,
3282
                                                 opts.key_name)
3283
    assert opts.ca_pem is not None
3284

    
3285
  for i in [key_path, cert_path]:
3286
    if not os.path.exists(i):
3287
      _Fail("File '%s' does not exist" % i)
3288

    
3289
  status_dir = _CreateImportExportStatusDir("%s-%s" % (prefix, component))
3290
  try:
3291
    status_file = utils.PathJoin(status_dir, _IES_STATUS_FILE)
3292
    pid_file = utils.PathJoin(status_dir, _IES_PID_FILE)
3293
    ca_file = utils.PathJoin(status_dir, _IES_CA_FILE)
3294

    
3295
    if opts.ca_pem is None:
3296
      # Use server.pem
3297
      ca = utils.ReadFile(pathutils.NODED_CERT_FILE)
3298
    else:
3299
      ca = opts.ca_pem
3300

    
3301
    # Write CA file
3302
    utils.WriteFile(ca_file, data=ca, mode=0400)
3303

    
3304
    cmd = [
3305
      pathutils.IMPORT_EXPORT_DAEMON,
3306
      status_file, mode,
3307
      "--key=%s" % key_path,
3308
      "--cert=%s" % cert_path,
3309
      "--ca=%s" % ca_file,
3310
      ]
3311

    
3312
    if host:
3313
      cmd.append("--host=%s" % host)
3314

    
3315
    if port:
3316
      cmd.append("--port=%s" % port)
3317

    
3318
    if opts.ipv6:
3319
      cmd.append("--ipv6")
3320
    else:
3321
      cmd.append("--ipv4")
3322

    
3323
    if opts.compress:
3324
      cmd.append("--compress=%s" % opts.compress)
3325

    
3326
    if opts.magic:
3327
      cmd.append("--magic=%s" % opts.magic)
3328

    
3329
    if exp_size is not None:
3330
      cmd.append("--expected-size=%s" % exp_size)
3331

    
3332
    if cmd_prefix:
3333
      cmd.append("--cmd-prefix=%s" % cmd_prefix)
3334

    
3335
    if cmd_suffix:
3336
      cmd.append("--cmd-suffix=%s" % cmd_suffix)
3337

    
3338
    if mode == constants.IEM_EXPORT:
3339
      # Retry connection a few times when connecting to remote peer
3340
      cmd.append("--connect-retries=%s" % constants.RIE_CONNECT_RETRIES)
3341
      cmd.append("--connect-timeout=%s" % constants.RIE_CONNECT_ATTEMPT_TIMEOUT)
3342
    elif opts.connect_timeout is not None:
3343
      assert mode == constants.IEM_IMPORT
3344
      # Overall timeout for establishing connection while listening
3345
      cmd.append("--connect-timeout=%s" % opts.connect_timeout)
3346

    
3347
    logfile = _InstanceLogName(prefix, instance.os, instance.name, component)
3348

    
3349
    # TODO: Once _InstanceLogName uses tempfile.mkstemp, StartDaemon has
3350
    # support for receiving a file descriptor for output
3351
    utils.StartDaemon(cmd, env=cmd_env, pidfile=pid_file,
3352
                      output=logfile)
3353

    
3354
    # The import/export name is simply the status directory name
3355
    return os.path.basename(status_dir)
3356

    
3357
  except Exception:
3358
    shutil.rmtree(status_dir, ignore_errors=True)
3359
    raise
3360

    
3361

    
3362
def GetImportExportStatus(names):
3363
  """Returns import/export daemon status.
3364

3365
  @type names: sequence
3366
  @param names: List of names
3367
  @rtype: List of dicts
3368
  @return: Returns a list of the state of each named import/export or None if a
3369
           status couldn't be read
3370

3371
  """
3372
  result = []
3373

    
3374
  for name in names:
3375
    status_file = utils.PathJoin(pathutils.IMPORT_EXPORT_DIR, name,
3376
                                 _IES_STATUS_FILE)
3377

    
3378
    try:
3379
      data = utils.ReadFile(status_file)
3380
    except EnvironmentError, err:
3381
      if err.errno != errno.ENOENT:
3382
        raise
3383
      data = None
3384

    
3385
    if not data:
3386
      result.append(None)
3387
      continue
3388

    
3389
    result.append(serializer.LoadJson(data))
3390

    
3391
  return result
3392

    
3393

    
3394
def AbortImportExport(name):
3395
  """Sends SIGTERM to a running import/export daemon.
3396

3397
  """
3398
  logging.info("Abort import/export %s", name)
3399

    
3400
  status_dir = utils.PathJoin(pathutils.IMPORT_EXPORT_DIR, name)
3401
  pid = utils.ReadLockedPidFile(utils.PathJoin(status_dir, _IES_PID_FILE))
3402

    
3403
  if pid:
3404
    logging.info("Import/export %s is running with PID %s, sending SIGTERM",
3405
                 name, pid)
3406
    utils.IgnoreProcessNotFound(os.kill, pid, signal.SIGTERM)
3407

    
3408

    
3409
def CleanupImportExport(name):
3410
  """Cleanup after an import or export.
3411

3412
  If the import/export daemon is still running it's killed. Afterwards the
3413
  whole status directory is removed.
3414

3415
  """
3416
  logging.info("Finalizing import/export %s", name)
3417

    
3418
  status_dir = utils.PathJoin(pathutils.IMPORT_EXPORT_DIR, name)
3419

    
3420
  pid = utils.ReadLockedPidFile(utils.PathJoin(status_dir, _IES_PID_FILE))
3421

    
3422
  if pid:
3423
    logging.info("Import/export %s is still running with PID %s",
3424
                 name, pid)
3425
    utils.KillProcess(pid, waitpid=False)
3426

    
3427
  shutil.rmtree(status_dir, ignore_errors=True)
3428

    
3429

    
3430
def _FindDisks(nodes_ip, disks):
3431
  """Sets the physical ID on disks and returns the block devices.
3432

3433
  """
3434
  # set the correct physical ID
3435
  my_name = netutils.Hostname.GetSysName()
3436
  for cf in disks:
3437
    cf.SetPhysicalID(my_name, nodes_ip)
3438

    
3439
  bdevs = []
3440

    
3441
  for cf in disks:
3442
    rd = _RecursiveFindBD(cf)
3443
    if rd is None:
3444
      _Fail("Can't find device %s", cf)
3445
    bdevs.append(rd)
3446
  return bdevs
3447

    
3448

    
3449
def DrbdDisconnectNet(nodes_ip, disks):
3450
  """Disconnects the network on a list of drbd devices.
3451

3452
  """
3453
  bdevs = _FindDisks(nodes_ip, disks)
3454

    
3455
  # disconnect disks
3456
  for rd in bdevs:
3457
    try:
3458
      rd.DisconnectNet()
3459
    except errors.BlockDeviceError, err:
3460
      _Fail("Can't change network configuration to standalone mode: %s",
3461
            err, exc=True)
3462

    
3463

    
3464
def DrbdAttachNet(nodes_ip, disks, instance_name, multimaster):
3465
  """Attaches the network on a list of drbd devices.
3466

3467
  """
3468
  bdevs = _FindDisks(nodes_ip, disks)
3469

    
3470
  if multimaster:
3471
    for idx, rd in enumerate(bdevs):
3472
      try:
3473
        _SymlinkBlockDev(instance_name, rd.dev_path, idx)
3474
      except EnvironmentError, err:
3475
        _Fail("Can't create symlink: %s", err)
3476
  # reconnect disks, switch to new master configuration and if
3477
  # needed primary mode
3478
  for rd in bdevs:
3479
    try:
3480
      rd.AttachNet(multimaster)
3481
    except errors.BlockDeviceError, err:
3482
      _Fail("Can't change network configuration: %s", err)
3483

    
3484
  # wait until the disks are connected; we need to retry the re-attach
3485
  # if the device becomes standalone, as this might happen if the one
3486
  # node disconnects and reconnects in a different mode before the
3487
  # other node reconnects; in this case, one or both of the nodes will
3488
  # decide it has wrong configuration and switch to standalone
3489

    
3490
  def _Attach():
3491
    all_connected = True
3492

    
3493
    for rd in bdevs:
3494
      stats = rd.GetProcStatus()
3495

    
3496
      all_connected = (all_connected and
3497
                       (stats.is_connected or stats.is_in_resync))
3498

    
3499
      if stats.is_standalone:
3500
        # peer had different config info and this node became
3501
        # standalone, even though this should not happen with the
3502
        # new staged way of changing disk configs
3503
        try:
3504
          rd.AttachNet(multimaster)
3505
        except errors.BlockDeviceError, err:
3506
          _Fail("Can't change network configuration: %s", err)
3507

    
3508
    if not all_connected:
3509
      raise utils.RetryAgain()
3510

    
3511
  try:
3512
    # Start with a delay of 100 miliseconds and go up to 5 seconds
3513
    utils.Retry(_Attach, (0.1, 1.5, 5.0), 2 * 60)
3514
  except utils.RetryTimeout:
3515
    _Fail("Timeout in disk reconnecting")
3516

    
3517
  if multimaster:
3518
    # change to primary mode
3519
    for rd in bdevs:
3520
      try:
3521
        rd.Open()
3522
      except errors.BlockDeviceError, err:
3523
        _Fail("Can't change to primary mode: %s", err)
3524

    
3525

    
3526
def DrbdWaitSync(nodes_ip, disks):
3527
  """Wait until DRBDs have synchronized.
3528

3529
  """
3530
  def _helper(rd):
3531
    stats = rd.GetProcStatus()
3532
    if not (stats.is_connected or stats.is_in_resync):
3533
      raise utils.RetryAgain()
3534
    return stats
3535

    
3536
  bdevs = _FindDisks(nodes_ip, disks)
3537

    
3538
  min_resync = 100
3539
  alldone = True
3540
  for rd in bdevs:
3541
    try:
3542
      # poll each second for 15 seconds
3543
      stats = utils.Retry(_helper, 1, 15, args=[rd])
3544
    except utils.RetryTimeout:
3545
      stats = rd.GetProcStatus()
3546
      # last check
3547
      if not (stats.is_connected or stats.is_in_resync):
3548
        _Fail("DRBD device %s is not in sync: stats=%s", rd, stats)
3549
    alldone = alldone and (not stats.is_in_resync)
3550
    if stats.sync_percent is not None:
3551
      min_resync = min(min_resync, stats.sync_percent)
3552

    
3553
  return (alldone, min_resync)
3554

    
3555

    
3556
def GetDrbdUsermodeHelper():
3557
  """Returns DRBD usermode helper currently configured.
3558

3559
  """
3560
  try:
3561
    return bdev.BaseDRBD.GetUsermodeHelper()
3562
  except errors.BlockDeviceError, err:
3563
    _Fail(str(err))
3564

    
3565

    
3566
def PowercycleNode(hypervisor_type):
3567
  """Hard-powercycle the node.
3568

3569
  Because we need to return first, and schedule the powercycle in the
3570
  background, we won't be able to report failures nicely.
3571

3572
  """
3573
  hyper = hypervisor.GetHypervisor(hypervisor_type)
3574
  try:
3575
    pid = os.fork()
3576
  except OSError:
3577
    # if we can't fork, we'll pretend that we're in the child process
3578
    pid = 0
3579
  if pid > 0:
3580
    return "Reboot scheduled in 5 seconds"
3581
  # ensure the child is running on ram
3582
  try:
3583
    utils.Mlockall()
3584
  except Exception: # pylint: disable=W0703
3585
    pass
3586
  time.sleep(5)
3587
  hyper.PowercycleNode()
3588

    
3589

    
3590
def _VerifyRestrictedCmdName(cmd):
3591
  """Verifies a remote command name.
3592

3593
  @type cmd: string
3594
  @param cmd: Command name
3595
  @rtype: tuple; (boolean, string or None)
3596
  @return: The tuple's first element is the status; if C{False}, the second
3597
    element is an error message string, otherwise it's C{None}
3598

3599
  """
3600
  if not cmd.strip():
3601
    return (False, "Missing command name")
3602

    
3603
  if os.path.basename(cmd) != cmd:
3604
    return (False, "Invalid command name")
3605

    
3606
  if not constants.EXT_PLUGIN_MASK.match(cmd):
3607
    return (False, "Command name contains forbidden characters")
3608

    
3609
  return (True, None)
3610

    
3611

    
3612
def _CommonRestrictedCmdCheck(path, owner):
3613
  """Common checks for remote command file system directories and files.
3614

3615
  @type path: string
3616
  @param path: Path to check
3617
  @param owner: C{None} or tuple containing UID and GID
3618
  @rtype: tuple; (boolean, string or C{os.stat} result)
3619
  @return: The tuple's first element is the status; if C{False}, the second
3620
    element is an error message string, otherwise it's the result of C{os.stat}
3621

3622
  """
3623
  if owner is None:
3624
    # Default to root as owner
3625
    owner = (0, 0)
3626

    
3627
  try:
3628
    st = os.stat(path)
3629
  except EnvironmentError, err:
3630
    return (False, "Can't stat(2) '%s': %s" % (path, err))
3631

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

    
3635
  if (st.st_uid, st.st_gid) != owner:
3636
    (owner_uid, owner_gid) = owner
3637
    return (False, "'%s' is not owned by %s:%s" % (path, owner_uid, owner_gid))
3638

    
3639
  return (True, st)
3640

    
3641

    
3642
def _VerifyRestrictedCmdDirectory(path, _owner=None):
3643
  """Verifies remote command directory.
3644

3645
  @type path: string
3646
  @param path: Path to check
3647
  @rtype: tuple; (boolean, string or None)
3648
  @return: The tuple's first element is the status; if C{False}, the second
3649
    element is an error message string, otherwise it's C{None}
3650

3651
  """
3652
  (status, value) = _CommonRestrictedCmdCheck(path, _owner)
3653

    
3654
  if not status:
3655
    return (False, value)
3656

    
3657
  if not stat.S_ISDIR(value.st_mode):
3658
    return (False, "Path '%s' is not a directory" % path)
3659

    
3660
  return (True, None)
3661

    
3662

    
3663
def _VerifyRestrictedCmd(path, cmd, _owner=None):
3664
  """Verifies a whole remote command and returns its executable filename.
3665

3666
  @type path: string
3667
  @param path: Directory containing remote commands
3668
  @type cmd: string
3669
  @param cmd: Command name
3670
  @rtype: tuple; (boolean, string)
3671
  @return: The tuple's first element is the status; if C{False}, the second
3672
    element is an error message string, otherwise the second element is the
3673
    absolute path to the executable
3674

3675
  """
3676
  executable = utils.PathJoin(path, cmd)
3677

    
3678
  (status, msg) = _CommonRestrictedCmdCheck(executable, _owner)
3679

    
3680
  if not status:
3681
    return (False, msg)
3682

    
3683
  if not utils.IsExecutable(executable):
3684
    return (False, "access(2) thinks '%s' can't be executed" % executable)
3685

    
3686
  return (True, executable)
3687

    
3688

    
3689
def _PrepareRestrictedCmd(path, cmd,
3690
                          _verify_dir=_VerifyRestrictedCmdDirectory,
3691
                          _verify_name=_VerifyRestrictedCmdName,
3692
                          _verify_cmd=_VerifyRestrictedCmd):
3693
  """Performs a number of tests on a remote command.
3694

3695
  @type path: string
3696
  @param path: Directory containing remote commands
3697
  @type cmd: string
3698
  @param cmd: Command name
3699
  @return: Same as L{_VerifyRestrictedCmd}
3700

3701
  """
3702
  # Verify the directory first
3703
  (status, msg) = _verify_dir(path)
3704
  if status:
3705
    # Check command if everything was alright
3706
    (status, msg) = _verify_name(cmd)
3707

    
3708
  if not status:
3709
    return (False, msg)
3710

    
3711
  # Check actual executable
3712
  return _verify_cmd(path, cmd)
3713

    
3714

    
3715
def RunRestrictedCmd(cmd,
3716
                     _lock_timeout=_RCMD_LOCK_TIMEOUT,
3717
                     _lock_file=pathutils.RESTRICTED_COMMANDS_LOCK_FILE,
3718
                     _path=pathutils.RESTRICTED_COMMANDS_DIR,
3719
                     _sleep_fn=time.sleep,
3720
                     _prepare_fn=_PrepareRestrictedCmd,
3721
                     _runcmd_fn=utils.RunCmd,
3722
                     _enabled=constants.ENABLE_RESTRICTED_COMMANDS):
3723
  """Executes a remote command after performing strict tests.
3724

3725
  @type cmd: string
3726
  @param cmd: Command name
3727
  @rtype: string
3728
  @return: Command output
3729
  @raise RPCFail: In case of an error
3730

3731
  """
3732
  logging.info("Preparing to run remote command '%s'", cmd)
3733

    
3734
  if not _enabled:
3735
    _Fail("Remote commands disabled at configure time")
3736

    
3737
  lock = None
3738
  try:
3739
    cmdresult = None
3740
    try:
3741
      lock = utils.FileLock.Open(_lock_file)
3742
      lock.Exclusive(blocking=True, timeout=_lock_timeout)
3743

    
3744
      (status, value) = _prepare_fn(_path, cmd)
3745

    
3746
      if status:
3747
        cmdresult = _runcmd_fn([value], env={}, reset_env=True,
3748
                               postfork_fn=lambda _: lock.Unlock())
3749
      else:
3750
        logging.error(value)
3751
    except Exception: # pylint: disable=W0703
3752
      # Keep original error in log
3753
      logging.exception("Caught exception")
3754

    
3755
    if cmdresult is None:
3756
      logging.info("Sleeping for %0.1f seconds before returning",
3757
                   _RCMD_INVALID_DELAY)
3758
      _sleep_fn(_RCMD_INVALID_DELAY)
3759

    
3760
      # Do not include original error message in returned error
3761
      _Fail("Executing command '%s' failed" % cmd)
3762
    elif cmdresult.failed or cmdresult.fail_reason:
3763
      _Fail("Remote command '%s' failed: %s; output: %s",
3764
            cmd, cmdresult.fail_reason, cmdresult.output)
3765
    else:
3766
      return cmdresult.output
3767
  finally:
3768
    if lock is not None:
3769
      # Release lock at last
3770
      lock.Close()
3771
      lock = None
3772

    
3773

    
3774
def SetWatcherPause(until, _filename=pathutils.WATCHER_PAUSEFILE):
3775
  """Creates or removes the watcher pause file.
3776

3777
  @type until: None or number
3778
  @param until: Unix timestamp saying until when the watcher shouldn't run
3779

3780
  """
3781
  if until is None:
3782
    logging.info("Received request to no longer pause watcher")
3783
    utils.RemoveFile(_filename)
3784
  else:
3785
    logging.info("Received request to pause watcher until %s", until)
3786

    
3787
    if not ht.TNumber(until):
3788
      _Fail("Duration must be numeric")
3789

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

    
3792

    
3793
class HooksRunner(object):
3794
  """Hook runner.
3795

3796
  This class is instantiated on the node side (ganeti-noded) and not
3797
  on the master side.
3798

3799
  """
3800
  def __init__(self, hooks_base_dir=None):
3801
    """Constructor for hooks runner.
3802

3803
    @type hooks_base_dir: str or None
3804
    @param hooks_base_dir: if not None, this overrides the
3805
        L{pathutils.HOOKS_BASE_DIR} (useful for unittests)
3806

3807
    """
3808
    if hooks_base_dir is None:
3809
      hooks_base_dir = pathutils.HOOKS_BASE_DIR
3810
    # yeah, _BASE_DIR is not valid for attributes, we use it like a
3811
    # constant
3812
    self._BASE_DIR = hooks_base_dir # pylint: disable=C0103
3813

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

3817
    """
3818
    assert len(node_list) == 1
3819
    node = node_list[0]
3820
    _, myself = ssconf.GetMasterAndMyself()
3821
    assert node == myself
3822

    
3823
    results = self.RunHooks(hpath, phase, env)
3824

    
3825
    # Return values in the form expected by HooksMaster
3826
    return {node: (None, False, results)}
3827

    
3828
  def RunHooks(self, hpath, phase, env):
3829
    """Run the scripts in the hooks directory.
3830

3831
    @type hpath: str
3832
    @param hpath: the path to the hooks directory which
3833
        holds the scripts
3834
    @type phase: str
3835
    @param phase: either L{constants.HOOKS_PHASE_PRE} or
3836
        L{constants.HOOKS_PHASE_POST}
3837
    @type env: dict
3838
    @param env: dictionary with the environment for the hook
3839
    @rtype: list
3840
    @return: list of 3-element tuples:
3841
      - script path
3842
      - script result, either L{constants.HKR_SUCCESS} or
3843
        L{constants.HKR_FAIL}
3844
      - output of the script
3845

3846
    @raise errors.ProgrammerError: for invalid input
3847
        parameters
3848

3849
    """
3850
    if phase == constants.HOOKS_PHASE_PRE:
3851
      suffix = "pre"
3852
    elif phase == constants.HOOKS_PHASE_POST:
3853
      suffix = "post"
3854
    else:
3855
      _Fail("Unknown hooks phase '%s'", phase)
3856

    
3857
    subdir = "%s-%s.d" % (hpath, suffix)
3858
    dir_name = utils.PathJoin(self._BASE_DIR, subdir)
3859

    
3860
    results = []
3861

    
3862
    if not os.path.isdir(dir_name):
3863
      # for non-existing/non-dirs, we simply exit instead of logging a
3864
      # warning at every operation
3865
      return results
3866

    
3867
    runparts_results = utils.RunParts(dir_name, env=env, reset_env=True)
3868

    
3869
    for (relname, relstatus, runresult) in runparts_results:
3870
      if relstatus == constants.RUNPARTS_SKIP:
3871
        rrval = constants.HKR_SKIP
3872
        output = ""
3873
      elif relstatus == constants.RUNPARTS_ERR:
3874
        rrval = constants.HKR_FAIL
3875
        output = "Hook script execution error: %s" % runresult
3876
      elif relstatus == constants.RUNPARTS_RUN:
3877
        if runresult.failed:
3878
          rrval = constants.HKR_FAIL
3879
        else:
3880
          rrval = constants.HKR_SUCCESS
3881
        output = utils.SafeEncode(runresult.output.strip())
3882
      results.append(("%s/%s" % (subdir, relname), rrval, output))
3883

    
3884
    return results
3885

    
3886

    
3887
class IAllocatorRunner(object):
3888
  """IAllocator runner.
3889

3890
  This class is instantiated on the node side (ganeti-noded) and not on
3891
  the master side.
3892

3893
  """
3894
  @staticmethod
3895
  def Run(name, idata):
3896
    """Run an iallocator script.
3897

3898
    @type name: str
3899
    @param name: the iallocator script name
3900
    @type idata: str
3901
    @param idata: the allocator input data
3902

3903
    @rtype: tuple
3904
    @return: two element tuple of:
3905
       - status
3906
       - either error message or stdout of allocator (for success)
3907

3908
    """
3909
    alloc_script = utils.FindFile(name, constants.IALLOCATOR_SEARCH_PATH,
3910
                                  os.path.isfile)
3911
    if alloc_script is None:
3912
      _Fail("iallocator module '%s' not found in the search path", name)
3913

    
3914
    fd, fin_name = tempfile.mkstemp(prefix="ganeti-iallocator.")
3915
    try:
3916
      os.write(fd, idata)
3917
      os.close(fd)
3918
      result = utils.RunCmd([alloc_script, fin_name])
3919
      if result.failed:
3920
        _Fail("iallocator module '%s' failed: %s, output '%s'",
3921
              name, result.fail_reason, result.output)
3922
    finally:
3923
      os.unlink(fin_name)
3924

    
3925
    return result.stdout
3926

    
3927

    
3928
class DevCacheManager(object):
3929
  """Simple class for managing a cache of block device information.
3930

3931
  """
3932
  _DEV_PREFIX = "/dev/"
3933
  _ROOT_DIR = pathutils.BDEV_CACHE_DIR
3934

    
3935
  @classmethod
3936
  def _ConvertPath(cls, dev_path):
3937
    """Converts a /dev/name path to the cache file name.
3938

3939
    This replaces slashes with underscores and strips the /dev
3940
    prefix. It then returns the full path to the cache file.
3941

3942
    @type dev_path: str
3943
    @param dev_path: the C{/dev/} path name
3944
    @rtype: str
3945
    @return: the converted path name
3946

3947
    """
3948
    if dev_path.startswith(cls._DEV_PREFIX):
3949
      dev_path = dev_path[len(cls._DEV_PREFIX):]
3950
    dev_path = dev_path.replace("/", "_")
3951
    fpath = utils.PathJoin(cls._ROOT_DIR, "bdev_%s" % dev_path)
3952
    return fpath
3953

    
3954
  @classmethod
3955
  def UpdateCache(cls, dev_path, owner, on_primary, iv_name):
3956
    """Updates the cache information for a given device.
3957

3958
    @type dev_path: str
3959
    @param dev_path: the pathname of the device
3960
    @type owner: str
3961
    @param owner: the owner (instance name) of the device
3962
    @type on_primary: bool
3963
    @param on_primary: whether this is the primary
3964
        node nor not
3965
    @type iv_name: str
3966
    @param iv_name: the instance-visible name of the
3967
        device, as in objects.Disk.iv_name
3968

3969
    @rtype: None
3970

3971
    """
3972
    if dev_path is None:
3973
      logging.error("DevCacheManager.UpdateCache got a None dev_path")
3974
      return
3975
    fpath = cls._ConvertPath(dev_path)
3976
    if on_primary:
3977
      state = "primary"
3978
    else:
3979
      state = "secondary"
3980
    if iv_name is None:
3981
      iv_name = "not_visible"
3982
    fdata = "%s %s %s\n" % (str(owner), state, iv_name)
3983
    try:
3984
      utils.WriteFile(fpath, data=fdata)
3985
    except EnvironmentError, err:
3986
      logging.exception("Can't update bdev cache for %s: %s", dev_path, err)
3987

    
3988
  @classmethod
3989
  def RemoveCache(cls, dev_path):
3990
    """Remove data for a dev_path.
3991

3992
    This is just a wrapper over L{utils.io.RemoveFile} with a converted
3993
    path name and logging.
3994

3995
    @type dev_path: str
3996
    @param dev_path: the pathname of the device
3997

3998
    @rtype: None
3999

4000
    """
4001
    if dev_path is None:
4002
      logging.error("DevCacheManager.RemoveCache got a None dev_path")
4003
      return
4004
    fpath = cls._ConvertPath(dev_path)
4005
    try:
4006
      utils.RemoveFile(fpath)
4007
    except EnvironmentError, err:
4008
      logging.exception("Can't update bdev cache for %s: %s", dev_path, err)