Synnefo » snf-ganeti

#

#

# Copyright (C) 2006, 2007, 2010, 2011, 2012, 2013 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

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful, but

# WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU

# General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA

# 02110-1301, USA.

"""Block device abstraction"""

import re

import errno

import stat

import os

import logging

import math

from ganeti import utils

from ganeti import errors

from ganeti import constants

from ganeti import objects

from ganeti import compat

from ganeti import pathutils

from ganeti import serializer

from ganeti.storage import base

from ganeti.storage import drbd

from ganeti.storage import filestorage

class RbdShowmappedJsonError(Exception):

  """`rbd showmmapped' JSON formatting error Exception class.

  """

  pass

def _CheckResult(result):

  """Throws an error if the given result is a failed one.

  @param result: result from RunCmd

  """

  if result.failed:

    base.ThrowError("Command: %s error: %s - %s",

                    result.cmd, result.fail_reason, result.output)

class LogicalVolume(base.BlockDev):

  """Logical Volume block device.

  """

  _VALID_NAME_RE = re.compile("^[a-zA-Z0-9+_.-]*$")

  _PARSE_PV_DEV_RE = re.compile(r"^([^ ()]+)\([0-9]+\)$")

  _INVALID_NAMES = compat.UniqueFrozenset([".", "..", "snapshot", "pvmove"])

  _INVALID_SUBSTRINGS = compat.UniqueFrozenset(["_mlog", "_mimage"])

  def __init__(self, unique_id, children, size, params, dyn_params, *args):

    """Attaches to a LV device.

    The unique_id is a tuple (vg_name, lv_name)

    """

    super(LogicalVolume, self).__init__(unique_id, children, size, params,

                                        dyn_params, *args)

    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._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 = self.pe_size = self.stripe_count = None

    self.pv_names = None

    self.Attach()

  @staticmethod

  def _GetStdPvSize(pvs_info):

    """Return the the standard PV size (used with exclusive storage).

    @param pvs_info: list of objects.LvmPvInfo, cannot be empty

    @rtype: float

    @return: size in MiB

    """

    assert len(pvs_info) > 0

    smallest = min([pv.size for pv in pvs_info])

    return smallest / (1 + constants.PART_MARGIN + constants.PART_RESERVED)

  @staticmethod

  def _ComputeNumPvs(size, pvs_info):

    """Compute the number of PVs needed for an LV (with exclusive storage).

    @type size: float

    @param size: LV size in MiB

    @param pvs_info: list of objects.LvmPvInfo, cannot be empty

    @rtype: integer

    @return: number of PVs needed

    """

    assert len(pvs_info) > 0

    pv_size = float(LogicalVolume._GetStdPvSize(pvs_info))

    return int(math.ceil(float(size) / pv_size))

  @staticmethod

  def _GetEmptyPvNames(pvs_info, max_pvs=None):

    """Return a list of empty PVs, by name.

    """

    empty_pvs = filter(objects.LvmPvInfo.IsEmpty, pvs_info)

    if max_pvs is not None:

      empty_pvs = empty_pvs[:max_pvs]

    return map((lambda pv: pv.name), empty_pvs)

  @classmethod

  def Create(cls, unique_id, children, size, spindles, params, excl_stor,

             dyn_params):

    """Create a new logical volume.

    """

    if not isinstance(unique_id, (tuple, list)) or len(unique_id) != 2:

      raise errors.ProgrammerError("Invalid configuration data %s" %

                                   str(unique_id))

    vg_name, lv_name = unique_id

    cls._ValidateName(vg_name)

    cls._ValidateName(lv_name)

    pvs_info = cls.GetPVInfo([vg_name])

    if not pvs_info:

      if excl_stor:

        msg = "No (empty) PVs found"

      else:

        msg = "Can't compute PV info for vg %s" % vg_name

      base.ThrowError(msg)

    pvs_info.sort(key=(lambda pv: pv.free), reverse=True)

    pvlist = [pv.name for pv in pvs_info]

    if compat.any(":" in v for v in pvlist):

      base.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'")

    current_pvs = len(pvlist)

    desired_stripes = params[constants.LDP_STRIPES]

    stripes = min(current_pvs, desired_stripes)

    if excl_stor:

      if spindles is None:

        base.ThrowError("Unspecified number of spindles: this is required"

                        "when exclusive storage is enabled, try running"

                        " gnt-cluster repair-disk-sizes")

      (err_msgs, _) = utils.LvmExclusiveCheckNodePvs(pvs_info)

      if err_msgs:

        for m in err_msgs:

          logging.warning(m)

      req_pvs = cls._ComputeNumPvs(size, pvs_info)

      if spindles < req_pvs:

        base.ThrowError("Requested number of spindles (%s) is not enough for"

                        " a disk of %d MB (at least %d spindles needed)",

                        spindles, size, req_pvs)

      else:

        req_pvs = spindles

      pvlist = cls._GetEmptyPvNames(pvs_info, req_pvs)

      current_pvs = len(pvlist)

      if current_pvs < req_pvs:

        base.ThrowError("Not enough empty PVs (spindles) to create a disk of %d"

                        " MB: %d available, %d needed",

                        size, current_pvs, req_pvs)

      assert current_pvs == len(pvlist)

      # We must update stripes to be sure to use all the desired spindles

      stripes = current_pvs

      if stripes > desired_stripes:

        # Don't warn when lowering stripes, as it's no surprise

        logging.warning("Using %s stripes instead of %s, to be able to use"

                        " %s spindles", stripes, desired_stripes, current_pvs)

    else:

      if stripes < desired_stripes:

        logging.warning("Could not use %d stripes for VG %s, as only %d PVs are"

                        " available.", desired_stripes, vg_name, current_pvs)

      free_size = sum([pv.free for pv in pvs_info])

      # The size constraint should have been checked from the master before

      # calling the create function.

      if free_size < size:

        base.ThrowError("Not enough free space: required %s,"

                        " available %s", size, free_size)

    # If the free space is not well distributed, we won't be able to

    # create an optimally-striped volume; in that case, we want to try

    # with N, N-1, ..., 2, and finally 1 (non-stripped) number of

    # stripes

    cmd = ["lvcreate", "-L%dm" % size, "-n%s" % lv_name]

    for stripes_arg in range(stripes, 0, -1):

      result = utils.RunCmd(cmd + ["-i%d" % stripes_arg] + [vg_name] + pvlist)

      if not result.failed:

        break

    if result.failed:

      base.ThrowError("LV create failed (%s): %s",

                      result.fail_reason, result.output)

    return LogicalVolume(unique_id, children, size, params, dyn_params)

  @staticmethod

  def _GetVolumeInfo(lvm_cmd, fields):

    """Returns LVM Volume infos using lvm_cmd

    @param lvm_cmd: Should be one of "pvs", "vgs" or "lvs"

    @param fields: Fields to return

    @return: A list of dicts each with the parsed fields

    """

    if not fields:

      raise errors.ProgrammerError("No fields specified")

    sep = "|"

    cmd = [lvm_cmd, "--noheadings", "--nosuffix", "--units=m", "--unbuffered",

           "--separator=%s" % sep, "-o%s" % ",".join(fields)]

    result = utils.RunCmd(cmd)

    if result.failed:

      raise errors.CommandError("Can't get the volume information: %s - %s" %

                                (result.fail_reason, result.output))

    data = []

    for line in result.stdout.splitlines():

      splitted_fields = line.strip().split(sep)

      if len(fields) != len(splitted_fields):

        raise errors.CommandError("Can't parse %s output: line '%s'" %

                                  (lvm_cmd, line))

      data.append(splitted_fields)

    return data

  @classmethod

  def GetPVInfo(cls, vg_names, filter_allocatable=True, include_lvs=False):

    """Get the free space info for PVs in a volume group.

    @param vg_names: list of volume group names, if empty all will be returned

    @param filter_allocatable: whether to skip over unallocatable PVs

    @param include_lvs: whether to include a list of LVs hosted on each PV

    @rtype: list

    @return: list of objects.LvmPvInfo objects

    """

    # We request "lv_name" field only if we care about LVs, so we don't get

    # a long list of entries with many duplicates unless we really have to.

    # The duplicate "pv_name" field will be ignored.

    if include_lvs:

      lvfield = "lv_name"

    else:

      lvfield = "pv_name"

    try:

      info = cls._GetVolumeInfo("pvs", ["pv_name", "vg_name", "pv_free",

                                        "pv_attr", "pv_size", lvfield])

    except errors.GenericError, err:

      logging.error("Can't get PV information: %s", err)

      return None

    # When asked for LVs, "pvs" may return multiple entries for the same PV-LV

    # pair. We sort entries by PV name and then LV name, so it's easy to weed

    # out duplicates.

    if include_lvs:

      info.sort(key=(lambda i: (i[0], i[5])))

    data = []

    lastpvi = None

    for (pv_name, vg_name, pv_free, pv_attr, pv_size, lv_name) in info:

      # (possibly) skip over pvs which are not allocatable

      if filter_allocatable and pv_attr[0] != "a":

        continue

      # (possibly) skip over pvs which are not in the right volume group(s)

      if vg_names and vg_name not in vg_names:

        continue

      # Beware of duplicates (check before inserting)

      if lastpvi and lastpvi.name == pv_name:

        if include_lvs and lv_name:

          if not lastpvi.lv_list or lastpvi.lv_list[-1] != lv_name:

            lastpvi.lv_list.append(lv_name)

      else:

        if include_lvs and lv_name:

          lvl = [lv_name]

        else:

          lvl = []

        lastpvi = objects.LvmPvInfo(name=pv_name, vg_name=vg_name,

                                    size=float(pv_size), free=float(pv_free),

                                    attributes=pv_attr, lv_list=lvl)

        data.append(lastpvi)

    return data

  @classmethod

  def _GetRawFreePvInfo(cls, vg_name):

    """Return info (size/free) about PVs.

    @type vg_name: string

    @param vg_name: VG name

    @rtype: tuple

    @return: (standard_pv_size_in_MiB, number_of_free_pvs, total_number_of_pvs)

    """

    pvs_info = cls.GetPVInfo([vg_name])

    if not pvs_info:

      pv_size = 0.0

      free_pvs = 0

      num_pvs = 0

    else:

      pv_size = cls._GetStdPvSize(pvs_info)

      free_pvs = len(cls._GetEmptyPvNames(pvs_info))

      num_pvs = len(pvs_info)

    return (pv_size, free_pvs, num_pvs)

  @classmethod

  def _GetExclusiveStorageVgFree(cls, vg_name):

    """Return the free disk space in the given VG, in exclusive storage mode.

    @type vg_name: string

    @param vg_name: VG name

    @rtype: float

    @return: free space in MiB

    """

    (pv_size, free_pvs, _) = cls._GetRawFreePvInfo(vg_name)

    return pv_size * free_pvs

  @classmethod

  def GetVgSpindlesInfo(cls, vg_name):

    """Get the free space info for specific VGs.

    @param vg_name: volume group name

    @rtype: tuple

    @return: (free_spindles, total_spindles)

    """

    (_, free_pvs, num_pvs) = cls._GetRawFreePvInfo(vg_name)

    return (free_pvs, num_pvs)

  @classmethod

  def GetVGInfo(cls, vg_names, excl_stor, filter_readonly=True):

    """Get the free space info for specific VGs.

    @param vg_names: list of volume group names, if empty all will be returned

    @param excl_stor: whether exclusive_storage is enabled

    @param filter_readonly: whether to skip over readonly VGs

    @rtype: list

    @return: list of tuples (free_space, total_size, name) with free_space in

             MiB

    """

    try:

      info = cls._GetVolumeInfo("vgs", ["vg_name", "vg_free", "vg_attr",

                                        "vg_size"])

    except errors.GenericError, err:

      logging.error("Can't get VG information: %s", err)

      return None

    data = []

    for vg_name, vg_free, vg_attr, vg_size in info:

      # (possibly) skip over vgs which are not writable

      if filter_readonly and vg_attr[0] == "r":

        continue

      # (possibly) skip over vgs which are not in the right volume group(s)

      if vg_names and vg_name not in vg_names:

        continue

      # Exclusive storage needs a different concept of free space

      if excl_stor:

        es_free = cls._GetExclusiveStorageVgFree(vg_name)

        assert es_free <= vg_free

        vg_free = es_free

      data.append((float(vg_free), float(vg_size), vg_name))

    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

        compat.any(substring in name for substring in cls._INVALID_SUBSTRINGS)):

      base.ThrowError("Invalid LVM name '%s'", name)

  def Remove(self):

    """Remove this logical volume.

    """

    if not self.minor and not self.Attach():

      # the LV does not exist

      return

    result = utils.RunCmd(["lvremove", "-f", "%s/%s" %

                           (self._vg_name, self._lv_name)])

    if result.failed:

      base.ThrowError("Can't lvremove: %s - %s",

                      result.fail_reason, result.output)

  def Rename(self, new_id):

    """Rename this logical volume.

    """

    if not isinstance(new_id, (tuple, list)) or len(new_id) != 2:

      raise errors.ProgrammerError("Invalid new logical id '%s'" % new_id)

    new_vg, new_name = new_id

    if new_vg != self._vg_name:

      raise errors.ProgrammerError("Can't move a logical volume across"

                                   " volume groups (from %s to to %s)" %

                                   (self._vg_name, new_vg))

    result = utils.RunCmd(["lvrename", new_vg, self._lv_name, new_name])

    if result.failed:

      base.ThrowError("Failed to rename the logical volume: %s", result.output)

    self._lv_name = new_name

    self.dev_path = utils.PathJoin("/dev", self._vg_name, self._lv_name)

  @classmethod

  def _ParseLvInfoLine(cls, line, sep):

    """Parse one line of the lvs output used in L{_GetLvInfo}.

    """

    elems = line.strip().rstrip(sep).split(sep)

    if len(elems) != 6:

      base.ThrowError("Can't parse LVS output, len(%s) != 6", str(elems))

    (status, major, minor, pe_size, stripes, pvs) = elems

    if len(status) < 6:

      base.ThrowError("lvs lv_attr is not at least 6 characters (%s)", status)

    try:

      major = int(major)

      minor = int(minor)

    except (TypeError, ValueError), err:

      base.ThrowError("lvs major/minor cannot be parsed: %s", str(err))

    try:

      pe_size = int(float(pe_size))

    except (TypeError, ValueError), err:

      base.ThrowError("Can't parse vg extent size: %s", err)

    try:

      stripes = int(stripes)

    except (TypeError, ValueError), err:

      base.ThrowError("Can't parse the number of stripes: %s", err)

    pv_names = []

    for pv in pvs.split(","):

      m = re.match(cls._PARSE_PV_DEV_RE, pv)

      if not m:

        base.ThrowError("Can't parse this device list: %s", pvs)

      pv_names.append(m.group(1))

    assert len(pv_names) > 0

    return (status, major, minor, pe_size, stripes, pv_names)

  @classmethod

  def _GetLvInfo(cls, dev_path, _run_cmd=utils.RunCmd):

    """Get info about the given existing LV to be used.

    """

    sep = "|"

    result = _run_cmd(["lvs", "--noheadings", "--separator=%s" % sep,

                       "--units=k", "--nosuffix",

                       "-olv_attr,lv_kernel_major,lv_kernel_minor,"

                       "vg_extent_size,stripes,devices", dev_path])

    if result.failed:

      base.ThrowError("Can't find LV %s: %s, %s",

                      dev_path, result.fail_reason, result.output)

    # 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

      base.ThrowError("Can't parse LVS output, no lines? Got '%s'", str(out))

    pv_names = set()

    for line in out:

      (status, major, minor, pe_size, stripes, more_pvs) = \

        cls._ParseLvInfoLine(line, sep)

      pv_names.update(more_pvs)

    return (status, major, minor, pe_size, stripes, pv_names)

  def Attach(self):

    """Attach to an existing LV.

    This method will try to see if an existing and active LV exists

    which matches our name. If so, its major/minor will be

    recorded.

    """

    self.attached = False

    try:

      (status, major, minor, pe_size, stripes, pv_names) = \

        self._GetLvInfo(self.dev_path)

    except errors.BlockDeviceError:

      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.pv_names = pv_names

    self.attached = True

    return True

  def Assemble(self):

    """Assemble the device.

    We always run `lvchange -ay` on the LV to ensure it's active before

    use, as there were cases when xenvg was not active after boot

    (also possibly after disk issues).

    """

    result = utils.RunCmd(["lvchange", "-ay", self.dev_path])

    if result.failed:

      base.ThrowError("Can't activate lv %s: %s", self.dev_path, result.output)

  def Shutdown(self):

    """Shutdown the device.

    This is a no-op for the LV device type, as we don't deactivate the

    volumes on shutdown.

    """

    pass

  def GetSyncStatus(self):

    """Returns the sync status of the device.

    If this device is a mirroring device, this function returns the

    status of the mirror.

    For logical volumes, sync_percent and estimated_time are always

    None (no recovery in progress, as we don't handle the mirrored LV

    case). The is_degraded parameter is the inverse of the ldisk

    parameter.

    For the ldisk parameter, we check if the logical volume has the

    'virtual' type, which means it's not backed by existing storage

    anymore (read from it return I/O error). This happens after a

    physical disk failure and subsequent 'vgreduce --removemissing' on

    the volume group.

    The status was already read in Attach, so we just return it.

    @rtype: objects.BlockDevStatus

    """

    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.

    This is a no-op for the LV device type.

    """

    pass

  def Close(self):

    """Notifies that the device will no longer be used for I/O.

    This is a no-op for the LV device type.

    """

    pass

  def Snapshot(self, size):

    """Create a snapshot copy of an lvm block device.

    @returns: tuple (vg, lv)

    """

    snap_name = self._lv_name + ".snap"

    # remove existing snapshot if found

    snap = LogicalVolume((self._vg_name, snap_name), None, size, self.params,

                         self.dyn_params)

    base.IgnoreError(snap.Remove)

    vg_info = self.GetVGInfo([self._vg_name], False)

    if not vg_info:

      base.ThrowError("Can't compute VG info for vg %s", self._vg_name)

    free_size, _, _ = vg_info[0]

    if free_size < size:

      base.ThrowError("Not enough free space: required %s,"

                      " available %s", size, free_size)

    _CheckResult(utils.RunCmd(["lvcreate", "-L%dm" % size, "-s",

                               "-n%s" % snap_name, self.dev_path]))

    return (self._vg_name, snap_name)

  def _RemoveOldInfo(self):

    """Try to remove old tags from the lv.

    """

    result = utils.RunCmd(["lvs", "-o", "tags", "--noheadings", "--nosuffix",

                           self.dev_path])

    _CheckResult(result)

    raw_tags = result.stdout.strip()

    if raw_tags:

      for tag in raw_tags.split(","):

        _CheckResult(utils.RunCmd(["lvchange", "--deltag",

                                   tag.strip(), self.dev_path]))

  def SetInfo(self, text):

    """Update metadata with info text.

    """

    base.BlockDev.SetInfo(self, text)

    self._RemoveOldInfo()

    # Replace invalid characters

    text = re.sub("^[^A-Za-z0-9_+.]", "_", text)

    text = re.sub("[^-A-Za-z0-9_+.]", "_", text)

    # Only up to 128 characters are allowed

    text = text[:128]

    _CheckResult(utils.RunCmd(["lvchange", "--addtag", text, self.dev_path]))

  def _GetGrowthAvaliabilityExclStor(self):

    """Return how much the disk can grow with exclusive storage.

    @rtype: float

    @return: available space in Mib

    """

    pvs_info = self.GetPVInfo([self._vg_name])

    if not pvs_info:

      base.ThrowError("Cannot get information about PVs for %s", self.dev_path)

    std_pv_size = self._GetStdPvSize(pvs_info)

    free_space = sum(pvi.free - (pvi.size - std_pv_size)

                        for pvi in pvs_info

                        if pvi.name in self.pv_names)

    return free_space

  def Grow(self, amount, dryrun, backingstore, excl_stor):

    """Grow the logical volume.

    """

    if not backingstore:

      return

    if self.pe_size is None or self.stripe_count is None:

      if not self.Attach():

        base.ThrowError("Can't attach to LV during Grow()")

    full_stripe_size = self.pe_size * self.stripe_count

    # pe_size is in KB

    amount *= 1024

    rest = amount % full_stripe_size

    if rest != 0:

      amount += full_stripe_size - rest

    cmd = ["lvextend", "-L", "+%dk" % amount]

    if dryrun:

      cmd.append("--test")

    if excl_stor:

      free_space = self._GetGrowthAvaliabilityExclStor()

      # amount is in KiB, free_space in MiB

      if amount > free_space * 1024:

        base.ThrowError("Not enough free space to grow %s: %d MiB required,"

                        " %d available", self.dev_path, amount / 1024,

                        free_space)

      # Disk growth doesn't grow the number of spindles, so we must stay within

      # our assigned volumes

      pvlist = list(self.pv_names)

    else:

      pvlist = []

    # 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

    # supports 'cling'

    for alloc_policy in "contiguous", "cling", "normal":

      result = utils.RunCmd(cmd + ["--alloc", alloc_policy, self.dev_path] +

                            pvlist)

      if not result.failed:

        return

    base.ThrowError("Can't grow LV %s: %s", self.dev_path, result.output)

  def GetActualSpindles(self):

    """Return the number of spindles used.

    """

    assert self.attached, "BlockDevice not attached in GetActualSpindles()"

    return len(self.pv_names)

class FileStorage(base.BlockDev):

  """File device.

  This class represents a file storage backend device.

  The unique_id for the file device is a (file_driver, file_path) tuple.

  """

  def __init__(self, unique_id, children, size, params, dyn_params, *args):

    """Initalizes a file device backend.

    """

    if children:

      raise errors.BlockDeviceError("Invalid setup for file device")

    super(FileStorage, self).__init__(unique_id, children, size, params,

                                      dyn_params)

    if not isinstance(unique_id, (tuple, list)) or len(unique_id) != 2:

      raise ValueError("Invalid configuration data %s" % str(unique_id))