Statistics
| Branch: | Tag: | Revision:

root / lib / backend.py @ a59d5fa1

History | View | Annotate | Download (125.6 kB)

1
#
2
#
3

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

    
21

    
22
"""Functions used by the node daemon
23

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

29
"""
30

    
31
# pylint: disable=E1103
32

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

    
37

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

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

    
69

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

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

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

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

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

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

    
104

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

108
  Its argument is the error message.
109

110
  """
111

    
112

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

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

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

    
124

    
125
class InstReason(object):
126
  """Class representing the reason for a change of state of a VM.
127

128
  It is used to allow an easy serialization of the reason, so that it can be
129
  written on a file.
130

131
  """
132
  def __init__(self, source, text):
133
    """Initialize the class with all the required values.
134

135
    @type text: string
136
    @param text: The textual description of the reason for changing state
137
    @type source: string
138
    @param source: The source of the state change (RAPI, CLI, ...)
139

140
    """
141
    self.source = source
142
    self.text = text
143

    
144
  def GetJson(self):
145
    """Get the JSON representation of the InstReason.
146

147
    @rtype: string
148
    @return : The JSON representation of the object
149

150
    """
151
    return serializer.DumpJson(dict(source=self.source, text=self.text))
152

    
153
  def Store(self, instance_name):
154
    """Serialize on a file the reason for the last state change of an instance.
155

156
    The exact location of the file depends on the name of the instance and on
157
    the configuration of the Ganeti cluster defined at deploy time.
158

159
    @type instance_name: string
160
    @param instance_name: The name of the instance
161
    @rtype: None
162

163
    """
164
    filename = GetInstReasonFilename(instance_name)
165
    utils.WriteFile(filename, data=self.GetJson())
166

    
167

    
168
def _Fail(msg, *args, **kwargs):
169
  """Log an error and the raise an RPCFail exception.
170

171
  This exception is then handled specially in the ganeti daemon and
172
  turned into a 'failed' return type. As such, this function is a
173
  useful shortcut for logging the error and returning it to the master
174
  daemon.
175

176
  @type msg: string
177
  @param msg: the text of the exception
178
  @raise RPCFail
179

180
  """
181
  if args:
182
    msg = msg % args
183
  if "log" not in kwargs or kwargs["log"]: # if we should log this error
184
    if "exc" in kwargs and kwargs["exc"]:
185
      logging.exception(msg)
186
    else:
187
      logging.error(msg)
188
  raise RPCFail(msg)
189

    
190

    
191
def _GetConfig():
192
  """Simple wrapper to return a SimpleStore.
193

194
  @rtype: L{ssconf.SimpleStore}
195
  @return: a SimpleStore instance
196

197
  """
198
  return ssconf.SimpleStore()
199

    
200

    
201
def _GetSshRunner(cluster_name):
202
  """Simple wrapper to return an SshRunner.
203

204
  @type cluster_name: str
205
  @param cluster_name: the cluster name, which is needed
206
      by the SshRunner constructor
207
  @rtype: L{ssh.SshRunner}
208
  @return: an SshRunner instance
209

210
  """
211
  return ssh.SshRunner(cluster_name)
212

    
213

    
214
def _Decompress(data):
215
  """Unpacks data compressed by the RPC client.
216

217
  @type data: list or tuple
218
  @param data: Data sent by RPC client
219
  @rtype: str
220
  @return: Decompressed data
221

222
  """
223
  assert isinstance(data, (list, tuple))
224
  assert len(data) == 2
225
  (encoding, content) = data
226
  if encoding == constants.RPC_ENCODING_NONE:
227
    return content
228
  elif encoding == constants.RPC_ENCODING_ZLIB_BASE64:
229
    return zlib.decompress(base64.b64decode(content))
230
  else:
231
    raise AssertionError("Unknown data encoding")
232

    
233

    
234
def _CleanDirectory(path, exclude=None):
235
  """Removes all regular files in a directory.
236

237
  @type path: str
238
  @param path: the directory to clean
239
  @type exclude: list
240
  @param exclude: list of files to be excluded, defaults
241
      to the empty list
242

243
  """
244
  if path not in _ALLOWED_CLEAN_DIRS:
245
    _Fail("Path passed to _CleanDirectory not in allowed clean targets: '%s'",
246
          path)
247

    
248
  if not os.path.isdir(path):
249
    return
250
  if exclude is None:
251
    exclude = []
252
  else:
253
    # Normalize excluded paths
254
    exclude = [os.path.normpath(i) for i in exclude]
255

    
256
  for rel_name in utils.ListVisibleFiles(path):
257
    full_name = utils.PathJoin(path, rel_name)
258
    if full_name in exclude:
259
      continue
260
    if os.path.isfile(full_name) and not os.path.islink(full_name):
261
      utils.RemoveFile(full_name)
262

    
263

    
264
def _BuildUploadFileList():
265
  """Build the list of allowed upload files.
266

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

269
  """
270
  allowed_files = set([
271
    pathutils.CLUSTER_CONF_FILE,
272
    pathutils.ETC_HOSTS,
273
    pathutils.SSH_KNOWN_HOSTS_FILE,
274
    pathutils.VNC_PASSWORD_FILE,
275
    pathutils.RAPI_CERT_FILE,
276
    pathutils.SPICE_CERT_FILE,
277
    pathutils.SPICE_CACERT_FILE,
278
    pathutils.RAPI_USERS_FILE,
279
    pathutils.CONFD_HMAC_KEY,
280
    pathutils.CLUSTER_DOMAIN_SECRET_FILE,
281
    ])
282

    
283
  for hv_name in constants.HYPER_TYPES:
284
    hv_class = hypervisor.GetHypervisorClass(hv_name)
285
    allowed_files.update(hv_class.GetAncillaryFiles()[0])
286

    
287
  assert pathutils.FILE_STORAGE_PATHS_FILE not in allowed_files, \
288
    "Allowed file storage paths should never be uploaded via RPC"
289

    
290
  return frozenset(allowed_files)
291

    
292

    
293
_ALLOWED_UPLOAD_FILES = _BuildUploadFileList()
294

    
295

    
296
def JobQueuePurge():
297
  """Removes job queue files and archived jobs.
298

299
  @rtype: tuple
300
  @return: True, None
301

302
  """
303
  _CleanDirectory(pathutils.QUEUE_DIR, exclude=[pathutils.JOB_QUEUE_LOCK_FILE])
304
  _CleanDirectory(pathutils.JOB_QUEUE_ARCHIVE_DIR)
305

    
306

    
307
def GetMasterInfo():
308
  """Returns master information.
309

310
  This is an utility function to compute master information, either
311
  for consumption here or from the node daemon.
312

313
  @rtype: tuple
314
  @return: master_netdev, master_ip, master_name, primary_ip_family,
315
    master_netmask
316
  @raise RPCFail: in case of errors
317

318
  """
319
  try:
320
    cfg = _GetConfig()
321
    master_netdev = cfg.GetMasterNetdev()
322
    master_ip = cfg.GetMasterIP()
323
    master_netmask = cfg.GetMasterNetmask()
324
    master_node = cfg.GetMasterNode()
325
    primary_ip_family = cfg.GetPrimaryIPFamily()
326
  except errors.ConfigurationError, err:
327
    _Fail("Cluster configuration incomplete: %s", err, exc=True)
328
  return (master_netdev, master_ip, master_node, primary_ip_family,
329
          master_netmask)
330

    
331

    
332
def RunLocalHooks(hook_opcode, hooks_path, env_builder_fn):
333
  """Decorator that runs hooks before and after the decorated function.
334

335
  @type hook_opcode: string
336
  @param hook_opcode: opcode of the hook
337
  @type hooks_path: string
338
  @param hooks_path: path of the hooks
339
  @type env_builder_fn: function
340
  @param env_builder_fn: function that returns a dictionary containing the
341
    environment variables for the hooks. Will get all the parameters of the
342
    decorated function.
343
  @raise RPCFail: in case of pre-hook failure
344

345
  """
346
  def decorator(fn):
347
    def wrapper(*args, **kwargs):
348
      _, myself = ssconf.GetMasterAndMyself()
349
      nodes = ([myself], [myself])  # these hooks run locally
350

    
351
      env_fn = compat.partial(env_builder_fn, *args, **kwargs)
352

    
353
      cfg = _GetConfig()
354
      hr = HooksRunner()
355
      hm = mcpu.HooksMaster(hook_opcode, hooks_path, nodes, hr.RunLocalHooks,
356
                            None, env_fn, logging.warning, cfg.GetClusterName(),
357
                            cfg.GetMasterNode())
358

    
359
      hm.RunPhase(constants.HOOKS_PHASE_PRE)
360
      result = fn(*args, **kwargs)
361
      hm.RunPhase(constants.HOOKS_PHASE_POST)
362

    
363
      return result
364
    return wrapper
365
  return decorator
366

    
367

    
368
def _BuildMasterIpEnv(master_params, use_external_mip_script=None):
369
  """Builds environment variables for master IP hooks.
370

371
  @type master_params: L{objects.MasterNetworkParameters}
372
  @param master_params: network parameters of the master
373
  @type use_external_mip_script: boolean
374
  @param use_external_mip_script: whether to use an external master IP
375
    address setup script (unused, but necessary per the implementation of the
376
    _RunLocalHooks decorator)
377

378
  """
379
  # pylint: disable=W0613
380
  ver = netutils.IPAddress.GetVersionFromAddressFamily(master_params.ip_family)
381
  env = {
382
    "MASTER_NETDEV": master_params.netdev,
383
    "MASTER_IP": master_params.ip,
384
    "MASTER_NETMASK": str(master_params.netmask),
385
    "CLUSTER_IP_VERSION": str(ver),
386
  }
387

    
388
  return env
389

    
390

    
391
def _RunMasterSetupScript(master_params, action, use_external_mip_script):
392
  """Execute the master IP address setup script.
393

394
  @type master_params: L{objects.MasterNetworkParameters}
395
  @param master_params: network parameters of the master
396
  @type action: string
397
  @param action: action to pass to the script. Must be one of
398
    L{backend._MASTER_START} or L{backend._MASTER_STOP}
399
  @type use_external_mip_script: boolean
400
  @param use_external_mip_script: whether to use an external master IP
401
    address setup script
402
  @raise backend.RPCFail: if there are errors during the execution of the
403
    script
404

405
  """
406
  env = _BuildMasterIpEnv(master_params)
407

    
408
  if use_external_mip_script:
409
    setup_script = pathutils.EXTERNAL_MASTER_SETUP_SCRIPT
410
  else:
411
    setup_script = pathutils.DEFAULT_MASTER_SETUP_SCRIPT
412

    
413
  result = utils.RunCmd([setup_script, action], env=env, reset_env=True)
414

    
415
  if result.failed:
416
    _Fail("Failed to %s the master IP. Script return value: %s, output: '%s'" %
417
          (action, result.exit_code, result.output), log=True)
418

    
419

    
420
@RunLocalHooks(constants.FAKE_OP_MASTER_TURNUP, "master-ip-turnup",
421
               _BuildMasterIpEnv)
422
def ActivateMasterIp(master_params, use_external_mip_script):
423
  """Activate the IP address of the master daemon.
424

425
  @type master_params: L{objects.MasterNetworkParameters}
426
  @param master_params: network parameters of the master
427
  @type use_external_mip_script: boolean
428
  @param use_external_mip_script: whether to use an external master IP
429
    address setup script
430
  @raise RPCFail: in case of errors during the IP startup
431

432
  """
433
  _RunMasterSetupScript(master_params, _MASTER_START,
434
                        use_external_mip_script)
435

    
436

    
437
def StartMasterDaemons(no_voting):
438
  """Activate local node as master node.
439

440
  The function will start the master daemons (ganeti-masterd and ganeti-rapi).
441

442
  @type no_voting: boolean
443
  @param no_voting: whether to start ganeti-masterd without a node vote
444
      but still non-interactively
445
  @rtype: None
446

447
  """
448

    
449
  if no_voting:
450
    masterd_args = "--no-voting --yes-do-it"
451
  else:
452
    masterd_args = ""
453

    
454
  env = {
455
    "EXTRA_MASTERD_ARGS": masterd_args,
456
    }
457

    
458
  result = utils.RunCmd([pathutils.DAEMON_UTIL, "start-master"], env=env)
459
  if result.failed:
460
    msg = "Can't start Ganeti master: %s" % result.output
461
    logging.error(msg)
462
    _Fail(msg)
463

    
464

    
465
@RunLocalHooks(constants.FAKE_OP_MASTER_TURNDOWN, "master-ip-turndown",
466
               _BuildMasterIpEnv)
467
def DeactivateMasterIp(master_params, use_external_mip_script):
468
  """Deactivate the master IP on this node.
469

470
  @type master_params: L{objects.MasterNetworkParameters}
471
  @param master_params: network parameters of the master
472
  @type use_external_mip_script: boolean
473
  @param use_external_mip_script: whether to use an external master IP
474
    address setup script
475
  @raise RPCFail: in case of errors during the IP turndown
476

477
  """
478
  _RunMasterSetupScript(master_params, _MASTER_STOP,
479
                        use_external_mip_script)
480

    
481

    
482
def StopMasterDaemons():
483
  """Stop the master daemons on this node.
484

485
  Stop the master daemons (ganeti-masterd and ganeti-rapi) on this node.
486

487
  @rtype: None
488

489
  """
490
  # TODO: log and report back to the caller the error failures; we
491
  # need to decide in which case we fail the RPC for this
492

    
493
  result = utils.RunCmd([pathutils.DAEMON_UTIL, "stop-master"])
494
  if result.failed:
495
    logging.error("Could not stop Ganeti master, command %s had exitcode %s"
496
                  " and error %s",
497
                  result.cmd, result.exit_code, result.output)
498

    
499

    
500
def ChangeMasterNetmask(old_netmask, netmask, master_ip, master_netdev):
501
  """Change the netmask of the master IP.
502

503
  @param old_netmask: the old value of the netmask
504
  @param netmask: the new value of the netmask
505
  @param master_ip: the master IP
506
  @param master_netdev: the master network device
507

508
  """
509
  if old_netmask == netmask:
510
    return
511

    
512
  if not netutils.IPAddress.Own(master_ip):
513
    _Fail("The master IP address is not up, not attempting to change its"
514
          " netmask")
515

    
516
  result = utils.RunCmd([constants.IP_COMMAND_PATH, "address", "add",
517
                         "%s/%s" % (master_ip, netmask),
518
                         "dev", master_netdev, "label",
519
                         "%s:0" % master_netdev])
520
  if result.failed:
521
    _Fail("Could not set the new netmask on the master IP address")
522

    
523
  result = utils.RunCmd([constants.IP_COMMAND_PATH, "address", "del",
524
                         "%s/%s" % (master_ip, old_netmask),
525
                         "dev", master_netdev, "label",
526
                         "%s:0" % master_netdev])
527
  if result.failed:
528
    _Fail("Could not bring down the master IP address with the old netmask")
529

    
530

    
531
def EtcHostsModify(mode, host, ip):
532
  """Modify a host entry in /etc/hosts.
533

534
  @param mode: The mode to operate. Either add or remove entry
535
  @param host: The host to operate on
536
  @param ip: The ip associated with the entry
537

538
  """
539
  if mode == constants.ETC_HOSTS_ADD:
540
    if not ip:
541
      RPCFail("Mode 'add' needs 'ip' parameter, but parameter not"
542
              " present")
543
    utils.AddHostToEtcHosts(host, ip)
544
  elif mode == constants.ETC_HOSTS_REMOVE:
545
    if ip:
546
      RPCFail("Mode 'remove' does not allow 'ip' parameter, but"
547
              " parameter is present")
548
    utils.RemoveHostFromEtcHosts(host)
549
  else:
550
    RPCFail("Mode not supported")
551

    
552

    
553
def LeaveCluster(modify_ssh_setup):
554
  """Cleans up and remove the current node.
555

556
  This function cleans up and prepares the current node to be removed
557
  from the cluster.
558

559
  If processing is successful, then it raises an
560
  L{errors.QuitGanetiException} which is used as a special case to
561
  shutdown the node daemon.
562

563
  @param modify_ssh_setup: boolean
564

565
  """
566
  _CleanDirectory(pathutils.DATA_DIR)
567
  _CleanDirectory(pathutils.CRYPTO_KEYS_DIR)
568
  JobQueuePurge()
569

    
570
  if modify_ssh_setup:
571
    try:
572
      priv_key, pub_key, auth_keys = ssh.GetUserFiles(constants.SSH_LOGIN_USER)
573

    
574
      utils.RemoveAuthorizedKey(auth_keys, utils.ReadFile(pub_key))
575

    
576
      utils.RemoveFile(priv_key)
577
      utils.RemoveFile(pub_key)
578
    except errors.OpExecError:
579
      logging.exception("Error while processing ssh files")
580

    
581
  try:
582
    utils.RemoveFile(pathutils.CONFD_HMAC_KEY)
583
    utils.RemoveFile(pathutils.RAPI_CERT_FILE)
584
    utils.RemoveFile(pathutils.SPICE_CERT_FILE)
585
    utils.RemoveFile(pathutils.SPICE_CACERT_FILE)
586
    utils.RemoveFile(pathutils.NODED_CERT_FILE)
587
  except: # pylint: disable=W0702
588
    logging.exception("Error while removing cluster secrets")
589

    
590
  result = utils.RunCmd([pathutils.DAEMON_UTIL, "stop", constants.CONFD])
591
  if result.failed:
592
    logging.error("Command %s failed with exitcode %s and error %s",
593
                  result.cmd, result.exit_code, result.output)
594

    
595
  # Raise a custom exception (handled in ganeti-noded)
596
  raise errors.QuitGanetiException(True, "Shutdown scheduled")
597

    
598

    
599
def _GetVgInfo(name, excl_stor):
600
  """Retrieves information about a LVM volume group.
601

602
  """
603
  # TODO: GetVGInfo supports returning information for multiple VGs at once
604
  vginfo = bdev.LogicalVolume.GetVGInfo([name], excl_stor)
605
  if vginfo:
606
    vg_free = int(round(vginfo[0][0], 0))
607
    vg_size = int(round(vginfo[0][1], 0))
608
  else:
609
    vg_free = None
610
    vg_size = None
611

    
612
  return {
613
    "name": name,
614
    "vg_free": vg_free,
615
    "vg_size": vg_size,
616
    }
617

    
618

    
619
def _GetHvInfo(name):
620
  """Retrieves node information from a hypervisor.
621

622
  The information returned depends on the hypervisor. Common items:
623

624
    - vg_size is the size of the configured volume group in MiB
625
    - vg_free is the free size of the volume group in MiB
626
    - memory_dom0 is the memory allocated for domain0 in MiB
627
    - memory_free is the currently available (free) ram in MiB
628
    - memory_total is the total number of ram in MiB
629
    - hv_version: the hypervisor version, if available
630

631
  """
632
  return hypervisor.GetHypervisor(name).GetNodeInfo()
633

    
634

    
635
def _GetNamedNodeInfo(names, fn):
636
  """Calls C{fn} for all names in C{names} and returns a dictionary.
637

638
  @rtype: None or dict
639

640
  """
641
  if names is None:
642
    return None
643
  else:
644
    return map(fn, names)
645

    
646

    
647
def GetNodeInfo(vg_names, hv_names, excl_stor):
648
  """Gives back a hash with different information about the node.
649

650
  @type vg_names: list of string
651
  @param vg_names: Names of the volume groups to ask for disk space information
652
  @type hv_names: list of string
653
  @param hv_names: Names of the hypervisors to ask for node information
654
  @type excl_stor: boolean
655
  @param excl_stor: Whether exclusive_storage is active
656
  @rtype: tuple; (string, None/dict, None/dict)
657
  @return: Tuple containing boot ID, volume group information and hypervisor
658
    information
659

660
  """
661
  bootid = utils.ReadFile(_BOOT_ID_PATH, size=128).rstrip("\n")
662
  vg_info = _GetNamedNodeInfo(vg_names, (lambda vg: _GetVgInfo(vg, excl_stor)))
663
  hv_info = _GetNamedNodeInfo(hv_names, _GetHvInfo)
664

    
665
  return (bootid, vg_info, hv_info)
666

    
667

    
668
def _CheckExclusivePvs(pvi_list):
669
  """Check that PVs are not shared among LVs
670

671
  @type pvi_list: list of L{objects.LvmPvInfo} objects
672
  @param pvi_list: information about the PVs
673

674
  @rtype: list of tuples (string, list of strings)
675
  @return: offending volumes, as tuples: (pv_name, [lv1_name, lv2_name...])
676

677
  """
678
  res = []
679
  for pvi in pvi_list:
680
    if len(pvi.lv_list) > 1:
681
      res.append((pvi.name, pvi.lv_list))
682
  return res
683

    
684

    
685
def VerifyNode(what, cluster_name):
686
  """Verify the status of the local node.
687

688
  Based on the input L{what} parameter, various checks are done on the
689
  local node.
690

691
  If the I{filelist} key is present, this list of
692
  files is checksummed and the file/checksum pairs are returned.
693

694
  If the I{nodelist} key is present, we check that we have
695
  connectivity via ssh with the target nodes (and check the hostname
696
  report).
697

698
  If the I{node-net-test} key is present, we check that we have
699
  connectivity to the given nodes via both primary IP and, if
700
  applicable, secondary IPs.
701

702
  @type what: C{dict}
703
  @param what: a dictionary of things to check:
704
      - filelist: list of files for which to compute checksums
705
      - nodelist: list of nodes we should check ssh communication with
706
      - node-net-test: list of nodes we should check node daemon port
707
        connectivity with
708
      - hypervisor: list with hypervisors to run the verify for
709
  @rtype: dict
710
  @return: a dictionary with the same keys as the input dict, and
711
      values representing the result of the checks
712

713
  """
714
  result = {}
715
  my_name = netutils.Hostname.GetSysName()
716
  port = netutils.GetDaemonPort(constants.NODED)
717
  vm_capable = my_name not in what.get(constants.NV_VMNODES, [])
718

    
719
  if constants.NV_HYPERVISOR in what and vm_capable:
720
    result[constants.NV_HYPERVISOR] = tmp = {}
721
    for hv_name in what[constants.NV_HYPERVISOR]:
722
      try:
723
        val = hypervisor.GetHypervisor(hv_name).Verify()
724
      except errors.HypervisorError, err:
725
        val = "Error while checking hypervisor: %s" % str(err)
726
      tmp[hv_name] = val
727

    
728
  if constants.NV_HVPARAMS in what and vm_capable:
729
    result[constants.NV_HVPARAMS] = tmp = []
730
    for source, hv_name, hvparms in what[constants.NV_HVPARAMS]:
731
      try:
732
        logging.info("Validating hv %s, %s", hv_name, hvparms)
733
        hypervisor.GetHypervisor(hv_name).ValidateParameters(hvparms)
734
      except errors.HypervisorError, err:
735
        tmp.append((source, hv_name, str(err)))
736

    
737
  if constants.NV_FILELIST in what:
738
    fingerprints = utils.FingerprintFiles(map(vcluster.LocalizeVirtualPath,
739
                                              what[constants.NV_FILELIST]))
740
    result[constants.NV_FILELIST] = \
741
      dict((vcluster.MakeVirtualPath(key), value)
742
           for (key, value) in fingerprints.items())
743

    
744
  if constants.NV_NODELIST in what:
745
    (nodes, bynode) = what[constants.NV_NODELIST]
746

    
747
    # Add nodes from other groups (different for each node)
748
    try:
749
      nodes.extend(bynode[my_name])
750
    except KeyError:
751
      pass
752

    
753
    # Use a random order
754
    random.shuffle(nodes)
755

    
756
    # Try to contact all nodes
757
    val = {}
758
    for node in nodes:
759
      success, message = _GetSshRunner(cluster_name).VerifyNodeHostname(node)
760
      if not success:
761
        val[node] = message
762

    
763
    result[constants.NV_NODELIST] = val
764

    
765
  if constants.NV_NODENETTEST in what:
766
    result[constants.NV_NODENETTEST] = tmp = {}
767
    my_pip = my_sip = None
768
    for name, pip, sip in what[constants.NV_NODENETTEST]:
769
      if name == my_name:
770
        my_pip = pip
771
        my_sip = sip
772
        break
773
    if not my_pip:
774
      tmp[my_name] = ("Can't find my own primary/secondary IP"
775
                      " in the node list")
776
    else:
777
      for name, pip, sip in what[constants.NV_NODENETTEST]:
778
        fail = []
779
        if not netutils.TcpPing(pip, port, source=my_pip):
780
          fail.append("primary")
781
        if sip != pip:
782
          if not netutils.TcpPing(sip, port, source=my_sip):
783
            fail.append("secondary")
784
        if fail:
785
          tmp[name] = ("failure using the %s interface(s)" %
786
                       " and ".join(fail))
787

    
788
  if constants.NV_MASTERIP in what:
789
    # FIXME: add checks on incoming data structures (here and in the
790
    # rest of the function)
791
    master_name, master_ip = what[constants.NV_MASTERIP]
792
    if master_name == my_name:
793
      source = constants.IP4_ADDRESS_LOCALHOST
794
    else:
795
      source = None
796
    result[constants.NV_MASTERIP] = netutils.TcpPing(master_ip, port,
797
                                                     source=source)
798

    
799
  if constants.NV_USERSCRIPTS in what:
800
    result[constants.NV_USERSCRIPTS] = \
801
      [script for script in what[constants.NV_USERSCRIPTS]
802
       if not utils.IsExecutable(script)]
803

    
804
  if constants.NV_OOB_PATHS in what:
805
    result[constants.NV_OOB_PATHS] = tmp = []
806
    for path in what[constants.NV_OOB_PATHS]:
807
      try:
808
        st = os.stat(path)
809
      except OSError, err:
810
        tmp.append("error stating out of band helper: %s" % err)
811
      else:
812
        if stat.S_ISREG(st.st_mode):
813
          if stat.S_IMODE(st.st_mode) & stat.S_IXUSR:
814
            tmp.append(None)
815
          else:
816
            tmp.append("out of band helper %s is not executable" % path)
817
        else:
818
          tmp.append("out of band helper %s is not a file" % path)
819

    
820
  if constants.NV_LVLIST in what and vm_capable:
821
    try:
822
      val = GetVolumeList(utils.ListVolumeGroups().keys())
823
    except RPCFail, err:
824
      val = str(err)
825
    result[constants.NV_LVLIST] = val
826

    
827
  if constants.NV_INSTANCELIST in what and vm_capable:
828
    # GetInstanceList can fail
829
    try:
830
      val = GetInstanceList(what[constants.NV_INSTANCELIST])
831
    except RPCFail, err:
832
      val = str(err)
833
    result[constants.NV_INSTANCELIST] = val
834

    
835
  if constants.NV_VGLIST in what and vm_capable:
836
    result[constants.NV_VGLIST] = utils.ListVolumeGroups()
837

    
838
  if constants.NV_PVLIST in what and vm_capable:
839
    check_exclusive_pvs = constants.NV_EXCLUSIVEPVS in what
840
    val = bdev.LogicalVolume.GetPVInfo(what[constants.NV_PVLIST],
841
                                       filter_allocatable=False,
842
                                       include_lvs=check_exclusive_pvs)
843
    if check_exclusive_pvs:
844
      result[constants.NV_EXCLUSIVEPVS] = _CheckExclusivePvs(val)
845
      for pvi in val:
846
        # Avoid sending useless data on the wire
847
        pvi.lv_list = []
848
    result[constants.NV_PVLIST] = map(objects.LvmPvInfo.ToDict, val)
849

    
850
  if constants.NV_VERSION in what:
851
    result[constants.NV_VERSION] = (constants.PROTOCOL_VERSION,
852
                                    constants.RELEASE_VERSION)
853

    
854
  if constants.NV_HVINFO in what and vm_capable:
855
    hyper = hypervisor.GetHypervisor(what[constants.NV_HVINFO])
856
    result[constants.NV_HVINFO] = hyper.GetNodeInfo()
857

    
858
  if constants.NV_DRBDLIST in what and vm_capable:
859
    try:
860
      used_minors = bdev.DRBD8.GetUsedDevs().keys()
861
    except errors.BlockDeviceError, err:
862
      logging.warning("Can't get used minors list", exc_info=True)
863
      used_minors = str(err)
864
    result[constants.NV_DRBDLIST] = used_minors
865

    
866
  if constants.NV_DRBDHELPER in what and vm_capable:
867
    status = True
868
    try:
869
      payload = bdev.BaseDRBD.GetUsermodeHelper()
870
    except errors.BlockDeviceError, err:
871
      logging.error("Can't get DRBD usermode helper: %s", str(err))
872
      status = False
873
      payload = str(err)
874
    result[constants.NV_DRBDHELPER] = (status, payload)
875

    
876
  if constants.NV_NODESETUP in what:
877
    result[constants.NV_NODESETUP] = tmpr = []
878
    if not os.path.isdir("/sys/block") or not os.path.isdir("/sys/class/net"):
879
      tmpr.append("The sysfs filesytem doesn't seem to be mounted"
880
                  " under /sys, missing required directories /sys/block"
881
                  " and /sys/class/net")
882
    if (not os.path.isdir("/proc/sys") or
883
        not os.path.isfile("/proc/sysrq-trigger")):
884
      tmpr.append("The procfs filesystem doesn't seem to be mounted"
885
                  " under /proc, missing required directory /proc/sys and"
886
                  " the file /proc/sysrq-trigger")
887

    
888
  if constants.NV_TIME in what:
889
    result[constants.NV_TIME] = utils.SplitTime(time.time())
890

    
891
  if constants.NV_OSLIST in what and vm_capable:
892
    result[constants.NV_OSLIST] = DiagnoseOS()
893

    
894
  if constants.NV_BRIDGES in what and vm_capable:
895
    result[constants.NV_BRIDGES] = [bridge
896
                                    for bridge in what[constants.NV_BRIDGES]
897
                                    if not utils.BridgeExists(bridge)]
898

    
899
  if what.get(constants.NV_FILE_STORAGE_PATHS) == my_name:
900
    result[constants.NV_FILE_STORAGE_PATHS] = \
901
      bdev.ComputeWrongFileStoragePaths()
902

    
903
  return result
904

    
905

    
906
def GetBlockDevSizes(devices):
907
  """Return the size of the given block devices
908

909
  @type devices: list
910
  @param devices: list of block device nodes to query
911
  @rtype: dict
912
  @return:
913
    dictionary of all block devices under /dev (key). The value is their
914
    size in MiB.
915

916
    {'/dev/disk/by-uuid/123456-12321231-312312-312': 124}
917

918
  """
919
  DEV_PREFIX = "/dev/"
920
  blockdevs = {}
921

    
922
  for devpath in devices:
923
    if not utils.IsBelowDir(DEV_PREFIX, devpath):
924
      continue
925

    
926
    try:
927
      st = os.stat(devpath)
928
    except EnvironmentError, err:
929
      logging.warning("Error stat()'ing device %s: %s", devpath, str(err))
930
      continue
931

    
932
    if stat.S_ISBLK(st.st_mode):
933
      result = utils.RunCmd(["blockdev", "--getsize64", devpath])
934
      if result.failed:
935
        # We don't want to fail, just do not list this device as available
936
        logging.warning("Cannot get size for block device %s", devpath)
937
        continue
938

    
939
      size = int(result.stdout) / (1024 * 1024)
940
      blockdevs[devpath] = size
941
  return blockdevs
942

    
943

    
944
def GetVolumeList(vg_names):
945
  """Compute list of logical volumes and their size.
946

947
  @type vg_names: list
948
  @param vg_names: the volume groups whose LVs we should list, or
949
      empty for all volume groups
950
  @rtype: dict
951
  @return:
952
      dictionary of all partions (key) with value being a tuple of
953
      their size (in MiB), inactive and online status::
954

955
        {'xenvg/test1': ('20.06', True, True)}
956

957
      in case of errors, a string is returned with the error
958
      details.
959

960
  """
961
  lvs = {}
962
  sep = "|"
963
  if not vg_names:
964
    vg_names = []
965
  result = utils.RunCmd(["lvs", "--noheadings", "--units=m", "--nosuffix",
966
                         "--separator=%s" % sep,
967
                         "-ovg_name,lv_name,lv_size,lv_attr"] + vg_names)
968
  if result.failed:
969
    _Fail("Failed to list logical volumes, lvs output: %s", result.output)
970

    
971
  for line in result.stdout.splitlines():
972
    line = line.strip()
973
    match = _LVSLINE_REGEX.match(line)
974
    if not match:
975
      logging.error("Invalid line returned from lvs output: '%s'", line)
976
      continue
977
    vg_name, name, size, attr = match.groups()
978
    inactive = attr[4] == "-"
979
    online = attr[5] == "o"
980
    virtual = attr[0] == "v"
981
    if virtual:
982
      # we don't want to report such volumes as existing, since they
983
      # don't really hold data
984
      continue
985
    lvs[vg_name + "/" + name] = (size, inactive, online)
986

    
987
  return lvs
988

    
989

    
990
def ListVolumeGroups():
991
  """List the volume groups and their size.
992

993
  @rtype: dict
994
  @return: dictionary with keys volume name and values the
995
      size of the volume
996

997
  """
998
  return utils.ListVolumeGroups()
999

    
1000

    
1001
def NodeVolumes():
1002
  """List all volumes on this node.
1003

1004
  @rtype: list
1005
  @return:
1006
    A list of dictionaries, each having four keys:
1007
      - name: the logical volume name,
1008
      - size: the size of the logical volume
1009
      - dev: the physical device on which the LV lives
1010
      - vg: the volume group to which it belongs
1011

1012
    In case of errors, we return an empty list and log the
1013
    error.
1014

1015
    Note that since a logical volume can live on multiple physical
1016
    volumes, the resulting list might include a logical volume
1017
    multiple times.
1018

1019
  """
1020
  result = utils.RunCmd(["lvs", "--noheadings", "--units=m", "--nosuffix",
1021
                         "--separator=|",
1022
                         "--options=lv_name,lv_size,devices,vg_name"])
1023
  if result.failed:
1024
    _Fail("Failed to list logical volumes, lvs output: %s",
1025
          result.output)
1026

    
1027
  def parse_dev(dev):
1028
    return dev.split("(")[0]
1029

    
1030
  def handle_dev(dev):
1031
    return [parse_dev(x) for x in dev.split(",")]
1032

    
1033
  def map_line(line):
1034
    line = [v.strip() for v in line]
1035
    return [{"name": line[0], "size": line[1],
1036
             "dev": dev, "vg": line[3]} for dev in handle_dev(line[2])]
1037

    
1038
  all_devs = []
1039
  for line in result.stdout.splitlines():
1040
    if line.count("|") >= 3:
1041
      all_devs.extend(map_line(line.split("|")))
1042
    else:
1043
      logging.warning("Strange line in the output from lvs: '%s'", line)
1044
  return all_devs
1045

    
1046

    
1047
def BridgesExist(bridges_list):
1048
  """Check if a list of bridges exist on the current node.
1049

1050
  @rtype: boolean
1051
  @return: C{True} if all of them exist, C{False} otherwise
1052

1053
  """
1054
  missing = []
1055
  for bridge in bridges_list:
1056
    if not utils.BridgeExists(bridge):
1057
      missing.append(bridge)
1058

    
1059
  if missing:
1060
    _Fail("Missing bridges %s", utils.CommaJoin(missing))
1061

    
1062

    
1063
def GetInstanceList(hypervisor_list):
1064
  """Provides a list of instances.
1065

1066
  @type hypervisor_list: list
1067
  @param hypervisor_list: the list of hypervisors to query information
1068

1069
  @rtype: list
1070
  @return: a list of all running instances on the current node
1071
    - instance1.example.com
1072
    - instance2.example.com
1073

1074
  """
1075
  results = []
1076
  for hname in hypervisor_list:
1077
    try:
1078
      names = hypervisor.GetHypervisor(hname).ListInstances()
1079
      results.extend(names)
1080
    except errors.HypervisorError, err:
1081
      _Fail("Error enumerating instances (hypervisor %s): %s",
1082
            hname, err, exc=True)
1083

    
1084
  return results
1085

    
1086

    
1087
def GetInstanceInfo(instance, hname):
1088
  """Gives back the information about an instance as a dictionary.
1089

1090
  @type instance: string
1091
  @param instance: the instance name
1092
  @type hname: string
1093
  @param hname: the hypervisor type of the instance
1094

1095
  @rtype: dict
1096
  @return: dictionary with the following keys:
1097
      - memory: memory size of instance (int)
1098
      - state: xen state of instance (string)
1099
      - time: cpu time of instance (float)
1100
      - vcpus: the number of vcpus (int)
1101

1102
  """
1103
  output = {}
1104

    
1105
  iinfo = hypervisor.GetHypervisor(hname).GetInstanceInfo(instance)
1106
  if iinfo is not None:
1107
    output["memory"] = iinfo[2]
1108
    output["vcpus"] = iinfo[3]
1109
    output["state"] = iinfo[4]
1110
    output["time"] = iinfo[5]
1111

    
1112
  return output
1113

    
1114

    
1115
def GetInstanceMigratable(instance):
1116
  """Gives whether an instance can be migrated.
1117

1118
  @type instance: L{objects.Instance}
1119
  @param instance: object representing the instance to be checked.
1120

1121
  @rtype: tuple
1122
  @return: tuple of (result, description) where:
1123
      - result: whether the instance can be migrated or not
1124
      - description: a description of the issue, if relevant
1125

1126
  """
1127
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1128
  iname = instance.name
1129
  if iname not in hyper.ListInstances():
1130
    _Fail("Instance %s is not running", iname)
1131

    
1132
  for idx in range(len(instance.disks)):
1133
    link_name = _GetBlockDevSymlinkPath(iname, idx)
1134
    if not os.path.islink(link_name):
1135
      logging.warning("Instance %s is missing symlink %s for disk %d",
1136
                      iname, link_name, idx)
1137

    
1138

    
1139
def GetAllInstancesInfo(hypervisor_list):
1140
  """Gather data about all instances.
1141

1142
  This is the equivalent of L{GetInstanceInfo}, except that it
1143
  computes data for all instances at once, thus being faster if one
1144
  needs data about more than one instance.
1145

1146
  @type hypervisor_list: list
1147
  @param hypervisor_list: list of hypervisors to query for instance data
1148

1149
  @rtype: dict
1150
  @return: dictionary of instance: data, with data having the following keys:
1151
      - memory: memory size of instance (int)
1152
      - state: xen state of instance (string)
1153
      - time: cpu time of instance (float)
1154
      - vcpus: the number of vcpus
1155

1156
  """
1157
  output = {}
1158

    
1159
  for hname in hypervisor_list:
1160
    iinfo = hypervisor.GetHypervisor(hname).GetAllInstancesInfo()
1161
    if iinfo:
1162
      for name, _, memory, vcpus, state, times in iinfo:
1163
        value = {
1164
          "memory": memory,
1165
          "vcpus": vcpus,
1166
          "state": state,
1167
          "time": times,
1168
          }
1169
        if name in output:
1170
          # we only check static parameters, like memory and vcpus,
1171
          # and not state and time which can change between the
1172
          # invocations of the different hypervisors
1173
          for key in "memory", "vcpus":
1174
            if value[key] != output[name][key]:
1175
              _Fail("Instance %s is running twice"
1176
                    " with different parameters", name)
1177
        output[name] = value
1178

    
1179
  return output
1180

    
1181

    
1182
def _InstanceLogName(kind, os_name, instance, component):
1183
  """Compute the OS log filename for a given instance and operation.
1184

1185
  The instance name and os name are passed in as strings since not all
1186
  operations have these as part of an instance object.
1187

1188
  @type kind: string
1189
  @param kind: the operation type (e.g. add, import, etc.)
1190
  @type os_name: string
1191
  @param os_name: the os name
1192
  @type instance: string
1193
  @param instance: the name of the instance being imported/added/etc.
1194
  @type component: string or None
1195
  @param component: the name of the component of the instance being
1196
      transferred
1197

1198
  """
1199
  # TODO: Use tempfile.mkstemp to create unique filename
1200
  if component:
1201
    assert "/" not in component
1202
    c_msg = "-%s" % component
1203
  else:
1204
    c_msg = ""
1205
  base = ("%s-%s-%s%s-%s.log" %
1206
          (kind, os_name, instance, c_msg, utils.TimestampForFilename()))
1207
  return utils.PathJoin(pathutils.LOG_OS_DIR, base)
1208

    
1209

    
1210
def InstanceOsAdd(instance, reinstall, debug):
1211
  """Add an OS to an instance.
1212

1213
  @type instance: L{objects.Instance}
1214
  @param instance: Instance whose OS is to be installed
1215
  @type reinstall: boolean
1216
  @param reinstall: whether this is an instance reinstall
1217
  @type debug: integer
1218
  @param debug: debug level, passed to the OS scripts
1219
  @rtype: None
1220

1221
  """
1222
  inst_os = OSFromDisk(instance.os)
1223

    
1224
  create_env = OSEnvironment(instance, inst_os, debug)
1225
  if reinstall:
1226
    create_env["INSTANCE_REINSTALL"] = "1"
1227

    
1228
  logfile = _InstanceLogName("add", instance.os, instance.name, None)
1229

    
1230
  result = utils.RunCmd([inst_os.create_script], env=create_env,
1231
                        cwd=inst_os.path, output=logfile, reset_env=True)
1232
  if result.failed:
1233
    logging.error("os create command '%s' returned error: %s, logfile: %s,"
1234
                  " output: %s", result.cmd, result.fail_reason, logfile,
1235
                  result.output)
1236
    lines = [utils.SafeEncode(val)
1237
             for val in utils.TailFile(logfile, lines=20)]
1238
    _Fail("OS create script failed (%s), last lines in the"
1239
          " log file:\n%s", result.fail_reason, "\n".join(lines), log=False)
1240

    
1241

    
1242
def RunRenameInstance(instance, old_name, debug):
1243
  """Run the OS rename script for an instance.
1244

1245
  @type instance: L{objects.Instance}
1246
  @param instance: Instance whose OS is to be installed
1247
  @type old_name: string
1248
  @param old_name: previous instance name
1249
  @type debug: integer
1250
  @param debug: debug level, passed to the OS scripts
1251
  @rtype: boolean
1252
  @return: the success of the operation
1253

1254
  """
1255
  inst_os = OSFromDisk(instance.os)
1256

    
1257
  rename_env = OSEnvironment(instance, inst_os, debug)
1258
  rename_env["OLD_INSTANCE_NAME"] = old_name
1259

    
1260
  logfile = _InstanceLogName("rename", instance.os,
1261
                             "%s-%s" % (old_name, instance.name), None)
1262

    
1263
  result = utils.RunCmd([inst_os.rename_script], env=rename_env,
1264
                        cwd=inst_os.path, output=logfile, reset_env=True)
1265

    
1266
  if result.failed:
1267
    logging.error("os create command '%s' returned error: %s output: %s",
1268
                  result.cmd, result.fail_reason, result.output)
1269
    lines = [utils.SafeEncode(val)
1270
             for val in utils.TailFile(logfile, lines=20)]
1271
    _Fail("OS rename script failed (%s), last lines in the"
1272
          " log file:\n%s", result.fail_reason, "\n".join(lines), log=False)
1273

    
1274

    
1275
def _GetBlockDevSymlinkPath(instance_name, idx, _dir=None):
1276
  """Returns symlink path for block device.
1277

1278
  """
1279
  if _dir is None:
1280
    _dir = pathutils.DISK_LINKS_DIR
1281

    
1282
  return utils.PathJoin(_dir,
1283
                        ("%s%s%s" %
1284
                         (instance_name, constants.DISK_SEPARATOR, idx)))
1285

    
1286

    
1287
def _SymlinkBlockDev(instance_name, device_path, idx):
1288
  """Set up symlinks to a instance's block device.
1289

1290
  This is an auxiliary function run when an instance is start (on the primary
1291
  node) or when an instance is migrated (on the target node).
1292

1293

1294
  @param instance_name: the name of the target instance
1295
  @param device_path: path of the physical block device, on the node
1296
  @param idx: the disk index
1297
  @return: absolute path to the disk's symlink
1298

1299
  """
1300
  link_name = _GetBlockDevSymlinkPath(instance_name, idx)
1301
  try:
1302
    os.symlink(device_path, link_name)
1303
  except OSError, err:
1304
    if err.errno == errno.EEXIST:
1305
      if (not os.path.islink(link_name) or
1306
          os.readlink(link_name) != device_path):
1307
        os.remove(link_name)
1308
        os.symlink(device_path, link_name)
1309
    else:
1310
      raise
1311

    
1312
  return link_name
1313

    
1314

    
1315
def _RemoveBlockDevLinks(instance_name, disks):
1316
  """Remove the block device symlinks belonging to the given instance.
1317

1318
  """
1319
  for idx, _ in enumerate(disks):
1320
    link_name = _GetBlockDevSymlinkPath(instance_name, idx)
1321
    if os.path.islink(link_name):
1322
      try:
1323
        os.remove(link_name)
1324
      except OSError:
1325
        logging.exception("Can't remove symlink '%s'", link_name)
1326

    
1327

    
1328
def _GatherAndLinkBlockDevs(instance):
1329
  """Set up an instance's block device(s).
1330

1331
  This is run on the primary node at instance startup. The block
1332
  devices must be already assembled.
1333

1334
  @type instance: L{objects.Instance}
1335
  @param instance: the instance whose disks we shoul assemble
1336
  @rtype: list
1337
  @return: list of (disk_object, device_path)
1338

1339
  """
1340
  block_devices = []
1341
  for idx, disk in enumerate(instance.disks):
1342
    device = _RecursiveFindBD(disk)
1343
    if device is None:
1344
      raise errors.BlockDeviceError("Block device '%s' is not set up." %
1345
                                    str(disk))
1346
    device.Open()
1347
    try:
1348
      link_name = _SymlinkBlockDev(instance.name, device.dev_path, idx)
1349
    except OSError, e:
1350
      raise errors.BlockDeviceError("Cannot create block device symlink: %s" %
1351
                                    e.strerror)
1352

    
1353
    block_devices.append((disk, link_name))
1354

    
1355
  return block_devices
1356

    
1357

    
1358
def StartInstance(instance, startup_paused):
1359
  """Start an instance.
1360

1361
  @type instance: L{objects.Instance}
1362
  @param instance: the instance object
1363
  @type startup_paused: bool
1364
  @param instance: pause instance at startup?
1365
  @rtype: None
1366

1367
  """
1368
  running_instances = GetInstanceList([instance.hypervisor])
1369

    
1370
  if instance.name in running_instances:
1371
    logging.info("Instance %s already running, not starting", instance.name)
1372
    return
1373

    
1374
  try:
1375
    block_devices = _GatherAndLinkBlockDevs(instance)
1376
    hyper = hypervisor.GetHypervisor(instance.hypervisor)
1377
    hyper.StartInstance(instance, block_devices, startup_paused)
1378
  except errors.BlockDeviceError, err:
1379
    _Fail("Block device error: %s", err, exc=True)
1380
  except errors.HypervisorError, err:
1381
    _RemoveBlockDevLinks(instance.name, instance.disks)
1382
    _Fail("Hypervisor error: %s", err, exc=True)
1383

    
1384

    
1385
def InstanceShutdown(instance, timeout):
1386
  """Shut an instance down.
1387

1388
  @note: this functions uses polling with a hardcoded timeout.
1389

1390
  @type instance: L{objects.Instance}
1391
  @param instance: the instance object
1392
  @type timeout: integer
1393
  @param timeout: maximum timeout for soft shutdown
1394
  @rtype: None
1395

1396
  """
1397
  hv_name = instance.hypervisor
1398
  hyper = hypervisor.GetHypervisor(hv_name)
1399
  iname = instance.name
1400

    
1401
  if instance.name not in hyper.ListInstances():
1402
    logging.info("Instance %s not running, doing nothing", iname)
1403
    return
1404

    
1405
  class _TryShutdown:
1406
    def __init__(self):
1407
      self.tried_once = False
1408

    
1409
    def __call__(self):
1410
      if iname not in hyper.ListInstances():
1411
        return
1412

    
1413
      try:
1414
        hyper.StopInstance(instance, retry=self.tried_once)
1415
      except errors.HypervisorError, err:
1416
        if iname not in hyper.ListInstances():
1417
          # if the instance is no longer existing, consider this a
1418
          # success and go to cleanup
1419
          return
1420

    
1421
        _Fail("Failed to stop instance %s: %s", iname, err)
1422

    
1423
      self.tried_once = True
1424

    
1425
      raise utils.RetryAgain()
1426

    
1427
  try:
1428
    utils.Retry(_TryShutdown(), 5, timeout)
1429
  except utils.RetryTimeout:
1430
    # the shutdown did not succeed
1431
    logging.error("Shutdown of '%s' unsuccessful, forcing", iname)
1432

    
1433
    try:
1434
      hyper.StopInstance(instance, force=True)
1435
    except errors.HypervisorError, err:
1436
      if iname in hyper.ListInstances():
1437
        # only raise an error if the instance still exists, otherwise
1438
        # the error could simply be "instance ... unknown"!
1439
        _Fail("Failed to force stop instance %s: %s", iname, err)
1440

    
1441
    time.sleep(1)
1442

    
1443
    if iname in hyper.ListInstances():
1444
      _Fail("Could not shutdown instance %s even by destroy", iname)
1445

    
1446
  try:
1447
    hyper.CleanupInstance(instance.name)
1448
  except errors.HypervisorError, err:
1449
    logging.warning("Failed to execute post-shutdown cleanup step: %s", err)
1450

    
1451
  _RemoveBlockDevLinks(iname, instance.disks)
1452

    
1453

    
1454
def InstanceReboot(instance, reboot_type, shutdown_timeout):
1455
  """Reboot an instance.
1456

1457
  @type instance: L{objects.Instance}
1458
  @param instance: the instance object to reboot
1459
  @type reboot_type: str
1460
  @param reboot_type: the type of reboot, one the following
1461
    constants:
1462
      - L{constants.INSTANCE_REBOOT_SOFT}: only reboot the
1463
        instance OS, do not recreate the VM
1464
      - L{constants.INSTANCE_REBOOT_HARD}: tear down and
1465
        restart the VM (at the hypervisor level)
1466
      - the other reboot type (L{constants.INSTANCE_REBOOT_FULL}) is
1467
        not accepted here, since that mode is handled differently, in
1468
        cmdlib, and translates into full stop and start of the
1469
        instance (instead of a call_instance_reboot RPC)
1470
  @type shutdown_timeout: integer
1471
  @param shutdown_timeout: maximum timeout for soft shutdown
1472
  @rtype: None
1473

1474
  """
1475
  running_instances = GetInstanceList([instance.hypervisor])
1476

    
1477
  if instance.name not in running_instances:
1478
    _Fail("Cannot reboot instance %s that is not running", instance.name)
1479

    
1480
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1481
  if reboot_type == constants.INSTANCE_REBOOT_SOFT:
1482
    try:
1483
      hyper.RebootInstance(instance)
1484
    except errors.HypervisorError, err:
1485
      _Fail("Failed to soft reboot instance %s: %s", instance.name, err)
1486
  elif reboot_type == constants.INSTANCE_REBOOT_HARD:
1487
    try:
1488
      InstanceShutdown(instance, shutdown_timeout)
1489
      return StartInstance(instance, False)
1490
    except errors.HypervisorError, err:
1491
      _Fail("Failed to hard reboot instance %s: %s", instance.name, err)
1492
  else:
1493
    _Fail("Invalid reboot_type received: %s", reboot_type)
1494

    
1495

    
1496
def InstanceBalloonMemory(instance, memory):
1497
  """Resize an instance's memory.
1498

1499
  @type instance: L{objects.Instance}
1500
  @param instance: the instance object
1501
  @type memory: int
1502
  @param memory: new memory amount in MB
1503
  @rtype: None
1504

1505
  """
1506
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1507
  running = hyper.ListInstances()
1508
  if instance.name not in running:
1509
    logging.info("Instance %s is not running, cannot balloon", instance.name)
1510
    return
1511
  try:
1512
    hyper.BalloonInstanceMemory(instance, memory)
1513
  except errors.HypervisorError, err:
1514
    _Fail("Failed to balloon instance memory: %s", err, exc=True)
1515

    
1516

    
1517
def MigrationInfo(instance):
1518
  """Gather information about an instance to be migrated.
1519

1520
  @type instance: L{objects.Instance}
1521
  @param instance: the instance definition
1522

1523
  """
1524
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1525
  try:
1526
    info = hyper.MigrationInfo(instance)
1527
  except errors.HypervisorError, err:
1528
    _Fail("Failed to fetch migration information: %s", err, exc=True)
1529
  return info
1530

    
1531

    
1532
def AcceptInstance(instance, info, target):
1533
  """Prepare the node to accept an instance.
1534

1535
  @type instance: L{objects.Instance}
1536
  @param instance: the instance definition
1537
  @type info: string/data (opaque)
1538
  @param info: migration information, from the source node
1539
  @type target: string
1540
  @param target: target host (usually ip), on this node
1541

1542
  """
1543
  # TODO: why is this required only for DTS_EXT_MIRROR?
1544
  if instance.disk_template in constants.DTS_EXT_MIRROR:
1545
    # Create the symlinks, as the disks are not active
1546
    # in any way
1547
    try:
1548
      _GatherAndLinkBlockDevs(instance)
1549
    except errors.BlockDeviceError, err:
1550
      _Fail("Block device error: %s", err, exc=True)
1551

    
1552
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1553
  try:
1554
    hyper.AcceptInstance(instance, info, target)
1555
  except errors.HypervisorError, err:
1556
    if instance.disk_template in constants.DTS_EXT_MIRROR:
1557
      _RemoveBlockDevLinks(instance.name, instance.disks)
1558
    _Fail("Failed to accept instance: %s", err, exc=True)
1559

    
1560

    
1561
def FinalizeMigrationDst(instance, info, success):
1562
  """Finalize any preparation to accept an instance.
1563

1564
  @type instance: L{objects.Instance}
1565
  @param instance: the instance definition
1566
  @type info: string/data (opaque)
1567
  @param info: migration information, from the source node
1568
  @type success: boolean
1569
  @param success: whether the migration was a success or a failure
1570

1571
  """
1572
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1573
  try:
1574
    hyper.FinalizeMigrationDst(instance, info, success)
1575
  except errors.HypervisorError, err:
1576
    _Fail("Failed to finalize migration on the target node: %s", err, exc=True)
1577

    
1578

    
1579
def MigrateInstance(instance, target, live):
1580
  """Migrates an instance to another node.
1581

1582
  @type instance: L{objects.Instance}
1583
  @param instance: the instance definition
1584
  @type target: string
1585
  @param target: the target node name
1586
  @type live: boolean
1587
  @param live: whether the migration should be done live or not (the
1588
      interpretation of this parameter is left to the hypervisor)
1589
  @raise RPCFail: if migration fails for some reason
1590

1591
  """
1592
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1593

    
1594
  try:
1595
    hyper.MigrateInstance(instance, target, live)
1596
  except errors.HypervisorError, err:
1597
    _Fail("Failed to migrate instance: %s", err, exc=True)
1598

    
1599

    
1600
def FinalizeMigrationSource(instance, success, live):
1601
  """Finalize the instance migration on the source node.
1602

1603
  @type instance: L{objects.Instance}
1604
  @param instance: the instance definition of the migrated instance
1605
  @type success: bool
1606
  @param success: whether the migration succeeded or not
1607
  @type live: bool
1608
  @param live: whether the user requested a live migration or not
1609
  @raise RPCFail: If the execution fails for some reason
1610

1611
  """
1612
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1613

    
1614
  try:
1615
    hyper.FinalizeMigrationSource(instance, success, live)
1616
  except Exception, err:  # pylint: disable=W0703
1617
    _Fail("Failed to finalize the migration on the source node: %s", err,
1618
          exc=True)
1619

    
1620

    
1621
def GetMigrationStatus(instance):
1622
  """Get the migration status
1623

1624
  @type instance: L{objects.Instance}
1625
  @param instance: the instance that is being migrated
1626
  @rtype: L{objects.MigrationStatus}
1627
  @return: the status of the current migration (one of
1628
           L{constants.HV_MIGRATION_VALID_STATUSES}), plus any additional
1629
           progress info that can be retrieved from the hypervisor
1630
  @raise RPCFail: If the migration status cannot be retrieved
1631

1632
  """
1633
  hyper = hypervisor.GetHypervisor(instance.hypervisor)
1634
  try:
1635
    return hyper.GetMigrationStatus(instance)
1636
  except Exception, err:  # pylint: disable=W0703
1637
    _Fail("Failed to get migration status: %s", err, exc=True)
1638

    
1639

    
1640
def BlockdevCreate(disk, size, owner, on_primary, info, excl_stor):
1641
  """Creates a block device for an instance.
1642

1643
  @type disk: L{objects.Disk}
1644
  @param disk: the object describing the disk we should create
1645
  @type size: int
1646
  @param size: the size of the physical underlying device, in MiB
1647
  @type owner: str
1648
  @param owner: the name of the instance for which disk is created,
1649
      used for device cache data
1650
  @type on_primary: boolean
1651
  @param on_primary:  indicates if it is the primary node or not
1652
  @type info: string
1653
  @param info: string that will be sent to the physical device
1654
      creation, used for example to set (LVM) tags on LVs
1655
  @type excl_stor: boolean
1656
  @param excl_stor: Whether exclusive_storage is active
1657

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

1662
  """
1663
  # TODO: remove the obsolete "size" argument
1664
  # pylint: disable=W0613
1665
  clist = []
1666
  if disk.children:
1667
    for child in disk.children:
1668
      try:
1669
        crdev = _RecursiveAssembleBD(child, owner, on_primary)
1670
      except errors.BlockDeviceError, err:
1671
        _Fail("Can't assemble device %s: %s", child, err)
1672
      if on_primary or disk.AssembleOnSecondary():
1673
        # we need the children open in case the device itself has to
1674
        # be assembled
1675
        try:
1676
          # pylint: disable=E1103
1677
          crdev.Open()
1678
        except errors.BlockDeviceError, err:
1679
          _Fail("Can't make child '%s' read-write: %s", child, err)
1680
      clist.append(crdev)
1681

    
1682
  try:
1683
    device = bdev.Create(disk, clist, excl_stor)
1684
  except errors.BlockDeviceError, err:
1685
    _Fail("Can't create block device: %s", err)
1686

    
1687
  if on_primary or disk.AssembleOnSecondary():
1688
    try:
1689
      device.Assemble()
1690
    except errors.BlockDeviceError, err:
1691
      _Fail("Can't assemble device after creation, unusual event: %s", err)
1692
    if on_primary or disk.OpenOnSecondary():
1693
      try:
1694
        device.Open(force=True)
1695
      except errors.BlockDeviceError, err:
1696
        _Fail("Can't make device r/w after creation, unusual event: %s", err)
1697
    DevCacheManager.UpdateCache(device.dev_path, owner,
1698
                                on_primary, disk.iv_name)
1699

    
1700
  device.SetInfo(info)
1701

    
1702
  return device.unique_id
1703

    
1704

    
1705
def _WipeDevice(path, offset, size):
1706
  """This function actually wipes the device.
1707

1708
  @param path: The path to the device to wipe
1709
  @param offset: The offset in MiB in the file
1710
  @param size: The size in MiB to write
1711

1712
  """
1713
  # Internal sizes are always in Mebibytes; if the following "dd" command
1714
  # should use a different block size the offset and size given to this
1715
  # function must be adjusted accordingly before being passed to "dd".
1716
  block_size = 1024 * 1024
1717

    
1718
  cmd = [constants.DD_CMD, "if=/dev/zero", "seek=%d" % offset,
1719
         "bs=%s" % block_size, "oflag=direct", "of=%s" % path,
1720
         "count=%d" % size]
1721
  result = utils.RunCmd(cmd)
1722

    
1723
  if result.failed:
1724
    _Fail("Wipe command '%s' exited with error: %s; output: %s", result.cmd,
1725
          result.fail_reason, result.output)
1726

    
1727

    
1728
def BlockdevWipe(disk, offset, size):
1729
  """Wipes a block device.
1730

1731
  @type disk: L{objects.Disk}
1732
  @param disk: the disk object we want to wipe
1733
  @type offset: int
1734
  @param offset: The offset in MiB in the file
1735
  @type size: int
1736
  @param size: The size in MiB to write
1737

1738
  """
1739
  try:
1740
    rdev = _RecursiveFindBD(disk)
1741
  except errors.BlockDeviceError:
1742
    rdev = None
1743

    
1744
  if not rdev:
1745
    _Fail("Cannot execute wipe for device %s: device not found", disk.iv_name)
1746

    
1747
  # Do cross verify some of the parameters
1748
  if offset < 0:
1749
    _Fail("Negative offset")
1750
  if size < 0:
1751
    _Fail("Negative size")
1752
  if offset > rdev.size:
1753
    _Fail("Offset is bigger than device size")
1754
  if (offset + size) > rdev.size:
1755
    _Fail("The provided offset and size to wipe is bigger than device size")
1756

    
1757
  _WipeDevice(rdev.dev_path, offset, size)
1758

    
1759

    
1760
def BlockdevPauseResumeSync(disks, pause):
1761
  """Pause or resume the sync of the block device.
1762

1763
  @type disks: list of L{objects.Disk}
1764
  @param disks: the disks object we want to pause/resume
1765
  @type pause: bool
1766
  @param pause: Wheater to pause or resume
1767

1768
  """
1769
  success = []
1770
  for disk in disks:
1771
    try:
1772
      rdev = _RecursiveFindBD(disk)
1773
    except errors.BlockDeviceError:
1774
      rdev = None
1775

    
1776
    if not rdev:
1777
      success.append((False, ("Cannot change sync for device %s:"
1778
                              " device not found" % disk.iv_name)))
1779
      continue
1780

    
1781
    result = rdev.PauseResumeSync(pause)
1782

    
1783
    if result:
1784
      success.append((result, None))
1785
    else:
1786
      if pause:
1787
        msg = "Pause"
1788
      else:
1789
        msg = "Resume"
1790
      success.append((result, "%s for device %s failed" % (msg, disk.iv_name)))
1791

    
1792
  return success
1793

    
1794

    
1795
def BlockdevRemove(disk):
1796
  """Remove a block device.
1797

1798
  @note: This is intended to be called recursively.
1799

1800
  @type disk: L{objects.Disk}
1801
  @param disk: the disk object we should remove
1802
  @rtype: boolean
1803
  @return: the success of the operation
1804

1805
  """
1806
  msgs = []
1807
  try:
1808
    rdev = _RecursiveFindBD(disk)
1809
  except errors.BlockDeviceError, err:
1810
    # probably can't attach
1811
    logging.info("Can't attach to device %s in remove", disk)
1812
    rdev = None
1813
  if rdev is not None:
1814
    r_path = rdev.dev_path
1815
    try:
1816
      rdev.Remove()
1817
    except errors.BlockDeviceError, err:
1818
      msgs.append(str(err))
1819
    if not msgs:
1820
      DevCacheManager.RemoveCache(r_path)
1821

    
1822
  if disk.children:
1823
    for child in disk.children:
1824
      try:
1825
        BlockdevRemove(child)
1826
      except RPCFail, err:
1827
        msgs.append(str(err))
1828

    
1829
  if msgs:
1830
    _Fail("; ".join(msgs))
1831

    
1832

    
1833
def _RecursiveAssembleBD(disk, owner, as_primary):
1834
  """Activate a block device for an instance.
1835

1836
  This is run on the primary and secondary nodes for an instance.
1837

1838
  @note: this function is called recursively.
1839

1840
  @type disk: L{objects.Disk}
1841
  @param disk: the disk we try to assemble
1842
  @type owner: str
1843
  @param owner: the name of the instance which owns the disk
1844
  @type as_primary: boolean
1845
  @param as_primary: if we should make the block device
1846
      read/write
1847

1848
  @return: the assembled device or None (in case no device
1849
      was assembled)
1850
  @raise errors.BlockDeviceError: in case there is an error
1851
      during the activation of the children or the device
1852
      itself
1853

1854
  """
1855
  children = []
1856
  if disk.children:
1857
    mcn = disk.ChildrenNeeded()
1858
    if mcn == -1:
1859
      mcn = 0 # max number of Nones allowed
1860
    else:
1861
      mcn = len(disk.children) - mcn # max number of Nones
1862
    for chld_disk in disk.children:
1863
      try:
1864
        cdev = _RecursiveAssembleBD(chld_disk, owner, as_primary)
1865
      except errors.BlockDeviceError, err:
1866
        if children.count(None) >= mcn:
1867
          raise
1868
        cdev = None
1869
        logging.error("Error in child activation (but continuing): %s",
1870
                      str(err))
1871
      children.append(cdev)
1872

    
1873
  if as_primary or disk.AssembleOnSecondary():
1874
    r_dev = bdev.Assemble(disk, children)
1875
    result = r_dev
1876
    if as_primary or disk.OpenOnSecondary():
1877
      r_dev.Open()
1878
    DevCacheManager.UpdateCache(r_dev.dev_path, owner,
1879
                                as_primary, disk.iv_name)
1880

    
1881
  else:
1882
    result = True
1883
  return result
1884

    
1885

    
1886
def BlockdevAssemble(disk, owner, as_primary, idx):
1887
  """Activate a block device for an instance.
1888

1889
  This is a wrapper over _RecursiveAssembleBD.
1890

1891
  @rtype: str or boolean
1892
  @return: a C{/dev/...} path for primary nodes, and
1893
      C{True} for secondary nodes
1894

1895
  """
1896
  try:
1897
    result = _RecursiveAssembleBD(disk, owner, as_primary)
1898
    if isinstance(result, bdev.BlockDev):
1899
      # pylint: disable=E1103
1900
      result = result.dev_path
1901
      if as_primary:
1902
        _SymlinkBlockDev(owner, result, idx)
1903
  except errors.BlockDeviceError, err:
1904
    _Fail("Error while assembling disk: %s", err, exc=True)
1905
  except OSError, err:
1906
    _Fail("Error while symlinking disk: %s", err, exc=True)
1907

    
1908
  return result
1909

    
1910

    
1911
def BlockdevShutdown(disk):
1912
  """Shut down a block device.
1913

1914
  First, if the device is assembled (Attach() is successful), then
1915
  the device is shutdown. Then the children of the device are
1916
  shutdown.
1917

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

1922
  @type disk: L{objects.Disk}
1923
  @param disk: the description of the disk we should
1924
      shutdown
1925
  @rtype: None
1926

1927
  """
1928
  msgs = []
1929
  r_dev = _RecursiveFindBD(disk)
1930
  if r_dev is not None:
1931
    r_path = r_dev.dev_path
1932
    try:
1933
      r_dev.Shutdown()
1934
      DevCacheManager.RemoveCache(r_path)
1935
    except errors.BlockDeviceError, err:
1936
      msgs.append(str(err))
1937

    
1938
  if disk.children:
1939
    for child in disk.children:
1940
      try:
1941
        BlockdevShutdown(child)
1942
      except RPCFail, err:
1943
        msgs.append(str(err))
1944

    
1945
  if msgs:
1946
    _Fail("; ".join(msgs))
1947

    
1948

    
1949
def BlockdevAddchildren(parent_cdev, new_cdevs):
1950
  """Extend a mirrored block device.
1951

1952
  @type parent_cdev: L{objects.Disk}
1953
  @param parent_cdev: the disk to which we should add children
1954
  @type new_cdevs: list of L{objects.Disk}
1955
  @param new_cdevs: the list of children which we should add
1956
  @rtype: None
1957

1958
  """
1959
  parent_bdev = _RecursiveFindBD(parent_cdev)
1960
  if parent_bdev is None:
1961
    _Fail("Can't find parent device '%s' in add children", parent_cdev)
1962
  new_bdevs = [_RecursiveFindBD(disk) for disk in new_cdevs]
1963
  if new_bdevs.count(None) > 0:
1964
    _Fail("Can't find new device(s) to add: %s:%s", new_bdevs, new_cdevs)
1965
  parent_bdev.AddChildren(new_bdevs)
1966

    
1967

    
1968
def BlockdevRemovechildren(parent_cdev, new_cdevs):
1969
  """Shrink a mirrored block device.
1970

1971
  @type parent_cdev: L{objects.Disk}
1972
  @param parent_cdev: the disk from which we should remove children
1973
  @type new_cdevs: list of L{objects.Disk}
1974
  @param new_cdevs: the list of children which we should remove
1975
  @rtype: None
1976

1977
  """
1978
  parent_bdev = _RecursiveFindBD(parent_cdev)
1979
  if parent_bdev is None:
1980
    _Fail("Can't find parent device '%s' in remove children", parent_cdev)
1981
  devs = []
1982
  for disk in new_cdevs:
1983
    rpath = disk.StaticDevPath()
1984
    if rpath is None:
1985
      bd = _RecursiveFindBD(disk)
1986
      if bd is None:
1987
        _Fail("Can't find device %s while removing children", disk)
1988
      else:
1989
        devs.append(bd.dev_path)
1990
    else:
1991
      if not utils.IsNormAbsPath(rpath):
1992
        _Fail("Strange path returned from StaticDevPath: '%s'", rpath)
1993
      devs.append(rpath)
1994
  parent_bdev.RemoveChildren(devs)
1995

    
1996

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

2000
  @type disks: list of L{objects.Disk}
2001
  @param disks: the list of disks which we should query
2002
  @rtype: disk
2003
  @return: List of L{objects.BlockDevStatus}, one for each disk
2004
  @raise errors.BlockDeviceError: if any of the disks cannot be
2005
      found
2006

2007
  """
2008
  stats = []
2009
  for dsk in disks:
2010
    rbd = _RecursiveFindBD(dsk)
2011
    if rbd is None:
2012
      _Fail("Can't find device %s", dsk)
2013

    
2014
    stats.append(rbd.CombinedSyncStatus())
2015

    
2016
  return stats
2017

    
2018

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

2022
  @type disks: list of L{objects.Disk}
2023
  @param disks: the list of disks which we should query
2024
  @rtype: disk
2025
  @return: List of tuples, (bool, status), one for each disk; bool denotes
2026
    success/failure, status is L{objects.BlockDevStatus} on success, string
2027
    otherwise
2028

2029
  """
2030
  result = []
2031
  for disk in disks:
2032
    try:
2033
      rbd = _RecursiveFindBD(disk)
2034
      if rbd is None:
2035
        result.append((False, "Can't find device %s" % disk))
2036
        continue
2037

    
2038
      status = rbd.CombinedSyncStatus()
2039
    except errors.BlockDeviceError, err:
2040
      logging.exception("Error while getting disk status")
2041
      result.append((False, str(err)))
2042
    else:
2043
      result.append((True, status))
2044

    
2045
  assert len(disks) == len(result)
2046

    
2047
  return result
2048

    
2049

    
2050
def _RecursiveFindBD(disk):
2051
  """Check if a device is activated.
2052

2053
  If so, return information about the real device.
2054

2055
  @type disk: L{objects.Disk}
2056
  @param disk: the disk object we need to find
2057

2058
  @return: None if the device can't be found,
2059
      otherwise the device instance
2060

2061
  """
2062
  children = []
2063
  if disk.children:
2064
    for chdisk in disk.children:
2065
      children.append(_RecursiveFindBD(chdisk))
2066

    
2067
  return bdev.FindDevice(disk, children)
2068

    
2069

    
2070
def _OpenRealBD(disk):
2071
  """Opens the underlying block device of a disk.
2072

2073
  @type disk: L{objects.Disk}
2074
  @param disk: the disk object we want to open
2075

2076
  """
2077
  real_disk = _RecursiveFindBD(disk)
2078
  if real_disk is None:
2079
    _Fail("Block device '%s' is not set up", disk)
2080

    
2081
  real_disk.Open()
2082

    
2083
  return real_disk
2084

    
2085

    
2086
def BlockdevFind(disk):
2087
  """Check if a device is activated.
2088

2089
  If it is, return information about the real device.
2090

2091
  @type disk: L{objects.Disk}
2092
  @param disk: the disk to find
2093
  @rtype: None or objects.BlockDevStatus
2094
  @return: None if the disk cannot be found, otherwise a the current
2095
           information
2096

2097
  """
2098
  try:
2099
    rbd = _RecursiveFindBD(disk)
2100
  except errors.BlockDeviceError, err:
2101
    _Fail("Failed to find device: %s", err, exc=True)
2102

    
2103
  if rbd is None:
2104
    return None
2105

    
2106
  return rbd.GetSyncStatus()
2107

    
2108

    
2109
def BlockdevGetsize(disks):
2110
  """Computes the size of the given disks.
2111

2112
  If a disk is not found, returns None instead.
2113

2114
  @type disks: list of L{objects.Disk}
2115
  @param disks: the list of disk to compute the size for
2116
  @rtype: list
2117
  @return: list with elements None if the disk cannot be found,
2118
      otherwise the size
2119

2120
  """
2121
  result = []
2122
  for cf in disks:
2123
    try:
2124
      rbd = _RecursiveFindBD(cf)
2125
    except errors.BlockDeviceError:
2126
      result.append(None)
2127
      continue
2128
    if rbd is None:
2129
      result.append(None)
2130
    else:
2131
      result.append(rbd.GetActualSize())
2132
  return result
2133

    
2134

    
2135
def BlockdevExport(disk, dest_node, dest_path, cluster_name):
2136
  """Export a block device to a remote node.
2137

2138
  @type disk: L{objects.Disk}
2139
  @param disk: the description of the disk to export
2140
  @type dest_node: str
2141
  @param dest_node: the destination node to export to
2142
  @type dest_path: str
2143
  @param dest_path: the destination path on the target node
2144
  @type cluster_name: str
2145
  @param cluster_name: the cluster name, needed for SSH hostalias
2146
  @rtype: None
2147

2148
  """
2149
  real_disk = _OpenRealBD(disk)
2150

    
2151
  # the block size on the read dd is 1MiB to match our units
2152
  expcmd = utils.BuildShellCmd("set -e; set -o pipefail; "
2153
                               "dd if=%s bs=1048576 count=%s",
2154
                               real_disk.dev_path, str(disk.size))
2155

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

    
2165
  remotecmd = _GetSshRunner(cluster_name).BuildCmd(dest_node,
2166
                                                   constants.SSH_LOGIN_USER,
2167
                                                   destcmd)
2168

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

    
2172
  result = utils.RunCmd(["bash", "-c", command])
2173

    
2174
  if result.failed:
2175
    _Fail("Disk copy command '%s' returned error: %s"
2176
          " output: %s", command, result.fail_reason, result.output)
2177

    
2178

    
2179
def UploadFile(file_name, data, mode, uid, gid, atime, mtime):
2180
  """Write a file to the filesystem.
2181

2182
  This allows the master to overwrite(!) a file. It will only perform
2183
  the operation if the file belongs to a list of configuration files.
2184

2185
  @type file_name: str
2186
  @param file_name: the target file name
2187
  @type data: str
2188
  @param data: the new contents of the file
2189
  @type mode: int
2190
  @param mode: the mode to give the file (can be None)
2191
  @type uid: string
2192
  @param uid: the owner of the file
2193
  @type gid: string
2194
  @param gid: the group of the file
2195
  @type atime: float
2196
  @param atime: the atime to set on the file (can be None)
2197
  @type mtime: float
2198
  @param mtime: the mtime to set on the file (can be None)
2199
  @rtype: None
2200

2201
  """
2202
  file_name = vcluster.LocalizeVirtualPath(file_name)
2203

    
2204
  if not os.path.isabs(file_name):
2205
    _Fail("Filename passed to UploadFile is not absolute: '%s'", file_name)
2206

    
2207
  if file_name not in _ALLOWED_UPLOAD_FILES:
2208
    _Fail("Filename passed to UploadFile not in allowed upload targets: '%s'",
2209
          file_name)
2210

    
2211
  raw_data = _Decompress(data)
2212

    
2213
  if not (isinstance(uid, basestring) and isinstance(gid, basestring)):
2214
    _Fail("Invalid username/groupname type")
2215

    
2216
  getents = runtime.GetEnts()
2217
  uid = getents.LookupUser(uid)
2218
  gid = getents.LookupGroup(gid)
2219

    
2220
  utils.SafeWriteFile(file_name, None,
2221
                      data=raw_data, mode=mode, uid=uid, gid=gid,
2222
                      atime=atime, mtime=mtime)
2223

    
2224

    
2225
def RunOob(oob_program, command, node, timeout):
2226
  """Executes oob_program with given command on given node.
2227

2228
  @param oob_program: The path to the executable oob_program
2229
  @param command: The command to invoke on oob_program
2230
  @param node: The node given as an argument to the program
2231
  @param timeout: Timeout after which we kill the oob program
2232

2233
  @return: stdout
2234
  @raise RPCFail: If execution fails for some reason
2235

2236
  """
2237
  result = utils.RunCmd([oob_program, command, node], timeout=timeout)
2238

    
2239
  if result.failed:
2240
    _Fail("'%s' failed with reason '%s'; output: %s", result.cmd,
2241
          result.fail_reason, result.output)
2242

    
2243
  return result.stdout
2244

    
2245

    
2246
def _OSOndiskAPIVersion(os_dir):
2247
  """Compute and return the API version of a given OS.
2248

2249
  This function will try to read the API version of the OS residing in
2250
  the 'os_dir' directory.
2251

2252
  @type os_dir: str
2253
  @param os_dir: the directory in which we should look for the OS
2254
  @rtype: tuple
2255
  @return: tuple (status, data) with status denoting the validity and
2256
      data holding either the vaid versions or an error message
2257

2258
  """
2259
  api_file = utils.PathJoin(os_dir, constants.OS_API_FILE)
2260

    
2261
  try:
2262
    st = os.stat(api_file)
2263
  except EnvironmentError, err:
2264
    return False, ("Required file '%s' not found under path %s: %s" %
2265
                   (constants.OS_API_FILE, os_dir, utils.ErrnoOrStr(err)))
2266

    
2267
  if not stat.S_ISREG(stat.S_IFMT(st.st_mode)):
2268
    return False, ("File '%s' in %s is not a regular file" %
2269
                   (constants.OS_API_FILE, os_dir))
2270

    
2271
  try:
2272
    api_versions = utils.ReadFile(api_file).splitlines()
2273
  except EnvironmentError, err:
2274
    return False, ("Error while reading the API version file at %s: %s" %
2275
                   (api_file, utils.ErrnoOrStr(err)))
2276

    
2277
  try:
2278
    api_versions = [int(version.strip()) for version in api_versions]
2279
  except (TypeError, ValueError), err:
2280
    return False, ("API version(s) can't be converted to integer: %s" %
2281
                   str(err))
2282

    
2283
  return True, api_versions
2284

    
2285

    
2286
def DiagnoseOS(top_dirs=None):
2287
  """Compute the validity for all OSes.
2288

2289
  @type top_dirs: list
2290
  @param top_dirs: the list of directories in which to
2291
      search (if not given defaults to
2292
      L{pathutils.OS_SEARCH_PATH})
2293
  @rtype: list of L{objects.OS}
2294
  @return: a list of tuples (name, path, status, diagnose, variants,
2295
      parameters, api_version) for all (potential) OSes under all
2296
      search paths, where:
2297
          - name is the (potential) OS name
2298
          - path is the full path to the OS
2299
          - status True/False is the validity of the OS
2300
          - diagnose is the error message for an invalid OS, otherwise empty
2301
          - variants is a list of supported OS variants, if any
2302
          - parameters is a list of (name, help) parameters, if any
2303
          - api_version is a list of support OS API versions
2304

2305
  """
2306
  if top_dirs is None:
2307
    top_dirs = pathutils.OS_SEARCH_PATH
2308

    
2309
  result = []
2310
  for dir_name in top_dirs:
2311
    if os.path.isdir(dir_name):
2312
      try:
2313
        f_names = utils.ListVisibleFiles(dir_name)
2314
      except EnvironmentError, err:
2315
        logging.exception("Can't list the OS directory %s: %s", dir_name, err)
2316
        break
2317
      for name in f_names:
2318
        os_path = utils.PathJoin(dir_name, name)
2319
        status, os_inst = _TryOSFromDisk(name, base_dir=dir_name)
2320
        if status:
2321
          diagnose = ""
2322
          variants = os_inst.supported_variants
2323
          parameters = os_inst.supported_parameters
2324
          api_versions = os_inst.api_versions
2325
        else:
2326
          diagnose = os_inst
2327
          variants = parameters = api_versions = []
2328
        result.append((name, os_path, status, diagnose, variants,
2329
                       parameters, api_versions))
2330

    
2331
  return result
2332

    
2333

    
2334
def _TryOSFromDisk(name, base_dir=None):
2335
  """Create an OS instance from disk.
2336

2337
  This function will return an OS instance if the given name is a
2338
  valid OS name.
2339

2340
  @type base_dir: string
2341
  @keyword base_dir: Base directory containing OS installations.
2342
                     Defaults to a search in all the OS_SEARCH_PATH dirs.
2343
  @rtype: tuple
2344
  @return: success and either the OS instance if we find a valid one,
2345
      or error message
2346

2347
  """
2348
  if base_dir is None:
2349
    os_dir = utils.FindFile(name, pathutils.OS_SEARCH_PATH, os.path.isdir)
2350
  else:
2351
    os_dir = utils.FindFile(name, [base_dir], os.path.isdir)
2352

    
2353
  if os_dir is None:
2354
    return False, "Directory for OS %s not found in search path" % name
2355

    
2356
  status, api_versions = _OSOndiskAPIVersion(os_dir)
2357
  if not status:
2358
    # push the error up
2359
    return status, api_versions
2360

    
2361
  if not constants.OS_API_VERSIONS.intersection(api_versions):
2362
    return False, ("API version mismatch for path '%s': found %s, want %s." %
2363
                   (os_dir, api_versions, constants.OS_API_VERSIONS))
2364

    
2365
  # OS Files dictionary, we will populate it with the absolute path
2366
  # names; if the value is True, then it is a required file, otherwise
2367
  # an optional one
2368
  os_files = dict.fromkeys(constants.OS_SCRIPTS, True)
2369

    
2370
  if max(api_versions) >= constants.OS_API_V15:
2371
    os_files[constants.OS_VARIANTS_FILE] = False
2372

    
2373
  if max(api_versions) >= constants.OS_API_V20:
2374
    os_files[constants.OS_PARAMETERS_FILE] = True
2375
  else:
2376
    del os_files[constants.OS_SCRIPT_VERIFY]
2377

    
2378
  for (filename, required) in os_files.items():
2379
    os_files[filename] = utils.PathJoin(os_dir, filename)
2380

    
2381
    try:
2382
      st = os.stat(os_files[filename])
2383
    except EnvironmentError, err:
2384
      if err.errno == errno.ENOENT and not required:
2385
        del os_files[filename]
2386
        continue
2387
      return False, ("File '%s' under path '%s' is missing (%s)" %
2388
                     (filename, os_dir, utils.ErrnoOrStr(err)))
2389

    
2390
    if not stat.S_ISREG(stat.S_IFMT(st.st_mode)):
2391
      return False, ("File '%s' under path '%s' is not a regular file" %
2392
                     (filename, os_dir))
2393

    
2394
    if filename in constants.OS_SCRIPTS:
2395
      if stat.S_IMODE(st.st_mode) & stat.S_IXUSR != stat.S_IXUSR:
2396
        return False, ("File '%s' under path '%s' is not executable" %
2397
                       (filename, os_dir))
2398

    
2399
  variants = []
2400
  if constants.OS_VARIANTS_FILE in os_files:
2401
    variants_file = os_files[constants.OS_VARIANTS_FILE]
2402
    try:
2403
      variants = \
2404
        utils.FilterEmptyLinesAndComments(utils.ReadFile(variants_file))
2405
    except EnvironmentError, err:
2406
      # we accept missing files, but not other errors
2407
      if err.errno != errno.ENOENT:
2408
        return False, ("Error while reading the OS variants file at %s: %s" %
2409
                       (variants_file, utils.ErrnoOrStr(err)))
2410

    
2411
  parameters = []
2412
  if constants.OS_PARAMETERS_FILE in os_files:
2413
    parameters_file = os_files[constants.OS_PARAMETERS_FILE]
2414
    try:
2415
      parameters = utils.ReadFile(parameters_file).splitlines()
2416
    except EnvironmentError, err:
2417
      return False, ("Error while reading the OS parameters file at %s: %s" %
2418
                     (parameters_file, utils.ErrnoOrStr(err)))
2419
    parameters = [v.split(None, 1) for v in parameters]
2420

    
2421
  os_obj = objects.OS(name=name, path=os_dir,
2422
                      create_script=os_files[constants.OS_SCRIPT_CREATE],
2423
                      export_script=os_files[constants.OS_SCRIPT_EXPORT],
2424
                      import_script=os_files[constants.OS_SCRIPT_IMPORT],
2425
                      rename_script=os_files[constants.OS_SCRIPT_RENAME],
2426
                      verify_script=os_files.get(constants.OS_SCRIPT_VERIFY,
2427
                                                 None),
2428
                      supported_variants=variants,
2429
                      supported_parameters=parameters,
2430
                      api_versions=api_versions)
2431
  return True, os_obj
2432

    
2433

    
2434
def OSFromDisk(name, base_dir=None):
2435
  """Create an OS instance from disk.
2436

2437
  This function will return an OS instance if the given name is a
2438
  valid OS name. Otherwise, it will raise an appropriate
2439
  L{RPCFail} exception, detailing why this is not a valid OS.
2440

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

2444
  @type base_dir: string
2445
  @keyword base_dir: Base directory containing OS installations.
2446
                     Defaults to a search in all the OS_SEARCH_PATH dirs.
2447
  @rtype: L{objects.OS}
2448
  @return: the OS instance if we find a valid one
2449
  @raise RPCFail: if we don't find a valid OS
2450

2451
  """
2452
  name_only = objects.OS.GetName(name)
2453
  status, payload = _TryOSFromDisk(name_only, base_dir)
2454

    
2455
  if not status:
2456
    _Fail(payload)
2457

    
2458
  return payload
2459

    
2460

    
2461
def OSCoreEnv(os_name, inst_os, os_params, debug=0):
2462
  """Calculate the basic environment for an os script.
2463

2464
  @type os_name: str
2465
  @param os_name: full operating system name (including variant)
2466
  @type inst_os: L{objects.OS}
2467
  @param inst_os: operating system for which the environment is being built
2468
  @type os_params: dict
2469
  @param os_params: the OS parameters
2470
  @type debug: integer
2471
  @param debug: debug level (0 or 1, for OS Api 10)
2472
  @rtype: dict
2473
  @return: dict of environment variables
2474
  @raise errors.BlockDeviceError: if the block device
2475
      cannot be found
2476

2477
  """
2478
  result = {}
2479
  api_version = \
2480
    max(constants.OS_API_VERSIONS.intersection(inst_os.api_versions))
2481
  result["OS_API_VERSION"] = "%d" % api_version
2482
  result["OS_NAME"] = inst_os.name
2483
  result["DEBUG_LEVEL"] = "%d" % debug
2484

    
2485
  # OS variants
2486
  if api_version >= constants.OS_API_V15 and inst_os.supported_variants:
2487
    variant = objects.OS.GetVariant(os_name)
2488
    if not variant:
2489
      variant = inst_os.supported_variants[0]
2490
  else:
2491
    variant = ""
2492
  result["OS_VARIANT"] = variant
2493

    
2494
  # OS params
2495
  for pname, pvalue in os_params.items():
2496
    result["OSP_%s" % pname.upper()] = pvalue
2497

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

    
2503
  return result
2504

    
2505

    
2506
def OSEnvironment(instance, inst_os, debug=0):
2507
  """Calculate the environment for an os script.
2508

2509
  @type instance: L{objects.Instance}
2510
  @param instance: target instance for the os script run
2511
  @type inst_os: L{objects.OS}
2512
  @param inst_os: operating system for which the environment is being built
2513
  @type debug: integer
2514
  @param debug: debug level (0 or 1, for OS Api 10)
2515
  @rtype: dict
2516
  @return: dict of environment variables
2517
  @raise errors.BlockDeviceError: if the block device
2518
      cannot be found
2519

2520
  """
2521
  result = OSCoreEnv(instance.os, inst_os, instance.osparams, debug=debug)
2522

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

    
2526
  result["HYPERVISOR"] = instance.hypervisor
2527
  result["DISK_COUNT"] = "%d" % len(instance.disks)
2528
  result["NIC_COUNT"] = "%d" % len(instance.nics)
2529
  result["INSTANCE_SECONDARY_NODES"] = \
2530
      ("%s" % " ".join(instance.secondary_nodes))
2531

    
2532
  # Disks
2533
  for idx, disk in enumerate(instance.disks):
2534
    real_disk = _OpenRealBD(disk)
2535
    result["DISK_%d_PATH" % idx] = real_disk.dev_path
2536
    result["DISK_%d_ACCESS" % idx] = disk.mode
2537
    if constants.HV_DISK_TYPE in instance.hvparams:
2538
      result["DISK_%d_FRONTEND_TYPE" % idx] = \
2539
        instance.hvparams[constants.HV_DISK_TYPE]
2540
    if disk.dev_type in constants.LDS_BLOCK:
2541
      result["DISK_%d_BACKEND_TYPE" % idx] = "block"
2542
    elif disk.dev_type == constants.LD_FILE:
2543
      result["DISK_%d_BACKEND_TYPE" % idx] = \
2544
        "file:%s" % disk.physical_id[0]
2545

    
2546
  # NICs
2547
  for idx, nic in enumerate(instance.nics):
2548
    result["NIC_%d_MAC" % idx] = nic.mac
2549
    if nic.ip:
2550
      result["NIC_%d_IP" % idx] = nic.ip
2551
    result["NIC_%d_MODE" % idx] = nic.nicparams[constants.NIC_MODE]
2552
    if nic.nicparams[constants.NIC_MODE] == constants.NIC_MODE_BRIDGED:
2553
      result["NIC_%d_BRIDGE" % idx] = nic.nicparams[constants.NIC_LINK]
2554
    if nic.nicparams[constants.NIC_LINK]:
2555
      result["NIC_%d_LINK" % idx] = nic.nicparams[constants.NIC_LINK]
2556
    if nic.netinfo:
2557
      nobj = objects.Network.FromDict(nic.netinfo)
2558
      result.update(nobj.HooksDict("NIC_%d_" % idx))
2559
    elif nic.network:
2560
      # FIXME: broken network reference: the instance NIC specifies a network,
2561
      # but the relevant network entry was not in the config. This should be
2562
      # made impossible.
2563
      result["INSTANCE_NIC%d_NETWORK" % idx] = nic.network
2564
    if constants.HV_NIC_TYPE in instance.hvparams:
2565
      result["NIC_%d_FRONTEND_TYPE" % idx] = \
2566
        instance.hvparams[constants.HV_NIC_TYPE]
2567

    
2568
  # HV/BE params
2569
  for source, kind in [(instance.beparams, "BE"), (instance.hvparams, "HV")]:
2570
    for key, value in source.items():
2571
      result["INSTANCE_%s_%s" % (kind, key)] = str(value)
2572

    
2573
  return result
2574

    
2575

    
2576
def DiagnoseExtStorage(top_dirs=None):
2577
  """Compute the validity for all ExtStorage Providers.
2578

2579
  @type top_dirs: list
2580
  @param top_dirs: the list of directories in which to
2581
      search (if not given defaults to
2582
      L{pathutils.ES_SEARCH_PATH})
2583
  @rtype: list of L{objects.ExtStorage}
2584
  @return: a list of tuples (name, path, status, diagnose, parameters)
2585
      for all (potential) ExtStorage Providers under all
2586
      search paths, where:
2587
          - name is the (potential) ExtStorage Provider
2588
          - path is the full path to the ExtStorage Provider
2589
          - status True/False is the validity of the ExtStorage Provider
2590
          - diagnose is the error message for an invalid ExtStorage Provider,
2591
            otherwise empty
2592
          - parameters is a list of (name, help) parameters, if any
2593

2594
  """
2595
  if top_dirs is None:
2596
    top_dirs = pathutils.ES_SEARCH_PATH
2597

    
2598
  result = []
2599
  for dir_name in top_dirs:
2600
    if os.path.isdir(dir_name):
2601
      try:
2602
        f_names = utils.ListVisibleFiles(dir_name)
2603
      except EnvironmentError, err:
2604
        logging.exception("Can't list the ExtStorage directory %s: %s",
2605
                          dir_name, err)
2606
        break
2607
      for name in f_names:
2608
        es_path = utils.PathJoin(dir_name, name)
2609
        status, es_inst = bdev.ExtStorageFromDisk(name, base_dir=dir_name)
2610
        if status:
2611
          diagnose = ""
2612
          parameters = es_inst.supported_parameters
2613
        else:
2614
          diagnose = es_inst
2615
          parameters = []
2616
        result.append((name, es_path, status, diagnose, parameters))
2617

    
2618
  return result
2619

    
2620

    
2621
def BlockdevGrow(disk, amount, dryrun, backingstore):
2622
  """Grow a stack of block devices.
2623

2624
  This function is called recursively, with the childrens being the
2625
  first ones to resize.
2626

2627
  @type disk: L{objects.Disk}
2628
  @param disk: the disk to be grown
2629
  @type amount: integer
2630
  @param amount: the amount (in mebibytes) to grow with
2631
  @type dryrun: boolean
2632
  @param dryrun: whether to execute the operation in simulation mode
2633
      only, without actually increasing the size
2634
  @param backingstore: whether to execute the operation on backing storage
2635
      only, or on "logical" storage only; e.g. DRBD is logical storage,
2636
      whereas LVM, file, RBD are backing storage
2637
  @rtype: (status, result)
2638
  @return: a tuple with the status of the operation (True/False), and
2639
      the errors message if status is False
2640

2641
  """
2642
  r_dev = _RecursiveFindBD(disk)
2643
  if r_dev is None:
2644
    _Fail("Cannot find block device %s", disk)
2645

    
2646
  try:
2647
    r_dev.Grow(amount, dryrun, backingstore)
2648
  except errors.BlockDeviceError, err:
2649
    _Fail("Failed to grow block device: %s", err, exc=True)
2650

    
2651

    
2652
def BlockdevSnapshot(disk):
2653
  """Create a snapshot copy of a block device.
2654

2655
  This function is called recursively, and the snapshot is actually created
2656
  just for the leaf lvm backend device.
2657

2658
  @type disk: L{objects.Disk}
2659
  @param disk: the disk to be snapshotted
2660
  @rtype: string
2661
  @return: snapshot disk ID as (vg, lv)
2662

2663
  """
2664
  if disk.dev_type == constants.LD_DRBD8:
2665
    if not disk.children:
2666
      _Fail("DRBD device '%s' without backing storage cannot be snapshotted",
2667
            disk.unique_id)
2668
    return BlockdevSnapshot(disk.children[0])
2669
  elif disk.dev_type == constants.LD_LV:
2670
    r_dev = _RecursiveFindBD(disk)
2671
    if r_dev is not None:
2672
      # FIXME: choose a saner value for the snapshot size
2673
      # let's stay on the safe side and ask for the full size, for now
2674
      return r_dev.Snapshot(disk.size)
2675
    else:
2676
      _Fail("Cannot find block device %s", disk)
2677
  else:
2678
    _Fail("Cannot snapshot non-lvm block device '%s' of type '%s'",
2679
          disk.unique_id, disk.dev_type)
2680

    
2681

    
2682
def BlockdevSetInfo(disk, info):
2683
  """Sets 'metadata' information on block devices.
2684

2685
  This function sets 'info' metadata on block devices. Initial
2686
  information is set at device creation; this function should be used
2687
  for example after renames.
2688

2689
  @type disk: L{objects.Disk}
2690
  @param disk: the disk to be grown
2691
  @type info: string
2692
  @param info: new 'info' metadata
2693
  @rtype: (status, result)
2694
  @return: a tuple with the status of the operation (True/False), and
2695
      the errors message if status is False
2696

2697
  """
2698
  r_dev = _RecursiveFindBD(disk)
2699
  if r_dev is None:
2700
    _Fail("Cannot find block device %s", disk)
2701

    
2702
  try:
2703
    r_dev.SetInfo(info)
2704
  except errors.BlockDeviceError, err:
2705
    _Fail("Failed to set information on block device: %s", err, exc=True)
2706

    
2707

    
2708
def FinalizeExport(instance, snap_disks):
2709
  """Write out the export configuration information.
2710

2711
  @type instance: L{objects.Instance}
2712
  @param instance: the instance which we export, used for
2713
      saving configuration
2714
  @type snap_disks: list of L{objects.Disk}
2715
  @param snap_disks: list of snapshot block devices, which
2716
      will be used to get the actual name of the dump file
2717

2718
  @rtype: None
2719

2720
  """
2721
  destdir = utils.PathJoin(pathutils.EXPORT_DIR, instance.name + ".new")
2722
  finaldestdir = utils.PathJoin(pathutils.EXPORT_DIR, instance.name)
2723

    
2724
  config = objects.SerializableConfigParser()
2725

    
2726
  config.add_section(constants.INISECT_EXP)
2727
  config.set(constants.INISECT_EXP, "version", "0")
2728
  config.set(constants.INISECT_EXP, "timestamp", "%d" % int(time.time()))
2729
  config.set(constants.INISECT_EXP, "source", instance.primary_node)
2730
  config.set(constants.INISECT_EXP, "os", instance.os)
2731
  config.set(constants.INISECT_EXP, "compression", "none")
2732

    
2733
  config.add_section(constants.INISECT_INS)
2734
  config.set(constants.INISECT_INS, "name", instance.name)
2735
  config.set(constants.INISECT_INS, "maxmem", "%d" %
2736
             instance.beparams[constants.BE_MAXMEM])
2737
  config.set(constants.INISECT_INS, "minmem", "%d" %
2738
             instance.beparams[constants.BE_MINMEM])
2739
  # "memory" is deprecated, but useful for exporting to old ganeti versions
2740
  config.set(constants.INISECT_INS, "memory", "%d" %
2741
             instance.beparams[constants.BE_MAXMEM])
2742
  config.set(constants.INISECT_INS, "vcpus", "%d" %
2743
             instance.beparams[constants.BE_VCPUS])
2744
  config.set(constants.INISECT_INS, "disk_template", instance.disk_template)
2745
  config.set(constants.INISECT_INS, "hypervisor", instance.hypervisor)
2746
  config.set(constants.INISECT_INS, "tags", " ".join(instance.GetTags()))
2747

    
2748
  nic_total = 0
2749
  for nic_count, nic in enumerate(instance.nics):
2750
    nic_total += 1
2751
    config.set(constants.INISECT_INS, "nic%d_mac" %
2752
               nic_count, "%s" % nic.mac)
2753
    config.set(constants.INISECT_INS, "nic%d_ip" % nic_count, "%s" % nic.ip)
2754
    config.set(constants.INISECT_INS, "nic%d_network" % nic_count,
2755
               "%s" % nic.network)
2756
    for param in constants.NICS_PARAMETER_TYPES:
2757
      config.set(constants.INISECT_INS, "nic%d_%s" % (nic_count, param),
2758
                 "%s" % nic.nicparams.get(param, None))
2759
  # TODO: redundant: on load can read nics until it doesn't exist
2760
  config.set(constants.INISECT_INS, "nic_count", "%d" % nic_total)
2761

    
2762
  disk_total = 0
2763
  for disk_count, disk in enumerate(snap_disks):
2764
    if disk:
2765
      disk_total += 1
2766
      config.set(constants.INISECT_INS, "disk%d_ivname" % disk_count,
2767
                 ("%s" % disk.iv_name))
2768
      config.set(constants.INISECT_INS, "disk%d_dump" % disk_count,
2769
                 ("%s" % disk.physical_id[1]))
2770
      config.set(constants.INISECT_INS, "disk%d_size" % disk_count,
2771
                 ("%d" % disk.size))
2772

    
2773
  config.set(constants.INISECT_INS, "disk_count", "%d" % disk_total)
2774

    
2775
  # New-style hypervisor/backend parameters
2776

    
2777
  config.add_section(constants.INISECT_HYP)
2778
  for name, value in instance.hvparams.items():
2779
    if name not in constants.HVC_GLOBALS:
2780
      config.set(constants.INISECT_HYP, name, str(value))
2781

    
2782
  config.add_section(constants.INISECT_BEP)
2783
  for name, value in instance.beparams.items():
2784
    config.set(constants.INISECT_BEP, name, str(value))
2785

    
2786
  config.add_section(constants.INISECT_OSP)
2787
  for name, value in instance.osparams.items():
2788
    config.set(constants.INISECT_OSP, name, str(value))
2789

    
2790
  utils.WriteFile(utils.PathJoin(destdir, constants.EXPORT_CONF_FILE),
2791
                  data=config.Dumps())
2792
  shutil.rmtree(finaldestdir, ignore_errors=True)
2793
  shutil.move(destdir, finaldestdir)
2794

    
2795

    
2796
def ExportInfo(dest):
2797
  """Get export configuration information.
2798

2799
  @type dest: str
2800
  @param dest: directory containing the export
2801

2802
  @rtype: L{objects.SerializableConfigParser}
2803
  @return: a serializable config file containing the
2804
      export info
2805

2806
  """
2807
  cff = utils.PathJoin(dest, constants.EXPORT_CONF_FILE)
2808

    
2809
  config = objects.SerializableConfigParser()
2810
  config.read(cff)
2811

    
2812
  if (not config.has_section(constants.INISECT_EXP) or
2813
      not config.has_section(constants.INISECT_INS)):
2814
    _Fail("Export info file doesn't have the required fields")
2815

    
2816
  return config.Dumps()
2817

    
2818

    
2819
def ListExports():
2820
  """Return a list of exports currently available on this machine.
2821

2822
  @rtype: list
2823
  @return: list of the exports
2824

2825
  """
2826
  if os.path.isdir(pathutils.EXPORT_DIR):
2827
    return sorted(utils.ListVisibleFiles(pathutils.EXPORT_DIR))
2828
  else:
2829
    _Fail("No exports directory")
2830

    
2831

    
2832
def RemoveExport(export):
2833
  """Remove an existing export from the node.
2834

2835
  @type export: str
2836
  @param export: the name of the export to remove
2837
  @rtype: None
2838

2839
  """
2840
  target = utils.PathJoin(pathutils.EXPORT_DIR, export)
2841

    
2842
  try:
2843
    shutil.rmtree(target)
2844
  except EnvironmentError, err:
2845
    _Fail("Error while removing the export: %s", err, exc=True)
2846

    
2847

    
2848
def BlockdevRename(devlist):
2849
  """Rename a list of block devices.
2850

2851
  @type devlist: list of tuples
2852
  @param devlist: list of tuples of the form  (disk,
2853
      new_logical_id, new_physical_id); disk is an
2854
      L{objects.Disk} object describing the current disk,
2855
      and new logical_id/physical_id is the name we
2856
      rename it to
2857
  @rtype: boolean
2858
  @return: True if all renames succeeded, False otherwise
2859

2860
  """
2861
  msgs = []
2862
  result = True
2863
  for disk, unique_id in devlist:
2864
    dev = _RecursiveFindBD(disk)
2865
    if dev is None:
2866
      msgs.append("Can't find device %s in rename" % str(disk))
2867
      result = False
2868
      continue
2869
    try:
2870
      old_rpath = dev.dev_path
2871
      dev.Rename(unique_id)
2872
      new_rpath = dev.dev_path
2873
      if old_rpath != new_rpath:
2874
        DevCacheManager.RemoveCache(old_rpath)
2875
        # FIXME: we should add the new cache information here, like:
2876
        # DevCacheManager.UpdateCache(new_rpath, owner, ...)
2877
        # but we don't have the owner here - maybe parse from existing
2878
        # cache? for now, we only lose lvm data when we rename, which
2879
        # is less critical than DRBD or MD
2880
    except errors.BlockDeviceError, err:
2881
      msgs.append("Can't rename device '%s' to '%s': %s" %
2882
                  (dev, unique_id, err))
2883
      logging.exception("Can't rename device '%s' to '%s'", dev, unique_id)
2884
      result = False
2885
  if not result:
2886
    _Fail("; ".join(msgs))
2887

    
2888

    
2889
def _TransformFileStorageDir(fs_dir):
2890
  """Checks whether given file_storage_dir is valid.
2891

2892
  Checks wheter the given fs_dir is within the cluster-wide default
2893
  file_storage_dir or the shared_file_storage_dir, which are stored in
2894
  SimpleStore. Only paths under those directories are allowed.
2895

2896
  @type fs_dir: str
2897
  @param fs_dir: the path to check
2898

2899
  @return: the normalized path if valid, None otherwise
2900

2901
  """
2902
  if not (constants.ENABLE_FILE_STORAGE or
2903
          constants.ENABLE_SHARED_FILE_STORAGE):
2904
    _Fail("File storage disabled at configure time")
2905

    
2906
  bdev.CheckFileStoragePath(fs_dir)
2907

    
2908
  return os.path.normpath(fs_dir)
2909

    
2910

    
2911
def CreateFileStorageDir(file_storage_dir):
2912
  """Create file storage directory.
2913

2914
  @type file_storage_dir: str
2915
  @param file_storage_dir: directory to create
2916

2917
  @rtype: tuple
2918
  @return: tuple with first element a boolean indicating wheter dir
2919
      creation was successful or not
2920

2921
  """
2922
  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
2923
  if os.path.exists(file_storage_dir):
2924
    if not os.path.isdir(file_storage_dir):
2925
      _Fail("Specified storage dir '%s' is not a directory",
2926
            file_storage_dir)
2927
  else:
2928
    try:
2929
      os.makedirs(file_storage_dir, 0750)
2930
    except OSError, err:
2931
      _Fail("Cannot create file storage directory '%s': %s",
2932
            file_storage_dir, err, exc=True)
2933

    
2934

    
2935
def RemoveFileStorageDir(file_storage_dir):
2936
  """Remove file storage directory.
2937

2938
  Remove it only if it's empty. If not log an error and return.
2939

2940
  @type file_storage_dir: str
2941
  @param file_storage_dir: the directory we should cleanup
2942
  @rtype: tuple (success,)
2943
  @return: tuple of one element, C{success}, denoting
2944
      whether the operation was successful
2945

2946
  """
2947
  file_storage_dir = _TransformFileStorageDir(file_storage_dir)
2948
  if os.path.exists(file_storage_dir):
2949
    if not os.path.isdir(file_storage_dir):
2950
      _Fail("Specified Storage directory '%s' is not a directory",
2951
            file_storage_dir)
2952
    # deletes dir only if empty, otherwise we want to fail the rpc call
2953
    try:
2954
      os.rmdir(file_storage_dir)
2955
    except OSError, err:
2956
      _Fail("Cannot remove file storage directory '%s': %s",
2957
            file_storage_dir, err)
2958

    
2959

    
2960
def RenameFileStorageDir(old_file_storage_dir, new_file_storage_dir):
2961
  """Rename the file storage directory.
2962

2963
  @type old_file_storage_dir: str
2964
  @param old_file_storage_dir: the current path
2965
  @type new_file_storage_dir: str
2966
  @param new_file_storage_dir: the name we should rename to
2967
  @rtype: tuple (success,)
2968
  @return: tuple of one element, C{success}, denoting
2969
      whether the operation was successful
2970

2971
  """
2972
  old_file_storage_dir = _TransformFileStorageDir(old_file_storage_dir)
2973
  new_file_storage_dir = _TransformFileStorageDir(new_file_storage_dir)
2974
  if not os.path.exists(new_file_storage_dir):
2975
    if os.path.isdir(old_file_storage_dir):
2976
      try:
2977
        os.rename(old_file_storage_dir, new_file_storage_dir)
2978
      except OSError, err:
2979
        _Fail("Cannot rename '%s' to '%s': %s",
2980
              old_file_storage_dir, new_file_storage_dir, err)
2981
    else:
2982
      _Fail("Specified storage dir '%s' is not a directory",
2983
            old_file_storage_dir)
2984
  else:
2985
    if os.path.exists(old_file_storage_dir):
2986
      _Fail("Cannot rename '%s' to '%s': both locations exist",
2987
            old_file_storage_dir, new_file_storage_dir)
2988

    
2989

    
2990
def _EnsureJobQueueFile(file_name):
2991
  """Checks whether the given filename is in the queue directory.
2992

2993
  @type file_name: str
2994
  @param file_name: the file name we should check
2995
  @rtype: None
2996
  @raises RPCFail: if the file is not valid
2997

2998
  """
2999
  if not utils.IsBelowDir(pathutils.QUEUE_DIR, file_name):
3000
    _Fail("Passed job queue file '%s' does not belong to"
3001
          " the queue directory '%s'", file_name, pathutils.QUEUE_DIR)
3002

    
3003

    
3004
def JobQueueUpdate(file_name, content):
3005
  """Updates a file in the queue directory.
3006

3007
  This is just a wrapper over L{utils.io.WriteFile}, with proper
3008
  checking.
3009

3010
  @type file_name: str
3011
  @param file_name: the job file name
3012
  @type content: str
3013
  @param content: the new job contents
3014
  @rtype: boolean
3015
  @return: the success of the operation
3016

3017
  """
3018
  file_name = vcluster.LocalizeVirtualPath(file_name)
3019

    
3020
  _EnsureJobQueueFile(file_name)
3021
  getents = runtime.GetEnts()
3022

    
3023
  # Write and replace the file atomically
3024
  utils.WriteFile(file_name, data=_Decompress(content), uid=getents.masterd_uid,
3025
                  gid=getents.masterd_gid)
3026

    
3027

    
3028
def JobQueueRename(old, new):
3029
  """Renames a job queue file.
3030

3031
  This is just a wrapper over os.rename with proper checking.
3032

3033
  @type old: str
3034
  @param old: the old (actual) file name
3035
  @type new: str
3036
  @param new: the desired file name
3037
  @rtype: tuple
3038
  @return: the success of the operation and payload
3039

3040
  """
3041
  old = vcluster.LocalizeVirtualPath(old)
3042
  new = vcluster.LocalizeVirtualPath(new)
3043

    
3044
  _EnsureJobQueueFile(old)
3045
  _EnsureJobQueueFile(new)
3046

    
3047
  getents = runtime.GetEnts()
3048

    
3049
  utils.RenameFile(old, new, mkdir=True, mkdir_mode=0700,
3050
                   dir_uid=getents.masterd_uid, dir_gid=getents.masterd_gid)
3051

    
3052

    
3053
def BlockdevClose(instance_name, disks):
3054
  """Closes the given block devices.
3055

3056
  This means they will be switched to secondary mode (in case of
3057
  DRBD).
3058

3059
  @param instance_name: if the argument is not empty, the symlinks
3060
      of this instance will be removed
3061
  @type disks: list of L{objects.Disk}
3062
  @param disks: the list of disks to be closed
3063
  @rtype: tuple (success, message)
3064
  @return: a tuple of success and message, where success
3065
      indicates the succes of the operation, and message
3066
      which will contain the error details in case we
3067
      failed
3068

3069
  """
3070
  bdevs = []
3071
  for cf in disks:
3072
    rd = _RecursiveFindBD(cf)
3073
    if rd is None:
3074
      _Fail("Can't find device %s", cf)
3075
    bdevs.append(rd)
3076

    
3077
  msg = []
3078
  for rd in bdevs:
3079
    try:
3080
      rd.Close()
3081
    except errors.BlockDeviceError, err:
3082
      msg.append(str(err))
3083
  if msg:
3084
    _Fail("Can't make devices secondary: %s", ",".join(msg))
3085
  else:
3086
    if instance_name:
3087
      _RemoveBlockDevLinks(instance_name, disks)
3088

    
3089

    
3090
def ValidateHVParams(hvname, hvparams):
3091
  """Validates the given hypervisor parameters.
3092

3093
  @type hvname: string
3094
  @param hvname: the hypervisor name
3095
  @type hvparams: dict
3096
  @param hvparams: the hypervisor parameters to be validated
3097
  @rtype: None
3098

3099
  """
3100
  try:
3101
    hv_type = hypervisor.GetHypervisor(hvname)
3102
    hv_type.ValidateParameters(hvparams)
3103
  except errors.HypervisorError, err:
3104
    _Fail(str(err), log=False)
3105

    
3106

    
3107
def _CheckOSPList(os_obj, parameters):
3108
  """Check whether a list of parameters is supported by the OS.
3109

3110
  @type os_obj: L{objects.OS}
3111
  @param os_obj: OS object to check
3112
  @type parameters: list
3113
  @param parameters: the list of parameters to check
3114

3115
  """
3116
  supported = [v[0] for v in os_obj.supported_parameters]
3117
  delta = frozenset(parameters).difference(supported)
3118
  if delta:
3119
    _Fail("The following parameters are not supported"
3120
          " by the OS %s: %s" % (os_obj.name, utils.CommaJoin(delta)))
3121

    
3122

    
3123
def ValidateOS(required, osname, checks, osparams):
3124
  """Validate the given OS' parameters.
3125

3126
  @type required: boolean
3127
  @param required: whether absence of the OS should translate into
3128
      failure or not
3129
  @type osname: string
3130
  @param osname: the OS to be validated
3131
  @type checks: list
3132
  @param checks: list of the checks to run (currently only 'parameters')
3133
  @type osparams: dict
3134
  @param osparams: dictionary with OS parameters
3135
  @rtype: boolean
3136
  @return: True if the validation passed, or False if the OS was not
3137
      found and L{required} was false
3138

3139
  """
3140
  if not constants.OS_VALIDATE_CALLS.issuperset(checks):
3141
    _Fail("Unknown checks required for OS %s: %s", osname,
3142
          set(checks).difference(constants.OS_VALIDATE_CALLS))
3143

    
3144
  name_only = objects.OS.GetName(osname)
3145
  status, tbv = _TryOSFromDisk(name_only, None)
3146

    
3147
  if not status:
3148
    if required:
3149
      _Fail(tbv)
3150
    else:
3151
      return False
3152

    
3153
  if max(tbv.api_versions) < constants.OS_API_V20:
3154
    return True
3155

    
3156
  if constants.OS_VALIDATE_PARAMETERS in checks:
3157
    _CheckOSPList(tbv, osparams.keys())
3158

    
3159
  validate_env = OSCoreEnv(osname, tbv, osparams)
3160
  result = utils.RunCmd([tbv.verify_script] + checks, env=validate_env,
3161
                        cwd=tbv.path, reset_env=True)
3162
  if result.failed:
3163
    logging.error("os validate command '%s' returned error: %s output: %s",
3164
                  result.cmd, result.fail_reason, result.output)
3165
    _Fail("OS validation script failed (%s), output: %s",
3166
          result.fail_reason, result.output, log=False)
3167

    
3168
  return True
3169

    
3170

    
3171
def DemoteFromMC():
3172
  """Demotes the current node from master candidate role.
3173

3174
  """
3175
  # try to ensure we're not the master by mistake
3176
  master, myself = ssconf.GetMasterAndMyself()
3177
  if master == myself:
3178
    _Fail("ssconf status shows I'm the master node, will not demote")
3179

    
3180
  result = utils.RunCmd([pathutils.DAEMON_UTIL, "check", constants.MASTERD])
3181
  if not result.failed:
3182
    _Fail("The master daemon is running, will not demote")
3183

    
3184
  try:
3185
    if os.path.isfile(pathutils.CLUSTER_CONF_FILE):
3186
      utils.CreateBackup(pathutils.CLUSTER_CONF_FILE)
3187
  except EnvironmentError, err:
3188
    if err.errno != errno.ENOENT:
3189
      _Fail("Error while backing up cluster file: %s", err, exc=True)
3190

    
3191
  utils.RemoveFile(pathutils.CLUSTER_CONF_FILE)
3192

    
3193

    
3194
def _GetX509Filenames(cryptodir, name):
3195
  """Returns the full paths for the private key and certificate.
3196

3197
  """
3198
  return (utils.PathJoin(cryptodir, name),
3199
          utils.PathJoin(cryptodir, name, _X509_KEY_FILE),
3200
          utils.PathJoin(cryptodir, name, _X509_CERT_FILE))
3201

    
3202

    
3203
def CreateX509Certificate(validity, cryptodir=pathutils.CRYPTO_KEYS_DIR):
3204
  """Creates a new X509 certificate for SSL/TLS.
3205

3206
  @type validity: int
3207
  @param validity: Validity in seconds
3208
  @rtype: tuple; (string, string)
3209
  @return: Certificate name and public part
3210

3211
  """
3212
  (key_pem, cert_pem) = \
3213
    utils.GenerateSelfSignedX509Cert(netutils.Hostname.GetSysName(),
3214
                                     min(validity, _MAX_SSL_CERT_VALIDITY))
3215

    
3216
  cert_dir = tempfile.mkdtemp(dir=cryptodir,
3217
                              prefix="x509-%s-" % utils.TimestampForFilename())
3218
  try:
3219
    name = os.path.basename(cert_dir)
3220
    assert len(name) > 5
3221

    
3222
    (_, key_file, cert_file) = _GetX509Filenames(cryptodir, name)
3223

    
3224
    utils.WriteFile(key_file, mode=0400, data=key_pem)
3225
    utils.WriteFile(cert_file, mode=0400, data=cert_pem)
3226

    
3227
    # Never return private key as it shouldn't leave the node
3228
    return (name, cert_pem)
3229
  except Exception:
3230
    shutil.rmtree(cert_dir, ignore_errors=True)
3231
    raise
3232

    
3233

    
3234
def RemoveX509Certificate(name, cryptodir=pathutils.CRYPTO_KEYS_DIR):
3235
  """Removes a X509 certificate.
3236

3237
  @type name: string
3238
  @param name: Certificate name
3239

3240
  """
3241
  (cert_dir, key_file, cert_file) = _GetX509Filenames(cryptodir, name)
3242

    
3243
  utils.RemoveFile(key_file)
3244
  utils.RemoveFile(cert_file)
3245

    
3246
  try:
3247
    os.rmdir(cert_dir)
3248
  except EnvironmentError, err:
3249
    _Fail("Cannot remove certificate directory '%s': %s",
3250
          cert_dir, err)
3251

    
3252

    
3253
def _GetImportExportIoCommand(instance, mode, ieio, ieargs):
3254
  """Returns the command for the requested input/output.
3255

3256
  @type instance: L{objects.Instance}
3257
  @param instance: The instance object
3258
  @param mode: Import/export mode
3259
  @param ieio: Input/output type
3260
  @param ieargs: Input/output arguments
3261

3262
  """
3263
  assert mode in (constants.IEM_IMPORT, constants.IEM_EXPORT)
3264

    
3265
  env = None
3266
  prefix = None
3267
  suffix = None
3268
  exp_size = None
3269

    
3270
  if ieio == constants.IEIO_FILE:
3271
    (filename, ) = ieargs
3272

    
3273
    if not utils.IsNormAbsPath(filename):
3274
      _Fail("Path '%s' is not normalized or absolute", filename)
3275

    
3276
    real_filename = os.path.realpath(filename)
3277
    directory = os.path.dirname(real_filename)
3278

    
3279
    if not utils.IsBelowDir(pathutils.EXPORT_DIR, real_filename):
3280
      _Fail("File '%s' is not under exports directory '%s': %s",
3281
            filename, pathutils.EXPORT_DIR, real_filename)
3282

    
3283
    # Create directory
3284
    utils.Makedirs(directory, mode=0750)
3285

    
3286
    quoted_filename = utils.ShellQuote(filename)
3287

    
3288
    if mode == constants.IEM_IMPORT:
3289
      suffix = "> %s" % quoted_filename
3290
    elif mode == constants.IEM_EXPORT:
3291
      suffix = "< %s" % quoted_filename
3292

    
3293
      # Retrieve file size
3294
      try:
3295
        st = os.stat(filename)
3296
      except EnvironmentError, err:
3297
        logging.error("Can't stat(2) %s: %s", filename, err)
3298
      else:
3299
        exp_size = utils.BytesToMebibyte(st.st_size)
3300

    
3301
  elif ieio == constants.IEIO_RAW_DISK:
3302
    (disk, ) = ieargs
3303

    
3304
    real_disk = _OpenRealBD(disk)
3305

    
3306
    if mode == constants.IEM_IMPORT:
3307
      # we set here a smaller block size as, due to transport buffering, more
3308
      # than 64-128k will mostly ignored; we use nocreat to fail if the device
3309
      # is not already there or we pass a wrong path; we use notrunc to no
3310
      # attempt truncate on an LV device; we use oflag=dsync to not buffer too
3311
      # much memory; this means that at best, we flush every 64k, which will
3312
      # not be very fast
3313
      suffix = utils.BuildShellCmd(("| dd of=%s conv=nocreat,notrunc"
3314
                                    " bs=%s oflag=dsync"),
3315
                                    real_disk.dev_path,
3316
                                    str(64 * 1024))
3317

    
3318
    elif mode == constants.IEM_EXPORT:
3319
      # the block size on the read dd is 1MiB to match our units
3320
      prefix = utils.BuildShellCmd("dd if=%s bs=%s count=%s |",
3321
                                   real_disk.dev_path,
3322
                                   str(1024 * 1024), # 1 MB
3323
                                   str(disk.size))
3324
      exp_size = disk.size
3325

    
3326
  elif ieio == constants.IEIO_SCRIPT:
3327
    (disk, disk_index, ) = ieargs
3328

    
3329
    assert isinstance(disk_index, (int, long))
3330

    
3331
    real_disk = _OpenRealBD(disk)
3332

    
3333
    inst_os = OSFromDisk(instance.os)
3334
    env = OSEnvironment(instance, inst_os)
3335

    
3336
    if mode == constants.IEM_IMPORT:
3337
      env["IMPORT_DEVICE"] = env["DISK_%d_PATH" % disk_index]
3338
      env["IMPORT_INDEX"] = str(disk_index)
3339
      script = inst_os.import_script
3340

    
3341
    elif mode == constants.IEM_EXPORT:
3342
      env["EXPORT_DEVICE"] = real_disk.dev_path
3343
      env["EXPORT_INDEX"] = str(disk_index)
3344
      script = inst_os.export_script
3345

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

    
3349
    if mode == constants.IEM_IMPORT:
3350
      suffix = "| %s" % script_cmd
3351

    
3352
    elif mode == constants.IEM_EXPORT:
3353
      prefix = "%s |" % script_cmd
3354

    
3355
    # Let script predict size
3356
    exp_size = constants.IE_CUSTOM_SIZE
3357

    
3358
  else:
3359
    _Fail("Invalid %s I/O mode %r", mode, ieio)
3360

    
3361
  return (env, prefix, suffix, exp_size)
3362

    
3363

    
3364
def _CreateImportExportStatusDir(prefix):
3365
  """Creates status directory for import/export.
3366

3367
  """
3368
  return tempfile.mkdtemp(dir=pathutils.IMPORT_EXPORT_DIR,
3369
                          prefix=("%s-%s-" %
3370
                                  (prefix, utils.TimestampForFilename())))
3371

    
3372

    
3373
def StartImportExportDaemon(mode, opts, host, port, instance, component,
3374
                            ieio, ieioargs):
3375
  """Starts an import or export daemon.
3376

3377
  @param mode: Import/output mode
3378
  @type opts: L{objects.ImportExportOptions}
3379
  @param opts: Daemon options
3380
  @type host: string
3381
  @param host: Remote host for export (None for import)
3382
  @type port: int
3383
  @param port: Remote port for export (None for import)
3384
  @type instance: L{objects.Instance}
3385
  @param instance: Instance object
3386
  @type component: string
3387
  @param component: which part of the instance is transferred now,
3388
      e.g. 'disk/0'
3389
  @param ieio: Input/output type
3390
  @param ieioargs: Input/output arguments
3391

3392
  """
3393
  if mode == constants.IEM_IMPORT:
3394
    prefix = "import"
3395

    
3396
    if not (host is None and port is None):
3397
      _Fail("Can not specify host or port on import")
3398

    
3399
  elif mode == constants.IEM_EXPORT:
3400
    prefix = "export"
3401

    
3402
    if host is None or port is None:
3403
      _Fail("Host and port must be specified for an export")
3404

    
3405
  else:
3406
    _Fail("Invalid mode %r", mode)
3407

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

    
3411
  (cmd_env, cmd_prefix, cmd_suffix, exp_size) = \
3412
    _GetImportExportIoCommand(instance, mode, ieio, ieioargs)
3413

    
3414
  if opts.key_name is None:
3415
    # Use server.pem
3416
    key_path = pathutils.NODED_CERT_FILE
3417
    cert_path = pathutils.NODED_CERT_FILE
3418
    assert opts.ca_pem is None
3419
  else:
3420
    (_, key_path, cert_path) = _GetX509Filenames(pathutils.CRYPTO_KEYS_DIR,
3421
                                                 opts.key_name)
3422
    assert opts.ca_pem is not None
3423

    
3424
  for i in [key_path, cert_path]:
3425
    if not os.path.exists(i):
3426
      _Fail("File '%s' does not exist" % i)
3427

    
3428
  status_dir = _CreateImportExportStatusDir("%s-%s" % (prefix, component))
3429
  try:
3430
    status_file = utils.PathJoin(status_dir, _IES_STATUS_FILE)
3431
    pid_file = utils.PathJoin(status_dir, _IES_PID_FILE)
3432
    ca_file = utils.PathJoin(status_dir, _IES_CA_FILE)
3433

    
3434
    if opts.ca_pem is None:
3435
      # Use server.pem
3436
      ca = utils.ReadFile(pathutils.NODED_CERT_FILE)
3437
    else:
3438
      ca = opts.ca_pem
3439

    
3440
    # Write CA file
3441
    utils.WriteFile(ca_file, data=ca, mode=0400)
3442

    
3443
    cmd = [
3444
      pathutils.IMPORT_EXPORT_DAEMON,
3445
      status_file, mode,
3446
      "--key=%s" % key_path,
3447
      "--cert=%s" % cert_path,
3448
      "--ca=%s" % ca_file,
3449
      ]
3450

    
3451
    if host:
3452
      cmd.append("--host=%s" % host)
3453

    
3454
    if port:
3455
      cmd.append("--port=%s" % port)
3456

    
3457
    if opts.ipv6:
3458
      cmd.append("--ipv6")
3459
    else:
3460
      cmd.append("--ipv4")
3461

    
3462
    if opts.compress:
3463
      cmd.append("--compress=%s" % opts.compress)
3464

    
3465
    if opts.magic:
3466
      cmd.append("--magic=%s" % opts.magic)
3467

    
3468
    if exp_size is not None:
3469
      cmd.append("--expected-size=%s" % exp_size)
3470

    
3471
    if cmd_prefix:
3472
      cmd.append("--cmd-prefix=%s" % cmd_prefix)
3473

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

    
3477
    if mode == constants.IEM_EXPORT:
3478
      # Retry connection a few times when connecting to remote peer
3479
      cmd.append("--connect-retries=%s" % constants.RIE_CONNECT_RETRIES)
3480
      cmd.append("--connect-timeout=%s" % constants.RIE_CONNECT_ATTEMPT_TIMEOUT)
3481
    elif opts.connect_timeout is not None:
3482
      assert mode == constants.IEM_IMPORT
3483
      # Overall timeout for establishing connection while listening
3484
      cmd.append("--connect-timeout=%s" % opts.connect_timeout)
3485

    
3486
    logfile = _InstanceLogName(prefix, instance.os, instance.name, component)
3487

    
3488
    # TODO: Once _InstanceLogName uses tempfile.mkstemp, StartDaemon has
3489
    # support for receiving a file descriptor for output
3490
    utils.StartDaemon(cmd, env=cmd_env, pidfile=pid_file,
3491
                      output=logfile)
3492

    
3493
    # The import/export name is simply the status directory name
3494
    return os.path.basename(status_dir)
3495

    
3496
  except Exception:
3497
    shutil.rmtree(status_dir, ignore_errors=True)
3498
    raise
3499

    
3500

    
3501
def GetImportExportStatus(names):
3502
  """Returns import/export daemon status.
3503

3504
  @type names: sequence
3505
  @param names: List of names
3506
  @rtype: List of dicts
3507
  @return: Returns a list of the state of each named import/export or None if a
3508
           status couldn't be read
3509

3510
  """
3511
  result = []
3512

    
3513
  for name in names:
3514
    status_file = utils.PathJoin(pathutils.IMPORT_EXPORT_DIR, name,
3515
                                 _IES_STATUS_FILE)
3516

    
3517
    try:
3518
      data = utils.ReadFile(status_file)
3519
    except EnvironmentError, err:
3520
      if err.errno != errno.ENOENT:
3521
        raise
3522
      data = None
3523

    
3524
    if not data:
3525
      result.append(None)
3526
      continue
3527

    
3528
    result.append(serializer.LoadJson(data))
3529

    
3530
  return result
3531

    
3532

    
3533
def AbortImportExport(name):
3534
  """Sends SIGTERM to a running import/export daemon.
3535

3536
  """
3537
  logging.info("Abort import/export %s", name)
3538

    
3539
  status_dir = utils.PathJoin(pathutils.IMPORT_EXPORT_DIR, name)
3540
  pid = utils.ReadLockedPidFile(utils.PathJoin(status_dir, _IES_PID_FILE))
3541

    
3542
  if pid:
3543
    logging.info("Import/export %s is running with PID %s, sending SIGTERM",
3544
                 name, pid)
3545
    utils.IgnoreProcessNotFound(os.kill, pid, signal.SIGTERM)
3546

    
3547

    
3548
def CleanupImportExport(name):
3549
  """Cleanup after an import or export.
3550

3551
  If the import/export daemon is still running it's killed. Afterwards the
3552
  whole status directory is removed.
3553

3554
  """
3555
  logging.info("Finalizing import/export %s", name)
3556

    
3557
  status_dir = utils.PathJoin(pathutils.IMPORT_EXPORT_DIR, name)
3558

    
3559
  pid = utils.ReadLockedPidFile(utils.PathJoin(status_dir, _IES_PID_FILE))
3560

    
3561
  if pid:
3562
    logging.info("Import/export %s is still running with PID %s",
3563
                 name, pid)
3564
    utils.KillProcess(pid, waitpid=False)
3565

    
3566
  shutil.rmtree(status_dir, ignore_errors=True)
3567

    
3568

    
3569
def _FindDisks(nodes_ip, disks):
3570
  """Sets the physical ID on disks and returns the block devices.
3571

3572
  """
3573
  # set the correct physical ID
3574
  my_name = netutils.Hostname.GetSysName()
3575
  for cf in disks:
3576
    cf.SetPhysicalID(my_name, nodes_ip)
3577

    
3578
  bdevs = []
3579

    
3580
  for cf in disks:
3581
    rd = _RecursiveFindBD(cf)
3582
    if rd is None:
3583
      _Fail("Can't find device %s", cf)
3584
    bdevs.append(rd)
3585
  return bdevs
3586

    
3587

    
3588
def DrbdDisconnectNet(nodes_ip, disks):
3589
  """Disconnects the network on a list of drbd devices.
3590

3591
  """
3592
  bdevs = _FindDisks(nodes_ip, disks)
3593

    
3594
  # disconnect disks
3595
  for rd in bdevs:
3596
    try:
3597
      rd.DisconnectNet()
3598
    except errors.BlockDeviceError, err:
3599
      _Fail("Can't change network configuration to standalone mode: %s",
3600
            err, exc=True)
3601

    
3602

    
3603
def DrbdAttachNet(nodes_ip, disks, instance_name, multimaster):
3604
  """Attaches the network on a list of drbd devices.
3605

3606
  """
3607
  bdevs = _FindDisks(nodes_ip, disks)
3608

    
3609
  if multimaster:
3610
    for idx, rd in enumerate(bdevs):
3611
      try:
3612
        _SymlinkBlockDev(instance_name, rd.dev_path, idx)
3613
      except EnvironmentError, err:
3614
        _Fail("Can't create symlink: %s", err)
3615
  # reconnect disks, switch to new master configuration and if
3616
  # needed primary mode
3617
  for rd in bdevs:
3618
    try:
3619
      rd.AttachNet(multimaster)
3620
    except errors.BlockDeviceError, err:
3621
      _Fail("Can't change network configuration: %s", err)
3622

    
3623
  # wait until the disks are connected; we need to retry the re-attach
3624
  # if the device becomes standalone, as this might happen if the one
3625
  # node disconnects and reconnects in a different mode before the
3626
  # other node reconnects; in this case, one or both of the nodes will
3627
  # decide it has wrong configuration and switch to standalone
3628

    
3629
  def _Attach():
3630
    all_connected = True
3631

    
3632
    for rd in bdevs:
3633
      stats = rd.GetProcStatus()
3634

    
3635
      all_connected = (all_connected and
3636
                       (stats.is_connected or stats.is_in_resync))
3637

    
3638
      if stats.is_standalone:
3639
        # peer had different config info and this node became
3640
        # standalone, even though this should not happen with the
3641
        # new staged way of changing disk configs
3642
        try:
3643
          rd.AttachNet(multimaster)
3644
        except errors.BlockDeviceError, err:
3645
          _Fail("Can't change network configuration: %s", err)
3646

    
3647
    if not all_connected:
3648
      raise utils.RetryAgain()
3649

    
3650
  try:
3651
    # Start with a delay of 100 miliseconds and go up to 5 seconds
3652
    utils.Retry(_Attach, (0.1, 1.5, 5.0), 2 * 60)
3653
  except utils.RetryTimeout:
3654
    _Fail("Timeout in disk reconnecting")
3655

    
3656
  if multimaster:
3657
    # change to primary mode
3658
    for rd in bdevs:
3659
      try:
3660
        rd.Open()
3661
      except errors.BlockDeviceError, err:
3662
        _Fail("Can't change to primary mode: %s", err)
3663

    
3664

    
3665
def DrbdWaitSync(nodes_ip, disks):
3666
  """Wait until DRBDs have synchronized.
3667

3668
  """
3669
  def _helper(rd):
3670
    stats = rd.GetProcStatus()
3671
    if not (stats.is_connected or stats.is_in_resync):
3672
      raise utils.RetryAgain()
3673
    return stats
3674

    
3675
  bdevs = _FindDisks(nodes_ip, disks)
3676

    
3677
  min_resync = 100
3678
  alldone = True
3679
  for rd in bdevs:
3680
    try:
3681
      # poll each second for 15 seconds
3682
      stats = utils.Retry(_helper, 1, 15, args=[rd])
3683
    except utils.RetryTimeout:
3684
      stats = rd.GetProcStatus()
3685
      # last check
3686
      if not (stats.is_connected or stats.is_in_resync):
3687
        _Fail("DRBD device %s is not in sync: stats=%s", rd, stats)
3688
    alldone = alldone and (not stats.is_in_resync)
3689
    if stats.sync_percent is not None:
3690
      min_resync = min(min_resync, stats.sync_percent)
3691

    
3692
  return (alldone, min_resync)
3693

    
3694

    
3695
def GetDrbdUsermodeHelper():
3696
  """Returns DRBD usermode helper currently configured.
3697

3698
  """
3699
  try:
3700
    return bdev.BaseDRBD.GetUsermodeHelper()
3701
  except errors.BlockDeviceError, err:
3702
    _Fail(str(err))
3703

    
3704

    
3705
def PowercycleNode(hypervisor_type):
3706
  """Hard-powercycle the node.
3707

3708
  Because we need to return first, and schedule the powercycle in the
3709
  background, we won't be able to report failures nicely.
3710

3711
  """
3712
  hyper = hypervisor.GetHypervisor(hypervisor_type)
3713
  try:
3714
    pid = os.fork()
3715
  except OSError:
3716
    # if we can't fork, we'll pretend that we're in the child process
3717
    pid = 0
3718
  if pid > 0:
3719
    return "Reboot scheduled in 5 seconds"
3720
  # ensure the child is running on ram
3721
  try:
3722
    utils.Mlockall()
3723
  except Exception: # pylint: disable=W0703
3724
    pass
3725
  time.sleep(5)
3726
  hyper.PowercycleNode()
3727

    
3728

    
3729
def _VerifyRestrictedCmdName(cmd):
3730
  """Verifies a restricted command name.
3731

3732
  @type cmd: string
3733
  @param cmd: Command name
3734
  @rtype: tuple; (boolean, string or None)
3735
  @return: The tuple's first element is the status; if C{False}, the second
3736
    element is an error message string, otherwise it's C{None}
3737

3738
  """
3739
  if not cmd.strip():
3740
    return (False, "Missing command name")
3741

    
3742
  if os.path.basename(cmd) != cmd:
3743
    return (False, "Invalid command name")
3744

    
3745
  if not constants.EXT_PLUGIN_MASK.match(cmd):
3746
    return (False, "Command name contains forbidden characters")
3747

    
3748
  return (True, None)
3749

    
3750

    
3751
def _CommonRestrictedCmdCheck(path, owner):
3752
  """Common checks for restricted command file system directories and files.
3753

3754
  @type path: string
3755
  @param path: Path to check
3756
  @param owner: C{None} or tuple containing UID and GID
3757
  @rtype: tuple; (boolean, string or C{os.stat} result)
3758
  @return: The tuple's first element is the status; if C{False}, the second
3759
    element is an error message string, otherwise it's the result of C{os.stat}
3760

3761
  """
3762
  if owner is None:
3763
    # Default to root as owner
3764
    owner = (0, 0)
3765

    
3766
  try:
3767
    st = os.stat(path)
3768
  except EnvironmentError, err:
3769
    return (False, "Can't stat(2) '%s': %s" % (path, err))
3770

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

    
3774
  if (st.st_uid, st.st_gid) != owner:
3775
    (owner_uid, owner_gid) = owner
3776
    return (False, "'%s' is not owned by %s:%s" % (path, owner_uid, owner_gid))
3777

    
3778
  return (True, st)
3779

    
3780

    
3781
def _VerifyRestrictedCmdDirectory(path, _owner=None):
3782
  """Verifies restricted command directory.
3783

3784
  @type path: string
3785
  @param path: Path to check
3786
  @rtype: tuple; (boolean, string or None)
3787
  @return: The tuple's first element is the status; if C{False}, the second
3788
    element is an error message string, otherwise it's C{None}
3789

3790
  """
3791
  (status, value) = _CommonRestrictedCmdCheck(path, _owner)
3792

    
3793
  if not status:
3794
    return (False, value)
3795

    
3796
  if not stat.S_ISDIR(value.st_mode):
3797
    return (False, "Path '%s' is not a directory" % path)
3798

    
3799
  return (True, None)
3800

    
3801

    
3802
def _VerifyRestrictedCmd(path, cmd, _owner=None):
3803
  """Verifies a whole restricted command and returns its executable filename.
3804

3805
  @type path: string
3806
  @param path: Directory containing restricted commands
3807
  @type cmd: string
3808
  @param cmd: Command name
3809
  @rtype: tuple; (boolean, string)
3810
  @return: The tuple's first element is the status; if C{False}, the second
3811
    element is an error message string, otherwise the second element is the
3812
    absolute path to the executable
3813

3814
  """
3815
  executable = utils.PathJoin(path, cmd)
3816

    
3817
  (status, msg) = _CommonRestrictedCmdCheck(executable, _owner)
3818

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

    
3822
  if not utils.IsExecutable(executable):
3823
    return (False, "access(2) thinks '%s' can't be executed" % executable)
3824

    
3825
  return (True, executable)
3826

    
3827

    
3828
def _PrepareRestrictedCmd(path, cmd,
3829
                          _verify_dir=_VerifyRestrictedCmdDirectory,
3830
                          _verify_name=_VerifyRestrictedCmdName,
3831
                          _verify_cmd=_VerifyRestrictedCmd):
3832
  """Performs a number of tests on a restricted command.
3833

3834
  @type path: string
3835
  @param path: Directory containing restricted commands
3836
  @type cmd: string
3837
  @param cmd: Command name
3838
  @return: Same as L{_VerifyRestrictedCmd}
3839

3840
  """
3841
  # Verify the directory first
3842
  (status, msg) = _verify_dir(path)
3843
  if status:
3844
    # Check command if everything was alright
3845
    (status, msg) = _verify_name(cmd)
3846

    
3847
  if not status:
3848
    return (False, msg)
3849

    
3850
  # Check actual executable
3851
  return _verify_cmd(path, cmd)
3852

    
3853

    
3854
def RunRestrictedCmd(cmd,
3855
                     _lock_timeout=_RCMD_LOCK_TIMEOUT,
3856
                     _lock_file=pathutils.RESTRICTED_COMMANDS_LOCK_FILE,
3857
                     _path=pathutils.RESTRICTED_COMMANDS_DIR,
3858
                     _sleep_fn=time.sleep,
3859
                     _prepare_fn=_PrepareRestrictedCmd,
3860
                     _runcmd_fn=utils.RunCmd,
3861
                     _enabled=constants.ENABLE_RESTRICTED_COMMANDS):
3862
  """Executes a restricted command after performing strict tests.
3863

3864
  @type cmd: string
3865
  @param cmd: Command name
3866
  @rtype: string
3867
  @return: Command output
3868
  @raise RPCFail: In case of an error
3869

3870
  """
3871
  logging.info("Preparing to run restricted command '%s'", cmd)
3872

    
3873
  if not _enabled:
3874
    _Fail("Restricted commands disabled at configure time")
3875

    
3876
  lock = None
3877
  try:
3878
    cmdresult = None
3879
    try:
3880
      lock = utils.FileLock.Open(_lock_file)
3881
      lock.Exclusive(blocking=True, timeout=_lock_timeout)
3882

    
3883
      (status, value) = _prepare_fn(_path, cmd)
3884

    
3885
      if status:
3886
        cmdresult = _runcmd_fn([value], env={}, reset_env=True,
3887
                               postfork_fn=lambda _: lock.Unlock())
3888
      else:
3889
        logging.error(value)
3890
    except Exception: # pylint: disable=W0703
3891
      # Keep original error in log
3892
      logging.exception("Caught exception")
3893

    
3894
    if cmdresult is None:
3895
      logging.info("Sleeping for %0.1f seconds before returning",
3896
                   _RCMD_INVALID_DELAY)
3897
      _sleep_fn(_RCMD_INVALID_DELAY)
3898

    
3899
      # Do not include original error message in returned error
3900
      _Fail("Executing command '%s' failed" % cmd)
3901
    elif cmdresult.failed or cmdresult.fail_reason:
3902
      _Fail("Restricted command '%s' failed: %s; output: %s",
3903
            cmd, cmdresult.fail_reason, cmdresult.output)
3904
    else:
3905
      return cmdresult.output
3906
  finally:
3907
    if lock is not None:
3908
      # Release lock at last
3909
      lock.Close()
3910
      lock = None
3911

    
3912

    
3913
def SetWatcherPause(until, _filename=pathutils.WATCHER_PAUSEFILE):
3914
  """Creates or removes the watcher pause file.
3915

3916
  @type until: None or number
3917
  @param until: Unix timestamp saying until when the watcher shouldn't run
3918

3919
  """
3920
  if until is None:
3921
    logging.info("Received request to no longer pause watcher")
3922
    utils.RemoveFile(_filename)
3923
  else:
3924
    logging.info("Received request to pause watcher until %s", until)
3925

    
3926
    if not ht.TNumber(until):
3927
      _Fail("Duration must be numeric")
3928

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

    
3931

    
3932
class HooksRunner(object):
3933
  """Hook runner.
3934

3935
  This class is instantiated on the node side (ganeti-noded) and not
3936
  on the master side.
3937

3938
  """
3939
  def __init__(self, hooks_base_dir=None):
3940
    """Constructor for hooks runner.
3941

3942
    @type hooks_base_dir: str or None
3943
    @param hooks_base_dir: if not None, this overrides the
3944
        L{pathutils.HOOKS_BASE_DIR} (useful for unittests)
3945

3946
    """
3947
    if hooks_base_dir is None:
3948
      hooks_base_dir = pathutils.HOOKS_BASE_DIR
3949
    # yeah, _BASE_DIR is not valid for attributes, we use it like a
3950
    # constant
3951
    self._BASE_DIR = hooks_base_dir # pylint: disable=C0103
3952

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

3956
    """
3957
    assert len(node_list) == 1
3958
    node = node_list[0]
3959
    _, myself = ssconf.GetMasterAndMyself()
3960
    assert node == myself
3961

    
3962
    results = self.RunHooks(hpath, phase, env)
3963

    
3964
    # Return values in the form expected by HooksMaster
3965
    return {node: (None, False, results)}
3966

    
3967
  def RunHooks(self, hpath, phase, env):
3968
    """Run the scripts in the hooks directory.
3969

3970
    @type hpath: str
3971
    @param hpath: the path to the hooks directory which
3972
        holds the scripts
3973
    @type phase: str
3974
    @param phase: either L{constants.HOOKS_PHASE_PRE} or
3975
        L{constants.HOOKS_PHASE_POST}
3976
    @type env: dict
3977
    @param env: dictionary with the environment for the hook
3978
    @rtype: list
3979
    @return: list of 3-element tuples:
3980
      - script path
3981
      - script result, either L{constants.HKR_SUCCESS} or
3982
        L{constants.HKR_FAIL}
3983
      - output of the script
3984

3985
    @raise errors.ProgrammerError: for invalid input
3986
        parameters
3987

3988
    """
3989
    if phase == constants.HOOKS_PHASE_PRE:
3990
      suffix = "pre"
3991
    elif phase == constants.HOOKS_PHASE_POST:
3992
      suffix = "post"
3993
    else:
3994
      _Fail("Unknown hooks phase '%s'", phase)
3995

    
3996
    subdir = "%s-%s.d" % (hpath, suffix)
3997
    dir_name = utils.PathJoin(self._BASE_DIR, subdir)
3998

    
3999
    results = []
4000

    
4001
    if not os.path.isdir(dir_name):
4002
      # for non-existing/non-dirs, we simply exit instead of logging a
4003
      # warning at every operation
4004
      return results
4005

    
4006
    runparts_results = utils.RunParts(dir_name, env=env, reset_env=True)
4007

    
4008
    for (relname, relstatus, runresult) in runparts_results:
4009
      if relstatus == constants.RUNPARTS_SKIP:
4010
        rrval = constants.HKR_SKIP
4011
        output = ""
4012
      elif relstatus == constants.RUNPARTS_ERR:
4013
        rrval = constants.HKR_FAIL
4014
        output = "Hook script execution error: %s" % runresult
4015
      elif relstatus == constants.RUNPARTS_RUN:
4016
        if runresult.failed:
4017
          rrval = constants.HKR_FAIL
4018
        else:
4019
          rrval = constants.HKR_SUCCESS
4020
        output = utils.SafeEncode(runresult.output.strip())
4021
      results.append(("%s/%s" % (subdir, relname), rrval, output))
4022

    
4023
    return results
4024

    
4025

    
4026
class IAllocatorRunner(object):
4027
  """IAllocator runner.
4028

4029
  This class is instantiated on the node side (ganeti-noded) and not on
4030
  the master side.
4031

4032
  """
4033
  @staticmethod
4034
  def Run(name, idata):
4035
    """Run an iallocator script.
4036

4037
    @type name: str
4038
    @param name: the iallocator script name
4039
    @type idata: str
4040
    @param idata: the allocator input data
4041

4042
    @rtype: tuple
4043
    @return: two element tuple of:
4044
       - status
4045
       - either error message or stdout of allocator (for success)
4046

4047
    """
4048
    alloc_script = utils.FindFile(name, constants.IALLOCATOR_SEARCH_PATH,
4049
                                  os.path.isfile)
4050
    if alloc_script is None:
4051
      _Fail("iallocator module '%s' not found in the search path", name)
4052

    
4053
    fd, fin_name = tempfile.mkstemp(prefix="ganeti-iallocator.")
4054
    try:
4055
      os.write(fd, idata)
4056
      os.close(fd)
4057
      result = utils.RunCmd([alloc_script, fin_name])
4058
      if result.failed:
4059
        _Fail("iallocator module '%s' failed: %s, output '%s'",
4060
              name, result.fail_reason, result.output)
4061
    finally:
4062
      os.unlink(fin_name)
4063

    
4064
    return result.stdout
4065

    
4066

    
4067
class DevCacheManager(object):
4068
  """Simple class for managing a cache of block device information.
4069

4070
  """
4071
  _DEV_PREFIX = "/dev/"
4072
  _ROOT_DIR = pathutils.BDEV_CACHE_DIR
4073

    
4074
  @classmethod
4075
  def _ConvertPath(cls, dev_path):
4076
    """Converts a /dev/name path to the cache file name.
4077

4078
    This replaces slashes with underscores and strips the /dev
4079
    prefix. It then returns the full path to the cache file.
4080

4081
    @type dev_path: str
4082
    @param dev_path: the C{/dev/} path name
4083
    @rtype: str
4084
    @return: the converted path name
4085

4086
    """
4087
    if dev_path.startswith(cls._DEV_PREFIX):
4088
      dev_path = dev_path[len(cls._DEV_PREFIX):]
4089
    dev_path = dev_path.replace("/", "_")
4090
    fpath = utils.PathJoin(cls._ROOT_DIR, "bdev_%s" % dev_path)
4091
    return fpath
4092

    
4093
  @classmethod
4094
  def UpdateCache(cls, dev_path, owner, on_primary, iv_name):
4095
    """Updates the cache information for a given device.
4096

4097
    @type dev_path: str
4098
    @param dev_path: the pathname of the device
4099
    @type owner: str
4100
    @param owner: the owner (instance name) of the device
4101
    @type on_primary: bool
4102
    @param on_primary: whether this is the primary
4103
        node nor not
4104
    @type iv_name: str
4105
    @param iv_name: the instance-visible name of the
4106
        device, as in objects.Disk.iv_name
4107

4108
    @rtype: None
4109

4110
    """
4111
    if dev_path is None:
4112
      logging.error("DevCacheManager.UpdateCache got a None dev_path")
4113
      return
4114
    fpath = cls._ConvertPath(dev_path)
4115
    if on_primary:
4116
      state = "primary"
4117
    else:
4118
      state = "secondary"
4119