from ganeti import utils
from ganeti import errors
from ganeti import constants
+from ganeti import objects
+
+
+# Size of reads in _CanReadDevice
+_DEVICE_READ_SIZE = 128 * 1024
def _IgnoreError(fn, *args, **kwargs):
fn(*args, **kwargs)
return True
except errors.BlockDeviceError, err:
- logging.warning("Caught BlockDeviceError but ignoring: %s" % str(err))
+ logging.warning("Caught BlockDeviceError but ignoring: %s", str(err))
return False
raise errors.BlockDeviceError(msg)
+def _CanReadDevice(path):
+ """Check if we can read from the given device.
+
+ This tries to read the first 128k of the device.
+
+ """
+ try:
+ utils.ReadFile(path, size=_DEVICE_READ_SIZE)
+ return True
+ except EnvironmentError:
+ logging.warning("Can't read from device %s", path, exc_info=True)
+ return False
+
+
class BlockDev(object):
"""Block device abstract class.
data. This is only valid for some devices, the rest will always
return False (not degraded).
- @rtype: tuple
- @return: (sync_percent, estimated_time, is_degraded, ldisk)
+ @rtype: objects.BlockDevStatus
"""
- return None, None, False, False
-
+ return objects.BlockDevStatus(dev_path=self.dev_path,
+ major=self.major,
+ minor=self.minor,
+ sync_percent=None,
+ estimated_time=None,
+ is_degraded=False,
+ ldisk_status=constants.LDS_OKAY)
def CombinedSyncStatus(self):
"""Calculate the mirror status recursively for our children.
minimum percent and maximum time are calculated across our
children.
+ @rtype: objects.BlockDevStatus
+
"""
- min_percent, max_time, is_degraded, ldisk = self.GetSyncStatus()
+ status = self.GetSyncStatus()
+
+ min_percent = status.sync_percent
+ max_time = status.estimated_time
+ is_degraded = status.is_degraded
+ ldisk_status = status.ldisk_status
+
if self._children:
for child in self._children:
- c_percent, c_time, c_degraded, c_ldisk = child.GetSyncStatus()
+ child_status = child.GetSyncStatus()
+
if min_percent is None:
- min_percent = c_percent
- elif c_percent is not None:
- min_percent = min(min_percent, c_percent)
+ min_percent = child_status.sync_percent
+ elif child_status.sync_percent is not None:
+ min_percent = min(min_percent, child_status.sync_percent)
+
if max_time is None:
- max_time = c_time
- elif c_time is not None:
- max_time = max(max_time, c_time)
- is_degraded = is_degraded or c_degraded
- ldisk = ldisk or c_ldisk
- return min_percent, max_time, is_degraded, ldisk
+ max_time = child_status.estimated_time
+ elif child_status.estimated_time is not None:
+ max_time = max(max_time, child_status.estimated_time)
+
+ is_degraded = is_degraded or child_status.is_degraded
+
+ if ldisk_status is None:
+ ldisk_status = child_status.ldisk_status
+ elif child_status.ldisk_status is not None:
+ ldisk_status = max(ldisk_status, child_status.ldisk_status)
+
+ return objects.BlockDevStatus(dev_path=self.dev_path,
+ major=self.major,
+ minor=self.minor,
+ sync_percent=min_percent,
+ estimated_time=max_time,
+ is_degraded=is_degraded,
+ ldisk_status=ldisk_status)
def SetInfo(self, text):
"""
raise NotImplementedError
+ def GetActualSize(self):
+ """Return the actual disk size.
+
+ @note: the device needs to be active when this is called
+
+ """
+ assert self.attached, "BlockDevice not attached in GetActualSize()"
+ result = utils.RunCmd(["blockdev", "--getsize64", self.dev_path])
+ if result.failed:
+ _ThrowError("blockdev failed (%s): %s",
+ result.fail_reason, result.output)
+ try:
+ sz = int(result.output.strip())
+ except (ValueError, TypeError), err:
+ _ThrowError("Failed to parse blockdev output: %s", str(err))
+ return sz
+
def __repr__(self):
return ("<%s: unique_id: %s, children: %s, %s:%s, %s>" %
(self.__class__, self.unique_id, self._children,
"""Logical Volume block device.
"""
+ _VALID_NAME_RE = re.compile("^[a-zA-Z0-9+_.-]*$")
+ _INVALID_NAMES = frozenset([".", "..", "snapshot", "pvmove"])
+ _INVALID_SUBSTRINGS = frozenset(["_mlog", "_mimage"])
+
def __init__(self, unique_id, children, size):
"""Attaches to a LV device.
if not isinstance(unique_id, (tuple, list)) or len(unique_id) != 2:
raise ValueError("Invalid configuration data %s" % str(unique_id))
self._vg_name, self._lv_name = unique_id
- self.dev_path = "/dev/%s/%s" % (self._vg_name, self._lv_name)
+ self._ValidateName(self._vg_name)
+ self._ValidateName(self._lv_name)
+ self.dev_path = utils.PathJoin("/dev", self._vg_name, self._lv_name)
self._degraded = True
- self.major = self.minor = None
+ self.major = self.minor = self.pe_size = self.stripe_count = None
self.Attach()
@classmethod
raise errors.ProgrammerError("Invalid configuration data %s" %
str(unique_id))
vg_name, lv_name = unique_id
- pvs_info = cls.GetPVInfo(vg_name)
+ cls._ValidateName(vg_name)
+ cls._ValidateName(lv_name)
+ pvs_info = cls.GetPVInfo([vg_name])
if not pvs_info:
_ThrowError("Can't compute PV info for vg %s", vg_name)
pvs_info.sort()
pvs_info.reverse()
pvlist = [ pv[1] for pv in pvs_info ]
+ if utils.any(pvlist, lambda v: ":" in v):
+ _ThrowError("Some of your PVs have the invalid character ':' in their"
+ " name, this is not supported - please filter them out"
+ " in lvm.conf using either 'filter' or 'preferred_names'")
free_size = sum([ pv[0] for pv in pvs_info ])
current_pvs = len(pvlist)
stripes = min(current_pvs, constants.LVM_STRIPECOUNT)
return LogicalVolume(unique_id, children, size)
@staticmethod
- def GetPVInfo(vg_name):
+ def GetPVInfo(vg_names, filter_allocatable=True):
"""Get the free space info for PVs in a volume group.
- @param vg_name: the volume group name
+ @param vg_names: list of volume group names, if empty all will be returned
+ @param filter_allocatable: whether to skip over unallocatable PVs
@rtype: list
@return: list of tuples (free_space, name) with free_space in mebibytes
"""
+ sep = "|"
command = ["pvs", "--noheadings", "--nosuffix", "--units=m",
"-opv_name,vg_name,pv_free,pv_attr", "--unbuffered",
- "--separator=:"]
+ "--separator=%s" % sep ]
result = utils.RunCmd(command)
if result.failed:
logging.error("Can't get the PV information: %s - %s",
return None
data = []
for line in result.stdout.splitlines():
- fields = line.strip().split(':')
+ fields = line.strip().split(sep)
if len(fields) != 4:
logging.error("Can't parse pvs output: line '%s'", line)
return None
- # skip over pvs from another vg or ones which are not allocatable
- if fields[1] != vg_name or fields[3][0] != 'a':
+ # (possibly) skip over pvs which are not allocatable
+ if filter_allocatable and fields[3][0] != 'a':
+ continue
+ # (possibly) skip over pvs which are not in the right volume group(s)
+ if vg_names and fields[1] not in vg_names:
continue
- data.append((float(fields[2]), fields[0]))
+ data.append((float(fields[2]), fields[0], fields[1]))
return data
+ @classmethod
+ def _ValidateName(cls, name):
+ """Validates that a given name is valid as VG or LV name.
+
+ The list of valid characters and restricted names is taken out of
+ the lvm(8) manpage, with the simplification that we enforce both
+ VG and LV restrictions on the names.
+
+ """
+ if (not cls._VALID_NAME_RE.match(name) or
+ name in cls._INVALID_NAMES or
+ utils.any(cls._INVALID_SUBSTRINGS, lambda x: x in name)):
+ _ThrowError("Invalid LVM name '%s'", name)
+
def Remove(self):
"""Remove this logical volume.
if result.failed:
_ThrowError("Failed to rename the logical volume: %s", result.output)
self._lv_name = new_name
- self.dev_path = "/dev/%s/%s" % (self._vg_name, self._lv_name)
+ self.dev_path = utils.PathJoin("/dev", self._vg_name, self._lv_name)
def Attach(self):
"""Attach to an existing LV.
"""
self.attached = False
result = utils.RunCmd(["lvs", "--noheadings", "--separator=,",
- "-olv_attr,lv_kernel_major,lv_kernel_minor",
- self.dev_path])
+ "--units=m", "--nosuffix",
+ "-olv_attr,lv_kernel_major,lv_kernel_minor,"
+ "vg_extent_size,stripes", self.dev_path])
if result.failed:
logging.error("Can't find LV %s: %s, %s",
self.dev_path, result.fail_reason, result.output)
return False
- out = result.stdout.strip().rstrip(',')
+ # the output can (and will) have multiple lines for multi-segment
+ # LVs, as the 'stripes' parameter is a segment one, so we take
+ # only the last entry, which is the one we're interested in; note
+ # that with LVM2 anyway the 'stripes' value must be constant
+ # across segments, so this is a no-op actually
+ out = result.stdout.splitlines()
+ if not out: # totally empty result? splitlines() returns at least
+ # one line for any non-empty string
+ logging.error("Can't parse LVS output, no lines? Got '%s'", str(out))
+ return False
+ out = out[-1].strip().rstrip(',')
out = out.split(",")
- if len(out) != 3:
- logging.error("Can't parse LVS output, len(%s) != 3", str(out))
+ if len(out) != 5:
+ logging.error("Can't parse LVS output, len(%s) != 5", str(out))
return False
- status, major, minor = out[:3]
+ status, major, minor, pe_size, stripes = out
if len(status) != 6:
logging.error("lvs lv_attr is not 6 characters (%s)", status)
return False
try:
major = int(major)
minor = int(minor)
- except ValueError, err:
+ except (TypeError, ValueError), err:
logging.error("lvs major/minor cannot be parsed: %s", str(err))
+ try:
+ pe_size = int(float(pe_size))
+ except (TypeError, ValueError), err:
+ logging.error("Can't parse vg extent size: %s", err)
+ return False
+
+ try:
+ stripes = int(stripes)
+ except (TypeError, ValueError), err:
+ logging.error("Can't parse the number of stripes: %s", err)
+ return False
+
self.major = major
self.minor = minor
+ self.pe_size = pe_size
+ self.stripe_count = stripes
self._degraded = status[0] == 'v' # virtual volume, i.e. doesn't backing
# storage
self.attached = True
The status was already read in Attach, so we just return it.
- @rtype: tuple
- @return: (sync_percent, estimated_time, is_degraded, ldisk)
+ @rtype: objects.BlockDevStatus
"""
- return None, None, self._degraded, self._degraded
+ if self._degraded:
+ ldisk_status = constants.LDS_FAULTY
+ else:
+ ldisk_status = constants.LDS_OKAY
+
+ return objects.BlockDevStatus(dev_path=self.dev_path,
+ major=self.major,
+ minor=self.minor,
+ sync_percent=None,
+ estimated_time=None,
+ is_degraded=self._degraded,
+ ldisk_status=ldisk_status)
def Open(self, force=False):
"""Make the device ready for I/O.
snap = LogicalVolume((self._vg_name, snap_name), None, size)
_IgnoreError(snap.Remove)
- pvs_info = self.GetPVInfo(self._vg_name)
+ pvs_info = self.GetPVInfo([self._vg_name])
if not pvs_info:
_ThrowError("Can't compute PV info for vg %s", self._vg_name)
pvs_info.sort()
pvs_info.reverse()
- free_size, pv_name = pvs_info[0]
+ free_size, _, _ = pvs_info[0]
if free_size < size:
_ThrowError("Not enough free space: required %s,"
" available %s", size, free_size)
"""Grow the logical volume.
"""
+ if self.pe_size is None or self.stripe_count is None:
+ if not self.Attach():
+ _ThrowError("Can't attach to LV during Grow()")
+ full_stripe_size = self.pe_size * self.stripe_count
+ rest = amount % full_stripe_size
+ if rest != 0:
+ amount += full_stripe_size - rest
# we try multiple algorithms since the 'best' ones might not have
# space available in the right place, but later ones might (since
# they have less constraints); also note that only recent LVM
self.est_time = None
-class BaseDRBD(BlockDev):
+class BaseDRBD(BlockDev): # pylint: disable-msg=W0223
"""Base DRBD class.
This class contains a few bits of common functionality between the
"""
_VERSION_RE = re.compile(r"^version: (\d+)\.(\d+)\.(\d+)"
r" \(api:(\d+)/proto:(\d+)(?:-(\d+))?\)")
+ _VALID_LINE_RE = re.compile("^ *([0-9]+): cs:([^ ]+).*$")
+ _UNUSED_LINE_RE = re.compile("^ *([0-9]+): cs:Unconfigured$")
_DRBD_MAJOR = 147
_ST_UNCONFIGURED = "Unconfigured"
"""
try:
- stat = open(filename, "r")
- try:
- data = stat.read().splitlines()
- finally:
- stat.close()
+ data = utils.ReadFile(filename).splitlines()
except EnvironmentError, err:
if err.errno == errno.ENOENT:
_ThrowError("The file %s cannot be opened, check if the module"
_ThrowError("Can't read any data from %s", filename)
return data
- @staticmethod
- def _MassageProcData(data):
+ @classmethod
+ def _MassageProcData(cls, data):
"""Transform the output of _GetProdData into a nicer form.
@return: a dictionary of minor: joined lines from /proc/drbd
for that minor
"""
- lmatch = re.compile("^ *([0-9]+):.*$")
results = {}
old_minor = old_line = None
for line in data:
- lresult = lmatch.match(line)
+ if not line: # completely empty lines, as can be returned by drbd8.0+
+ continue
+ lresult = cls._VALID_LINE_RE.match(line)
if lresult is not None:
if old_minor is not None:
results[old_minor] = old_line
data = cls._GetProcData()
used_devs = {}
- valid_line = re.compile("^ *([0-9]+): cs:([^ ]+).*$")
for line in data:
- match = valid_line.match(line)
+ match = cls._VALID_LINE_RE.match(line)
if not match:
continue
minor = int(match.group(1))
result.fail_reason, result.output)
try:
sectors = int(result.stdout)
- except ValueError:
+ except (TypeError, ValueError):
_ThrowError("Invalid output from blockdev: '%s'", result.stdout)
bytes = sectors * 512
if bytes < 128 * 1024 * 1024: # less than 128MiB
def __init__(self, unique_id, children, size):
if children and children.count(None) > 0:
children = []
+ if len(children) not in (0, 2):
+ raise ValueError("Invalid configuration data %s" % str(children))
+ if not isinstance(unique_id, (tuple, list)) or len(unique_id) != 6:
+ raise ValueError("Invalid configuration data %s" % str(unique_id))
+ (self._lhost, self._lport,
+ self._rhost, self._rport,
+ self._aminor, self._secret) = unique_id
+ if children:
+ if not _CanReadDevice(children[1].dev_path):
+ logging.info("drbd%s: Ignoring unreadable meta device", self._aminor)
+ children = []
super(DRBD8, self).__init__(unique_id, children, size)
self.major = self._DRBD_MAJOR
version = self._GetVersion()
" usage: kernel is %s.%s, ganeti wants 8.x",
version['k_major'], version['k_minor'])
- if len(children) not in (0, 2):
- raise ValueError("Invalid configuration data %s" % str(children))
- if not isinstance(unique_id, (tuple, list)) or len(unique_id) != 6:
- raise ValueError("Invalid configuration data %s" % str(unique_id))
- (self._lhost, self._lport,
- self._rhost, self._rport,
- self._aminor, self._secret) = unique_id
if (self._lhost is not None and self._lhost == self._rhost and
self._lport == self._rport):
raise ValueError("Invalid configuration data, same local/remote %s" %
"""
data = cls._GetProcData()
- unused_line = re.compile("^ *([0-9]+): cs:Unconfigured$")
- used_line = re.compile("^ *([0-9]+): cs:")
highest = None
for line in data:
- match = unused_line.match(line)
+ match = cls._UNUSED_LINE_RE.match(line)
if match:
return int(match.group(1))
- match = used_line.match(line)
+ match = cls._VALID_LINE_RE.match(line)
if match:
minor = int(match.group(1))
highest = max(highest, minor)
"""
args = ["drbdsetup", cls._DevPath(minor), "disk",
backend, meta, "0",
- "-d", "%sm" % size,
"-e", "detach",
"--create-device"]
+ if size:
+ args.extend(["-d", "%sm" % size])
+ if not constants.DRBD_BARRIERS: # disable barriers, if configured so
+ version = cls._GetVersion()
+ # various DRBD versions support different disk barrier options;
+ # what we aim here is to revert back to the 'drain' method of
+ # disk flushes and to disable metadata barriers, in effect going
+ # back to pre-8.0.7 behaviour
+ vmaj = version['k_major']
+ vmin = version['k_minor']
+ vrel = version['k_point']
+ assert vmaj == 8
+ if vmin == 0: # 8.0.x
+ if vrel >= 12:
+ args.extend(['-i', '-m'])
+ elif vmin == 2: # 8.2.x
+ if vrel >= 7:
+ args.extend(['-i', '-m'])
+ elif vmaj >= 3: # 8.3.x or newer
+ args.extend(['-i', '-a', 'm'])
result = utils.RunCmd(args)
if result.failed:
_ThrowError("drbd%d: can't attach local disk: %s", minor, result.output)
_ThrowError("drbd%d: can't setup network: %s - %s",
minor, result.fail_reason, result.output)
- timeout = time.time() + 10
- ok = False
- while time.time() < timeout:
+ def _CheckNetworkConfig():
info = cls._GetDevInfo(cls._GetShowData(minor))
if not "local_addr" in info or not "remote_addr" in info:
- time.sleep(1)
- continue
+ raise utils.RetryAgain()
+
if (info["local_addr"] != (lhost, lport) or
info["remote_addr"] != (rhost, rport)):
- time.sleep(1)
- continue
- ok = True
- break
- if not ok:
+ raise utils.RetryAgain()
+
+ try:
+ utils.Retry(_CheckNetworkConfig, 1.0, 10.0)
+ except utils.RetryTimeout:
_ThrowError("drbd%d: timeout while configuring network", minor)
def AddChildren(self, devices):
We compute the ldisk parameter based on whether we have a local
disk or not.
- @rtype: tuple
- @return: (sync_percent, estimated_time, is_degraded, ldisk)
+ @rtype: objects.BlockDevStatus
"""
if self.minor is None and not self.Attach():
_ThrowError("drbd%d: can't Attach() in GetSyncStatus", self._aminor)
+
stats = self.GetProcStatus()
- ldisk = not stats.is_disk_uptodate
- is_degraded = not stats.is_connected
- return stats.sync_percent, stats.est_time, is_degraded or ldisk, ldisk
+ is_degraded = not stats.is_connected or not stats.is_disk_uptodate
+
+ if stats.is_disk_uptodate:
+ ldisk_status = constants.LDS_OKAY
+ elif stats.is_diskless:
+ ldisk_status = constants.LDS_FAULTY
+ else:
+ ldisk_status = constants.LDS_UNKNOWN
+
+ return objects.BlockDevStatus(dev_path=self.dev_path,
+ major=self.major,
+ minor=self.minor,
+ sync_percent=stats.sync_percent,
+ estimated_time=stats.est_time,
+ is_degraded=is_degraded,
+ ldisk_status=ldisk_status)
def Open(self, force=False):
"""Make the local state primary.
_ThrowError("drbd%d: DRBD disk missing network info in"
" DisconnectNet()", self.minor)
- ever_disconnected = _IgnoreError(self._ShutdownNet, self.minor)
- timeout_limit = time.time() + self._NET_RECONFIG_TIMEOUT
- sleep_time = 0.100 # we start the retry time at 100 milliseconds
- while time.time() < timeout_limit:
- status = self.GetProcStatus()
- if status.is_standalone:
- break
- # retry the disconnect, it seems possible that due to a
- # well-time disconnect on the peer, my disconnect command might
- # be ignored and forgotten
- ever_disconnected = _IgnoreError(self._ShutdownNet, self.minor) or \
- ever_disconnected
- time.sleep(sleep_time)
- sleep_time = min(2, sleep_time * 1.5)
+ class _DisconnectStatus:
+ def __init__(self, ever_disconnected):
+ self.ever_disconnected = ever_disconnected
- if not status.is_standalone:
- if ever_disconnected:
+ dstatus = _DisconnectStatus(_IgnoreError(self._ShutdownNet, self.minor))
+
+ def _WaitForDisconnect():
+ if self.GetProcStatus().is_standalone:
+ return
+
+ # retry the disconnect, it seems possible that due to a well-time
+ # disconnect on the peer, my disconnect command might be ignored and
+ # forgotten
+ dstatus.ever_disconnected = \
+ _IgnoreError(self._ShutdownNet, self.minor) or dstatus.ever_disconnected
+
+ raise utils.RetryAgain()
+
+ # Keep start time
+ start_time = time.time()
+
+ try:
+ # Start delay at 100 milliseconds and grow up to 2 seconds
+ utils.Retry(_WaitForDisconnect, (0.1, 1.5, 2.0),
+ self._NET_RECONFIG_TIMEOUT)
+ except utils.RetryTimeout:
+ if dstatus.ever_disconnected:
msg = ("drbd%d: device did not react to the"
" 'disconnect' command in a timely manner")
else:
msg = "drbd%d: can't shutdown network, even after multiple retries"
+
_ThrowError(msg, self.minor)
- reconfig_time = time.time() - timeout_limit + self._NET_RECONFIG_TIMEOUT
- if reconfig_time > 15: # hardcoded alert limit
+ reconfig_time = time.time() - start_time
+ if reconfig_time > (self._NET_RECONFIG_TIMEOUT * 0.25):
logging.info("drbd%d: DisconnectNet: detach took %.3f seconds",
self.minor, reconfig_time)
the attach if can return success.
"""
+ # TODO: Rewrite to not use a for loop just because there is 'break'
+ # pylint: disable-msg=W0631
net_data = (self._lhost, self._lport, self._rhost, self._rport)
for minor in (self._aminor,):
info = self._GetDevInfo(self._GetShowData(minor))
if len(self._children) != 2 or None in self._children:
_ThrowError("drbd%d: cannot grow diskless device", self.minor)
self._children[0].Grow(amount)
- result = utils.RunCmd(["drbdsetup", self.dev_path, "resize"])
+ result = utils.RunCmd(["drbdsetup", self.dev_path, "resize", "-s",
+ "%dm" % (self.size + amount)])
if result.failed:
_ThrowError("drbd%d: resize failed: %s", self.minor, result.output)
if err.errno != errno.ENOENT:
_ThrowError("Can't remove file '%s': %s", self.dev_path, err)
+ def Rename(self, new_id):
+ """Renames the file.
+
+ """
+ # TODO: implement rename for file-based storage
+ _ThrowError("Rename is not supported for file-based storage")
+
+ def Grow(self, amount):
+ """Grow the file
+
+ @param amount: the amount (in mebibytes) to grow with
+
+ """
+ # TODO: implement grow for file-based storage
+ _ThrowError("Grow not supported for file-based storage")
+
def Attach(self):
"""Attach to an existing file.
self.attached = os.path.exists(self.dev_path)
return self.attached
+ def GetActualSize(self):
+ """Return the actual disk size.
+
+ @note: the device needs to be active when this is called
+
+ """
+ assert self.attached, "BlockDevice not attached in GetActualSize()"
+ try:
+ st = os.stat(self.dev_path)
+ return st.st_size
+ except OSError, err:
+ _ThrowError("Can't stat %s: %s", self.dev_path, err)
+
@classmethod
def Create(cls, unique_id, children, size):
"""Create a new file.
DEV_MAP = {
constants.LD_LV: LogicalVolume,
constants.LD_DRBD8: DRBD8,
- constants.LD_FILE: FileStorage,
}
+if constants.ENABLE_FILE_STORAGE:
+ DEV_MAP[constants.LD_FILE] = FileStorage
+
def FindDevice(dev_type, unique_id, children, size):
"""Search for an existing, assembled device.