#
#
-# Copyright (C) 2006, 2007, 2008, 2009, 2010 Google Inc.
+# Copyright (C) 2006, 2007, 2008, 2009, 2010, 2011 Google Inc.
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
_IES_PID_FILE = "pid"
_IES_CA_FILE = "ca"
+#: Valid LVS output line regex
+_LVSLINE_REGEX = re.compile("^ *([^|]+)\|([^|]+)\|([0-9.]+)\|([^|]{6})\|?$")
+
class RPCFail(Exception):
"""Class denoting RPC failure.
"""
outputarray = {}
- vginfo = bdev.LogicalVolume.GetVGInfo([vgname])
- vg_free = vg_size = None
- if vginfo:
- vg_free = int(round(vginfo[0][0], 0))
- vg_size = int(round(vginfo[0][1], 0))
-
- outputarray['vg_size'] = vg_size
- outputarray['vg_free'] = vg_free
-
- hyper = hypervisor.GetHypervisor(hypervisor_type)
- hyp_info = hyper.GetNodeInfo()
- if hyp_info is not None:
- outputarray.update(hyp_info)
+ if vgname is not None:
+ vginfo = bdev.LogicalVolume.GetVGInfo([vgname])
+ vg_free = vg_size = None
+ if vginfo:
+ vg_free = int(round(vginfo[0][0], 0))
+ vg_size = int(round(vginfo[0][1], 0))
+ outputarray['vg_size'] = vg_size
+ outputarray['vg_free'] = vg_free
+
+ if hypervisor_type is not None:
+ hyper = hypervisor.GetHypervisor(hypervisor_type)
+ hyp_info = hyper.GetNodeInfo()
+ if hyp_info is not None:
+ outputarray.update(hyp_info)
outputarray["bootid"] = utils.ReadFile(_BOOT_ID_PATH, size=128).rstrip("\n")
val = "Error while checking hypervisor: %s" % str(err)
tmp[hv_name] = val
+ if constants.NV_HVPARAMS in what and vm_capable:
+ result[constants.NV_HVPARAMS] = tmp = []
+ for source, hv_name, hvparms in what[constants.NV_HVPARAMS]:
+ try:
+ logging.info("Validating hv %s, %s", hv_name, hvparms)
+ hypervisor.GetHypervisor(hv_name).ValidateParameters(hvparms)
+ except errors.HypervisorError, err:
+ tmp.append((source, hv_name, str(err)))
+
if constants.NV_FILELIST in what:
result[constants.NV_FILELIST] = utils.FingerprintFiles(
what[constants.NV_FILELIST])
result[constants.NV_MASTERIP] = netutils.TcpPing(master_ip, port,
source=source)
+ if constants.NV_OOB_PATHS in what:
+ result[constants.NV_OOB_PATHS] = tmp = []
+ for path in what[constants.NV_OOB_PATHS]:
+ try:
+ st = os.stat(path)
+ except OSError, err:
+ tmp.append("error stating out of band helper: %s" % err)
+ else:
+ if stat.S_ISREG(st.st_mode):
+ if stat.S_IMODE(st.st_mode) & stat.S_IXUSR:
+ tmp.append(None)
+ else:
+ tmp.append("out of band helper %s is not executable" % path)
+ else:
+ tmp.append("out of band helper %s is not a file" % path)
+
if constants.NV_LVLIST in what and vm_capable:
try:
- val = GetVolumeList(what[constants.NV_LVLIST])
+ val = GetVolumeList(utils.ListVolumeGroups().keys())
except RPCFail, err:
val = str(err)
result[constants.NV_LVLIST] = val
return result
-def GetVolumeList(vg_name):
+def GetVolumeList(vg_names):
"""Compute list of logical volumes and their size.
- @type vg_name: str
- @param vg_name: the volume group whose LVs we should list
+ @type vg_names: list
+ @param vg_names: the volume groups whose LVs we should list, or
+ empty for all volume groups
@rtype: dict
@return:
dictionary of all partions (key) with value being a tuple of
their size (in MiB), inactive and online status::
- {'test1': ('20.06', True, True)}
+ {'xenvg/test1': ('20.06', True, True)}
in case of errors, a string is returned with the error
details.
"""
lvs = {}
sep = '|'
+ if not vg_names:
+ vg_names = []
result = utils.RunCmd(["lvs", "--noheadings", "--units=m", "--nosuffix",
"--separator=%s" % sep,
- "-olv_name,lv_size,lv_attr", vg_name])
+ "-ovg_name,lv_name,lv_size,lv_attr"] + vg_names)
if result.failed:
_Fail("Failed to list logical volumes, lvs output: %s", result.output)
- valid_line_re = re.compile("^ *([^|]+)\|([0-9.]+)\|([^|]{6})\|?$")
for line in result.stdout.splitlines():
line = line.strip()
- match = valid_line_re.match(line)
+ match = _LVSLINE_REGEX.match(line)
if not match:
logging.error("Invalid line returned from lvs output: '%s'", line)
continue
- name, size, attr = match.groups()
+ vg_name, name, size, attr = match.groups()
inactive = attr[4] == '-'
online = attr[5] == 'o'
virtual = attr[0] == 'v'
# we don't want to report such volumes as existing, since they
# don't really hold data
continue
- lvs[name] = (size, inactive, online)
+ lvs[vg_name+"/"+name] = (size, inactive, online)
return lvs
def _GetBlockDevSymlinkPath(instance_name, idx):
- return utils.PathJoin(constants.DISK_LINKS_DIR,
- "%s:%d" % (instance_name, idx))
+ return utils.PathJoin(constants.DISK_LINKS_DIR, "%s%s%d" %
+ (instance_name, constants.DISK_SEPARATOR, idx))
def _SymlinkBlockDev(instance_name, device_path, idx):
_WipeDevice(rdev.dev_path, offset, size)
+def BlockdevPauseResumeSync(disks, pause):
+ """Pause or resume the sync of the block device.
+
+ @type disks: list of L{objects.Disk}
+ @param disks: the disks object we want to pause/resume
+ @type pause: bool
+ @param pause: Wheater to pause or resume
+
+ """
+ success = []
+ for disk in disks:
+ try:
+ rdev = _RecursiveFindBD(disk)
+ except errors.BlockDeviceError:
+ rdev = None
+
+ if not rdev:
+ success.append((False, ("Cannot change sync for device %s:"
+ " device not found" % disk.iv_name)))
+ continue
+
+ result = rdev.PauseResumeSync(pause)
+
+ if result:
+ success.append((result, None))
+ else:
+ if pause:
+ msg = "Pause"
+ else:
+ msg = "Resume"
+ success.append((result, "%s for device %s failed" % (msg, disk.iv_name)))
+
+ return success
+
+
def BlockdevRemove(disk):
"""Remove a block device.
return result
-def BlockdevAssemble(disk, owner, as_primary):
+def BlockdevAssemble(disk, owner, as_primary, idx):
"""Activate a block device for an instance.
This is a wrapper over _RecursiveAssembleBD.
if isinstance(result, bdev.BlockDev):
# pylint: disable-msg=E1103
result = result.dev_path
+ if as_primary:
+ _SymlinkBlockDev(owner, result, idx)
except errors.BlockDeviceError, err:
_Fail("Error while assembling disk: %s", err, exc=True)
+ except OSError, err:
+ _Fail("Error while symlinking disk: %s", err, exc=True)
return result
@type disks: list of L{objects.Disk}
@param disks: the list of disks which we should query
@rtype: disk
- @return:
- a list of (mirror_done, estimated_time) tuples, which
- are the result of L{bdev.BlockDev.CombinedSyncStatus}
+ @return: List of L{objects.BlockDevStatus}, one for each disk
@raise errors.BlockDeviceError: if any of the disks cannot be
found
return stats
+def BlockdevGetmirrorstatusMulti(disks):
+ """Get the mirroring status of a list of devices.
+
+ @type disks: list of L{objects.Disk}
+ @param disks: the list of disks which we should query
+ @rtype: disk
+ @return: List of tuples, (bool, status), one for each disk; bool denotes
+ success/failure, status is L{objects.BlockDevStatus} on success, string
+ otherwise
+
+ """
+ result = []
+ for disk in disks:
+ try:
+ rbd = _RecursiveFindBD(disk)
+ if rbd is None:
+ result.append((False, "Can't find device %s" % disk))
+ continue
+
+ status = rbd.CombinedSyncStatus()
+ except errors.BlockDeviceError, err:
+ logging.exception("Error while getting disk status")
+ result.append((False, str(err)))
+ else:
+ result.append((True, status))
+
+ assert len(disks) == len(result)
+
+ return result
+
+
def _RecursiveFindBD(disk):
"""Check if a device is activated.
atime=atime, mtime=mtime)
+def RunOob(oob_program, command, node, timeout):
+ """Executes oob_program with given command on given node.
+
+ @param oob_program: The path to the executable oob_program
+ @param command: The command to invoke on oob_program
+ @param node: The node given as an argument to the program
+ @param timeout: Timeout after which we kill the oob program
+
+ @return: stdout
+ @raise RPCFail: If execution fails for some reason
+
+ """
+ result = utils.RunCmd([oob_program, command, node], timeout=timeout)
+
+ if result.failed:
+ _Fail("'%s' failed with reason '%s'; output: %s", result.cmd,
+ result.fail_reason, result.output)
+
+ return result.stdout
+
+
def WriteSsconfFiles(values):
"""Update all ssconf files.
@type disk: L{objects.Disk}
@param disk: the disk to be snapshotted
@rtype: string
- @return: snapshot disk path
+ @return: snapshot disk ID as (vg, lv)
"""
if disk.dev_type == constants.LD_DRBD8:
config.set(constants.INISECT_EXP, 'timestamp', '%d' % int(time.time()))
config.set(constants.INISECT_EXP, 'source', instance.primary_node)
config.set(constants.INISECT_EXP, 'os', instance.os)
- config.set(constants.INISECT_EXP, 'compression', 'gzip')
+ config.set(constants.INISECT_EXP, "compression", "none")
config.add_section(constants.INISECT_INS)
config.set(constants.INISECT_INS, 'name', instance.name)
def JobQueueUpdate(file_name, content):
"""Updates a file in the queue directory.
- This is just a wrapper over L{utils.WriteFile}, with proper
+ This is just a wrapper over L{utils.io.WriteFile}, with proper
checking.
@type file_name: str
if port:
cmd.append("--port=%s" % port)
+ if opts.ipv6:
+ cmd.append("--ipv6")
+ else:
+ cmd.append("--ipv4")
+
if opts.compress:
cmd.append("--compress=%s" % opts.compress)
if cmd_suffix:
cmd.append("--cmd-suffix=%s" % cmd_suffix)
+ if mode == constants.IEM_EXPORT:
+ # Retry connection a few times when connecting to remote peer
+ cmd.append("--connect-retries=%s" % constants.RIE_CONNECT_RETRIES)
+ cmd.append("--connect-timeout=%s" % constants.RIE_CONNECT_ATTEMPT_TIMEOUT)
+ elif opts.connect_timeout is not None:
+ assert mode == constants.IEM_IMPORT
+ # Overall timeout for establishing connection while listening
+ cmd.append("--connect-timeout=%s" % opts.connect_timeout)
+
logfile = _InstanceLogName(prefix, instance.os, instance.name)
# TODO: Once _InstanceLogName uses tempfile.mkstemp, StartDaemon has
def RemoveCache(cls, dev_path):
"""Remove data for a dev_path.
- This is just a wrapper over L{utils.RemoveFile} with a converted
+ This is just a wrapper over L{utils.io.RemoveFile} with a converted
path name and logging.
@type dev_path: str