Remove deprecated “QueryLocks” LUXI request
[ganeti-local] / lib / bdev.py
index bfb37c3..9488826 100644 (file)
@@ -1,7 +1,7 @@
 #
 #
 
 #
 #
 
-# Copyright (C) 2006, 2007 Google Inc.
+# Copyright (C) 2006, 2007, 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
 #
 # 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
@@ -24,6 +24,8 @@
 import re
 import time
 import errno
 import re
 import time
 import errno
+import shlex
+import stat
 import pyparsing as pyp
 import os
 import logging
 import pyparsing as pyp
 import os
 import logging
@@ -32,6 +34,12 @@ from ganeti import utils
 from ganeti import errors
 from ganeti import constants
 from ganeti import objects
 from ganeti import errors
 from ganeti import constants
 from ganeti import objects
+from ganeti import compat
+from ganeti import netutils
+
+
+# Size of reads in _CanReadDevice
+_DEVICE_READ_SIZE = 128 * 1024
 
 
 def _IgnoreError(fn, *args, **kwargs):
 
 
 def _IgnoreError(fn, *args, **kwargs):
@@ -48,7 +56,7 @@ def _IgnoreError(fn, *args, **kwargs):
     fn(*args, **kwargs)
     return True
   except errors.BlockDeviceError, err:
     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
 
 
     return False
 
 
@@ -66,6 +74,20 @@ def _ThrowError(msg, *args):
   raise errors.BlockDeviceError(msg)
 
 
   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.
 
 class BlockDev(object):
   """Block device abstract class.
 
@@ -109,7 +131,7 @@ class BlockDev(object):
   after assembly we'll have our correct major/minor.
 
   """
   after assembly we'll have our correct major/minor.
 
   """
-  def __init__(self, unique_id, children, size):
+  def __init__(self, unique_id, children, size, params):
     self._children = children
     self.dev_path = None
     self.unique_id = unique_id
     self._children = children
     self.dev_path = None
     self.unique_id = unique_id
@@ -117,6 +139,7 @@ class BlockDev(object):
     self.minor = None
     self.attached = False
     self.size = size
     self.minor = None
     self.attached = False
     self.size = size
+    self.params = params
 
   def Assemble(self):
     """Assemble the device from its components.
 
   def Assemble(self):
     """Assemble the device from its components.
@@ -145,7 +168,7 @@ class BlockDev(object):
     raise NotImplementedError
 
   @classmethod
     raise NotImplementedError
 
   @classmethod
-  def Create(cls, unique_id, children, size):
+  def Create(cls, unique_id, children, size, params):
     """Create the device.
 
     If the device cannot be created, it will return None
     """Create the device.
 
     If the device cannot be created, it will return None
@@ -198,16 +221,36 @@ class BlockDev(object):
     """
     raise NotImplementedError
 
     """
     raise NotImplementedError
 
-  def SetSyncSpeed(self, speed):
-    """Adjust the sync speed of the mirror.
+  def SetSyncParams(self, params):
+    """Adjust the synchronization parameters of the mirror.
 
     In case this is not a mirroring device, this is no-op.
 
 
     In case this is not a mirroring device, this is no-op.
 
+    @param params: dictionary of LD level disk parameters related to the
+    synchronization.
+    @rtype: list
+    @return: a list of error messages, emitted both by the current node and by
+    children. An empty list means no errors.
+
+    """
+    result = []
+    if self._children:
+      for child in self._children:
+        result.extend(child.SetSyncParams(params))
+    return result
+
+  def PauseResumeSync(self, pause):
+    """Pause/Resume the sync of the mirror.
+
+    In case this is not a mirroring device, this is no-op.
+
+    @param pause: Whether to pause or resume
+
     """
     result = True
     if self._children:
       for child in self._children:
     """
     result = True
     if self._children:
       for child in self._children:
-        result = result and child.SetSyncSpeed(speed)
+        result = result and child.PauseResumeSync(pause)
     return result
 
   def GetSyncStatus(self):
     return result
 
   def GetSyncStatus(self):
@@ -238,7 +281,7 @@ class BlockDev(object):
                                   sync_percent=None,
                                   estimated_time=None,
                                   is_degraded=False,
                                   sync_percent=None,
                                   estimated_time=None,
                                   is_degraded=False,
-                                  ldisk_degraded=False)
+                                  ldisk_status=constants.LDS_OKAY)
 
   def CombinedSyncStatus(self):
     """Calculate the mirror status recursively for our children.
 
   def CombinedSyncStatus(self):
     """Calculate the mirror status recursively for our children.
@@ -255,7 +298,7 @@ class BlockDev(object):
     min_percent = status.sync_percent
     max_time = status.estimated_time
     is_degraded = status.is_degraded
     min_percent = status.sync_percent
     max_time = status.estimated_time
     is_degraded = status.is_degraded
-    ldisk_degraded = status.ldisk_degraded
+    ldisk_status = status.ldisk_status
 
     if self._children:
       for child in self._children:
 
     if self._children:
       for child in self._children:
@@ -272,7 +315,11 @@ class BlockDev(object):
           max_time = max(max_time, child_status.estimated_time)
 
         is_degraded = is_degraded or child_status.is_degraded
           max_time = max(max_time, child_status.estimated_time)
 
         is_degraded = is_degraded or child_status.is_degraded
-        ldisk_degraded = ldisk_degraded or child_status.ldisk_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,
 
     return objects.BlockDevStatus(dev_path=self.dev_path,
                                   major=self.major,
@@ -280,8 +327,7 @@ class BlockDev(object):
                                   sync_percent=min_percent,
                                   estimated_time=max_time,
                                   is_degraded=is_degraded,
                                   sync_percent=min_percent,
                                   estimated_time=max_time,
                                   is_degraded=is_degraded,
-                                  ldisk_degraded=ldisk_degraded)
-
+                                  ldisk_status=ldisk_status)
 
   def SetInfo(self, text):
     """Update metadata with info text.
 
   def SetInfo(self, text):
     """Update metadata with info text.
@@ -292,14 +338,35 @@ class BlockDev(object):
     for child in self._children:
       child.SetInfo(text)
 
     for child in self._children:
       child.SetInfo(text)
 
-  def Grow(self, amount):
+  def Grow(self, amount, dryrun):
     """Grow the block device.
 
     """Grow the block device.
 
+    @type amount: integer
     @param amount: the amount (in mebibytes) to grow with
     @param amount: the amount (in mebibytes) to grow with
+    @type dryrun: boolean
+    @param dryrun: whether to execute the operation in simulation mode
+        only, without actually increasing the size
 
     """
     raise NotImplementedError
 
 
     """
     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,
   def __repr__(self):
     return ("<%s: unique_id: %s, children: %s, %s:%s, %s>" %
             (self.__class__, self.unique_id, self._children,
@@ -310,23 +377,29 @@ class LogicalVolume(BlockDev):
   """Logical Volume block device.
 
   """
   """Logical Volume block device.
 
   """
-  def __init__(self, unique_id, children, size):
+  _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, params):
     """Attaches to a LV device.
 
     The unique_id is a tuple (vg_name, lv_name)
 
     """
     """Attaches to a LV device.
 
     The unique_id is a tuple (vg_name, lv_name)
 
     """
-    super(LogicalVolume, self).__init__(unique_id, children, size)
+    super(LogicalVolume, self).__init__(unique_id, children, size, params)
     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
     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 = self.pe_size = self.stripe_count = None
     self.Attach()
 
   @classmethod
     self._degraded = True
     self.major = self.minor = self.pe_size = self.stripe_count = None
     self.Attach()
 
   @classmethod
-  def Create(cls, unique_id, children, size):
+  def Create(cls, unique_id, children, size, params):
     """Create a new logical volume.
 
     """
     """Create a new logical volume.
 
     """
@@ -334,16 +407,26 @@ class LogicalVolume(BlockDev):
       raise errors.ProgrammerError("Invalid configuration data %s" %
                                    str(unique_id))
     vg_name, lv_name = unique_id
       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()
 
     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 ]
-    free_size = sum([ pv[0] for pv in pvs_info ])
+    pvlist = [pv[1] for pv in pvs_info]
+    if compat.any(":" in v for v in pvlist):
+      _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)
     current_pvs = len(pvlist)
-    stripes = min(current_pvs, constants.LVM_STRIPECOUNT)
+    desired_stripes = params[constants.LDP_STRIPES]
+    stripes = min(current_pvs, desired_stripes)
+    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)
 
     # The size constraint should have been checked from the master before
     # calling the create function.
 
     # The size constraint should have been checked from the master before
     # calling the create function.
@@ -362,39 +445,116 @@ class LogicalVolume(BlockDev):
     if result.failed:
       _ThrowError("LV create failed (%s): %s",
                   result.fail_reason, result.output)
     if result.failed:
       _ThrowError("LV create failed (%s): %s",
                   result.fail_reason, result.output)
-    return LogicalVolume(unique_id, children, size)
+    return LogicalVolume(unique_id, children, size, params)
 
   @staticmethod
 
   @staticmethod
-  def GetPVInfo(vg_name):
+  def _GetVolumeInfo(lvm_cmd, fields):
+    """Returns LVM Volumen 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):
     """Get the free space info for PVs in a volume group.
 
     """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
 
     """
 
     @rtype: list
     @return: list of tuples (free_space, name) with free_space in mebibytes
 
     """
-    command = ["pvs", "--noheadings", "--nosuffix", "--units=m",
-               "-opv_name,vg_name,pv_free,pv_attr", "--unbuffered",
-               "--separator=:"]
-    result = utils.RunCmd(command)
-    if result.failed:
-      logging.error("Can't get the PV information: %s - %s",
-                    result.fail_reason, result.output)
+    try:
+      info = cls._GetVolumeInfo("pvs", ["pv_name", "vg_name", "pv_free",
+                                        "pv_attr"])
+    except errors.GenericError, err:
+      logging.error("Can't get PV information: %s", err)
       return None
       return None
+
     data = []
     data = []
-    for line in result.stdout.splitlines():
-      fields = line.strip().split(':')
-      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':
+    for pv_name, vg_name, pv_free, pv_attr 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
         continue
-      data.append((float(fields[2]), fields[0]))
+      data.append((float(pv_free), pv_name, vg_name))
 
     return data
 
 
     return data
 
+  @classmethod
+  def GetVGInfo(cls, vg_names, 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 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
+      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)):
+      _ThrowError("Invalid LVM name '%s'", name)
+
   def Remove(self):
     """Remove this logical volume.
 
   def Remove(self):
     """Remove this logical volume.
 
@@ -422,7 +582,7 @@ class LogicalVolume(BlockDev):
     if result.failed:
       _ThrowError("Failed to rename the logical volume: %s", result.output)
     self._lv_name = new_name
     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.
 
   def Attach(self):
     """Attach to an existing LV.
@@ -451,7 +611,7 @@ class LogicalVolume(BlockDev):
                 # one line for any non-empty string
       logging.error("Can't parse LVS output, no lines? Got '%s'", str(out))
       return False
                 # 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[-1].strip().rstrip(",")
     out = out.split(",")
     if len(out) != 5:
       logging.error("Can't parse LVS output, len(%s) != 5", str(out))
     out = out.split(",")
     if len(out) != 5:
       logging.error("Can't parse LVS output, len(%s) != 5", str(out))
@@ -465,7 +625,7 @@ class LogicalVolume(BlockDev):
     try:
       major = int(major)
       minor = int(minor)
     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:
       logging.error("lvs major/minor cannot be parsed: %s", str(err))
 
     try:
@@ -484,7 +644,7 @@ class LogicalVolume(BlockDev):
     self.minor = minor
     self.pe_size = pe_size
     self.stripe_count = stripes
     self.minor = minor
     self.pe_size = pe_size
     self.stripe_count = stripes
-    self._degraded = status[0] == 'v' # virtual volume, i.e. doesn't backing
+    self._degraded = status[0] == "v" # virtual volume, i.e. doesn't backing
                                       # storage
     self.attached = True
     return True
                                       # storage
     self.attached = True
     return True
@@ -532,13 +692,18 @@ class LogicalVolume(BlockDev):
     @rtype: objects.BlockDevStatus
 
     """
     @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,
     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_degraded=self._degraded)
+                                  ldisk_status=ldisk_status)
 
   def Open(self, force=False):
     """Make the device ready for I/O.
 
   def Open(self, force=False):
     """Make the device ready for I/O.
@@ -559,19 +724,19 @@ class LogicalVolume(BlockDev):
   def Snapshot(self, size):
     """Create a snapshot copy of an lvm block device.
 
   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_name = self._lv_name + ".snap"
 
     # remove existing snapshot if found
-    snap = LogicalVolume((self._vg_name, snap_name), None, size)
+    snap = LogicalVolume((self._vg_name, snap_name), None, size, self.params)
     _IgnoreError(snap.Remove)
 
     _IgnoreError(snap.Remove)
 
-    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]
+    vg_info = self.GetVGInfo([self._vg_name])
+    if not vg_info:
+      _ThrowError("Can't compute VG info for vg %s", self._vg_name)
+    free_size, _, _ = vg_info[0]
     if free_size < size:
       _ThrowError("Not enough free space: required %s,"
                   " available %s", size, free_size)
     if free_size < size:
       _ThrowError("Not enough free space: required %s,"
                   " available %s", size, free_size)
@@ -582,7 +747,7 @@ class LogicalVolume(BlockDev):
       _ThrowError("command: %s error: %s - %s",
                   result.cmd, result.fail_reason, result.output)
 
       _ThrowError("command: %s error: %s - %s",
                   result.cmd, result.fail_reason, result.output)
 
-    return snap_name
+    return (self._vg_name, snap_name)
 
   def SetInfo(self, text):
     """Update metadata with info text.
 
   def SetInfo(self, text):
     """Update metadata with info text.
@@ -591,8 +756,8 @@ class LogicalVolume(BlockDev):
     BlockDev.SetInfo(self, text)
 
     # Replace invalid characters
     BlockDev.SetInfo(self, text)
 
     # Replace invalid characters
-    text = re.sub('^[^A-Za-z0-9_+.]', '_', text)
-    text = re.sub('[^-A-Za-z0-9_+.]', '_', text)
+    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]
 
     # Only up to 128 characters are allowed
     text = text[:128]
@@ -603,7 +768,7 @@ class LogicalVolume(BlockDev):
       _ThrowError("Command: %s error: %s - %s", result.cmd, result.fail_reason,
                   result.output)
 
       _ThrowError("Command: %s error: %s - %s", result.cmd, result.fail_reason,
                   result.output)
 
-  def Grow(self, amount):
+  def Grow(self, amount, dryrun):
     """Grow the logical volume.
 
     """
     """Grow the logical volume.
 
     """
@@ -614,13 +779,15 @@ class LogicalVolume(BlockDev):
     rest = amount % full_stripe_size
     if rest != 0:
       amount += full_stripe_size - rest
     rest = amount % full_stripe_size
     if rest != 0:
       amount += full_stripe_size - rest
+    cmd = ["lvextend", "-L", "+%dm" % amount]
+    if dryrun:
+      cmd.append("--test")
     # 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":
     # 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(["lvextend", "--alloc", alloc_policy,
-                             "-L", "+%dm" % amount, self.dev_path])
+      result = utils.RunCmd(cmd + ["--alloc", alloc_policy, self.dev_path])
       if not result.failed:
         return
     _ThrowError("Can't grow LV %s: %s", self.dev_path, result.output)
       if not result.failed:
         return
     _ThrowError("Can't grow LV %s: %s", self.dev_path, result.output)
@@ -636,7 +803,10 @@ class DRBD8Status(object):
   LINE_RE = re.compile(r"\s*[0-9]+:\s*cs:(\S+)\s+(?:st|ro):([^/]+)/(\S+)"
                        "\s+ds:([^/]+)/(\S+)\s+.*$")
   SYNC_RE = re.compile(r"^.*\ssync'ed:\s*([0-9.]+)%.*"
   LINE_RE = re.compile(r"\s*[0-9]+:\s*cs:(\S+)\s+(?:st|ro):([^/]+)/(\S+)"
                        "\s+ds:([^/]+)/(\S+)\s+.*$")
   SYNC_RE = re.compile(r"^.*\ssync'ed:\s*([0-9.]+)%.*"
-                       "\sfinish: ([0-9]+):([0-9]+):([0-9]+)\s.*$")
+                       # Due to a bug in drbd in the kernel, introduced in
+                       # commit 4b0715f096 (still unfixed as of 2011-08-22)
+                       "(?:\s|M)"
+                       "finish: ([0-9]+):([0-9]+):([0-9]+)\s.*$")
 
   CS_UNCONFIGURED = "Unconfigured"
   CS_STANDALONE = "StandAlone"
 
   CS_UNCONFIGURED = "Unconfigured"
   CS_STANDALONE = "StandAlone"
@@ -731,15 +901,17 @@ class DRBD8Status(object):
       self.est_time = None
 
 
       self.est_time = None
 
 
-class BaseDRBD(BlockDev):
+class BaseDRBD(BlockDev): # pylint: disable=W0223
   """Base DRBD class.
 
   This class contains a few bits of common functionality between the
   0.7 and 8.x versions of DRBD.
 
   """
   """Base DRBD class.
 
   This class contains a few bits of common functionality between the
   0.7 and 8.x versions of DRBD.
 
   """
-  _VERSION_RE = re.compile(r"^version: (\d+)\.(\d+)\.(\d+)"
+  _VERSION_RE = re.compile(r"^version: (\d+)\.(\d+)\.(\d+)(?:\.\d+)?"
                            r" \(api:(\d+)/proto:(\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"
 
   _DRBD_MAJOR = 147
   _ST_UNCONFIGURED = "Unconfigured"
@@ -747,6 +919,7 @@ class BaseDRBD(BlockDev):
   _ST_CONNECTED = "Connected"
 
   _STATUS_FILE = "/proc/drbd"
   _ST_CONNECTED = "Connected"
 
   _STATUS_FILE = "/proc/drbd"
+  _USERMODE_HELPER_FILE = "/sys/module/drbd/parameters/usermode_helper"
 
   @staticmethod
   def _GetProcData(filename=_STATUS_FILE):
 
   @staticmethod
   def _GetProcData(filename=_STATUS_FILE):
@@ -754,11 +927,7 @@ class BaseDRBD(BlockDev):
 
     """
     try:
 
     """
     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"
     except EnvironmentError, err:
       if err.errno == errno.ENOENT:
         _ThrowError("The file %s cannot be opened, check if the module"
@@ -769,19 +938,20 @@ class BaseDRBD(BlockDev):
       _ThrowError("Can't read any data from %s", filename)
     return data
 
       _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
 
     """
     """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:
     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
       if lresult is not None:
         if old_minor is not None:
           results[old_minor] = old_line
@@ -796,7 +966,7 @@ class BaseDRBD(BlockDev):
     return results
 
   @classmethod
     return results
 
   @classmethod
-  def _GetVersion(cls):
+  def _GetVersion(cls, proc_data):
     """Return the DRBD version.
 
     This will return a dict with keys:
     """Return the DRBD version.
 
     This will return a dict with keys:
@@ -808,7 +978,6 @@ class BaseDRBD(BlockDev):
       - proto2 (only on drbd > 8.2.X)
 
     """
       - proto2 (only on drbd > 8.2.X)
 
     """
-    proc_data = cls._GetProcData()
     first_line = proc_data[0].strip()
     version = cls._VERSION_RE.match(first_line)
     if not version:
     first_line = proc_data[0].strip()
     version = cls._VERSION_RE.match(first_line)
     if not version:
@@ -816,18 +985,35 @@ class BaseDRBD(BlockDev):
                                     first_line)
 
     values = version.groups()
                                     first_line)
 
     values = version.groups()
-    retval = {'k_major': int(values[0]),
-              'k_minor': int(values[1]),
-              'k_point': int(values[2]),
-              'api': int(values[3]),
-              'proto': int(values[4]),
+    retval = {"k_major": int(values[0]),
+              "k_minor": int(values[1]),
+              "k_point": int(values[2]),
+              "api": int(values[3]),
+              "proto": int(values[4]),
              }
     if values[5] is not None:
              }
     if values[5] is not None:
-      retval['proto2'] = values[5]
+      retval["proto2"] = values[5]
 
     return retval
 
   @staticmethod
 
     return retval
 
   @staticmethod
+  def GetUsermodeHelper(filename=_USERMODE_HELPER_FILE):
+    """Returns DRBD usermode_helper currently set.
+
+    """
+    try:
+      helper = utils.ReadFile(filename).splitlines()[0]
+    except EnvironmentError, err:
+      if err.errno == errno.ENOENT:
+        _ThrowError("The file %s cannot be opened, check if the module"
+                    " is loaded (%s)", filename, str(err))
+      else:
+        _ThrowError("Can't read DRBD helper file %s: %s", filename, str(err))
+    if not helper:
+      _ThrowError("Can't read any data from %s", filename)
+    return helper
+
+  @staticmethod
   def _DevPath(minor):
     """Return the path to a drbd device for a given minor.
 
   def _DevPath(minor):
     """Return the path to a drbd device for a given minor.
 
@@ -842,9 +1028,8 @@ class BaseDRBD(BlockDev):
     data = cls._GetProcData()
 
     used_devs = {}
     data = cls._GetProcData()
 
     used_devs = {}
-    valid_line = re.compile("^ *([0-9]+): cs:([^ ]+).*$")
     for line in data:
     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))
       if not match:
         continue
       minor = int(match.group(1))
@@ -883,19 +1068,19 @@ class BaseDRBD(BlockDev):
                   result.fail_reason, result.output)
     try:
       sectors = int(result.stdout)
                   result.fail_reason, result.output)
     try:
       sectors = int(result.stdout)
-    except ValueError:
+    except (TypeError, ValueError):
       _ThrowError("Invalid output from blockdev: '%s'", result.stdout)
       _ThrowError("Invalid output from blockdev: '%s'", result.stdout)
-    bytes = sectors * 512
-    if bytes < 128 * 1024 * 1024: # less than 128MiB
-      _ThrowError("Meta device too small (%.2fMib)", (bytes / 1024 / 1024))
+    num_bytes = sectors * 512
+    if num_bytes < 128 * 1024 * 1024: # less than 128MiB
+      _ThrowError("Meta device too small (%.2fMib)", (num_bytes / 1024 / 1024))
     # the maximum *valid* size of the meta device when living on top
     # of LVM is hard to compute: it depends on the number of stripes
     # and the PE size; e.g. a 2-stripe, 64MB PE will result in a 128MB
     # (normal size), but an eight-stripe 128MB PE will result in a 1GB
     # size meta device; as such, we restrict it to 1GB (a little bit
     # too generous, but making assumptions about PE size is hard)
     # the maximum *valid* size of the meta device when living on top
     # of LVM is hard to compute: it depends on the number of stripes
     # and the PE size; e.g. a 2-stripe, 64MB PE will result in a 128MB
     # (normal size), but an eight-stripe 128MB PE will result in a 1GB
     # size meta device; as such, we restrict it to 1GB (a little bit
     # too generous, but making assumptions about PE size is hard)
-    if bytes > 1024 * 1024 * 1024:
-      _ThrowError("Meta device too big (%.2fMiB)", (bytes / 1024 / 1024))
+    if num_bytes > 1024 * 1024 * 1024:
+      _ThrowError("Meta device too big (%.2fMiB)", (num_bytes / 1024 / 1024))
 
   def Rename(self, new_id):
     """Rename a device.
 
   def Rename(self, new_id):
     """Rename a device.
@@ -925,17 +1110,15 @@ class DRBD8(BaseDRBD):
   # timeout constants
   _NET_RECONFIG_TIMEOUT = 60
 
   # timeout constants
   _NET_RECONFIG_TIMEOUT = 60
 
-  def __init__(self, unique_id, children, size):
+  # command line options for barriers
+  _DISABLE_DISK_OPTION = "--no-disk-barrier"  # -a
+  _DISABLE_DRAIN_OPTION = "--no-disk-drain"   # -D
+  _DISABLE_FLUSH_OPTION = "--no-disk-flushes" # -i
+  _DISABLE_META_FLUSH_OPTION = "--no-md-flushes"  # -m
+
+  def __init__(self, unique_id, children, size, params):
     if children and children.count(None) > 0:
       children = []
     if children and children.count(None) > 0:
       children = []
-    super(DRBD8, self).__init__(unique_id, children, size)
-    self.major = self._DRBD_MAJOR
-    version = self._GetVersion()
-    if version['k_major'] != 8 :
-      _ThrowError("Mismatch in DRBD kernel version and requested ganeti"
-                  " 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:
     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:
@@ -943,6 +1126,18 @@ class DRBD8(BaseDRBD):
     (self._lhost, self._lport,
      self._rhost, self._rport,
      self._aminor, self._secret) = 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, params)
+    self.major = self._DRBD_MAJOR
+    version = self._GetVersion(self._GetProcData())
+    if version["k_major"] != 8:
+      _ThrowError("Mismatch in DRBD kernel version and requested ganeti"
+                  " usage: kernel is %s.%s, ganeti wants 8.x",
+                  version["k_major"], version["k_minor"])
+
     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" %
     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" %
@@ -956,6 +1151,17 @@ class DRBD8(BaseDRBD):
     This will not work if the given minor is in use.
 
     """
     This will not work if the given minor is in use.
 
     """
+    # Zero the metadata first, in order to make sure drbdmeta doesn't
+    # try to auto-detect existing filesystems or similar (see
+    # http://code.google.com/p/ganeti/issues/detail?id=182); we only
+    # care about the first 128MB of data in the device, even though it
+    # can be bigger
+    result = utils.RunCmd([constants.DD_CMD,
+                           "if=/dev/zero", "of=%s" % dev_path,
+                           "bs=1048576", "count=128", "oflag=direct"])
+    if result.failed:
+      _ThrowError("Can't wipe the meta device: %s", result.output)
+
     result = utils.RunCmd(["drbdmeta", "--force", cls._DevPath(minor),
                            "v08", dev_path, "0", "create-md"])
     if result.failed:
     result = utils.RunCmd(["drbdmeta", "--force", cls._DevPath(minor),
                            "v08", dev_path, "0", "create-md"])
     if result.failed:
@@ -971,14 +1177,12 @@ class DRBD8(BaseDRBD):
     """
     data = cls._GetProcData()
 
     """
     data = cls._GetProcData()
 
-    unused_line = re.compile("^ *([0-9]+): cs:Unconfigured$")
-    used_line = re.compile("^ *([0-9]+): cs:")
     highest = None
     for line in data:
     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))
       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)
       if match:
         minor = int(match.group(1))
         highest = max(highest, minor)
@@ -1003,38 +1207,41 @@ class DRBD8(BaseDRBD):
     # pyparsing setup
     lbrace = pyp.Literal("{").suppress()
     rbrace = pyp.Literal("}").suppress()
     # pyparsing setup
     lbrace = pyp.Literal("{").suppress()
     rbrace = pyp.Literal("}").suppress()
+    lbracket = pyp.Literal("[").suppress()
+    rbracket = pyp.Literal("]").suppress()
     semi = pyp.Literal(";").suppress()
     semi = pyp.Literal(";").suppress()
+    colon = pyp.Literal(":").suppress()
     # this also converts the value to an int
     number = pyp.Word(pyp.nums).setParseAction(lambda s, l, t: int(t[0]))
 
     # this also converts the value to an int
     number = pyp.Word(pyp.nums).setParseAction(lambda s, l, t: int(t[0]))
 
-    comment = pyp.Literal ("#") + pyp.Optional(pyp.restOfLine)
+    comment = pyp.Literal("#") + pyp.Optional(pyp.restOfLine)
     defa = pyp.Literal("_is_default").suppress()
     dbl_quote = pyp.Literal('"').suppress()
 
     defa = pyp.Literal("_is_default").suppress()
     dbl_quote = pyp.Literal('"').suppress()
 
-    keyword = pyp.Word(pyp.alphanums + '-')
+    keyword = pyp.Word(pyp.alphanums + "-")
 
     # value types
 
     # value types
-    value = pyp.Word(pyp.alphanums + '_-/.:')
+    value = pyp.Word(pyp.alphanums + "_-/.:")
     quoted = dbl_quote + pyp.CharsNotIn('"') + dbl_quote
     quoted = dbl_quote + pyp.CharsNotIn('"') + dbl_quote
-    addr_type = (pyp.Optional(pyp.Literal("ipv4")).suppress() +
-                 pyp.Optional(pyp.Literal("ipv6")).suppress())
-    addr_port = (addr_type + pyp.Word(pyp.nums + '.') +
-                 pyp.Literal(':').suppress() + number)
+    ipv4_addr = (pyp.Optional(pyp.Literal("ipv4")).suppress() +
+                 pyp.Word(pyp.nums + ".") + colon + number)
+    ipv6_addr = (pyp.Optional(pyp.Literal("ipv6")).suppress() +
+                 pyp.Optional(lbracket) + pyp.Word(pyp.hexnums + ":") +
+                 pyp.Optional(rbracket) + colon + number)
     # meta device, extended syntax
     # meta device, extended syntax
-    meta_value = ((value ^ quoted) + pyp.Literal('[').suppress() +
-                  number + pyp.Word(']').suppress())
+    meta_value = ((value ^ quoted) + lbracket + number + rbracket)
     # device name, extended syntax
     device_value = pyp.Literal("minor").suppress() + number
 
     # a statement
     stmt = (~rbrace + keyword + ~lbrace +
     # device name, extended syntax
     device_value = pyp.Literal("minor").suppress() + number
 
     # a statement
     stmt = (~rbrace + keyword + ~lbrace +
-            pyp.Optional(addr_port ^ value ^ quoted ^ meta_value ^
+            pyp.Optional(ipv4_addr ^ ipv6_addr ^ value ^ quoted ^ meta_value ^
                          device_value) +
             pyp.Optional(defa) + semi +
             pyp.Optional(pyp.restOfLine).suppress())
 
     # an entire section
                          device_value) +
             pyp.Optional(defa) + semi +
             pyp.Optional(pyp.restOfLine).suppress())
 
     # an entire section
-    section_name = pyp.Word(pyp.alphas + '_')
+    section_name = pyp.Word(pyp.alphas + "_")
     section = section_name + lbrace + pyp.ZeroOrMore(pyp.Group(stmt)) + rbrace
 
     bnf = pyp.ZeroOrMore(pyp.Group(section ^ stmt))
     section = section_name + lbrace + pyp.ZeroOrMore(pyp.Group(stmt)) + rbrace
 
     bnf = pyp.ZeroOrMore(pyp.Group(section ^ stmt))
@@ -1150,22 +1357,110 @@ class DRBD8(BaseDRBD):
               info["remote_addr"] == (self._rhost, self._rport))
     return retval
 
               info["remote_addr"] == (self._rhost, self._rport))
     return retval
 
-  @classmethod
-  def _AssembleLocal(cls, minor, backend, meta, size):
+  def _AssembleLocal(self, minor, backend, meta, size):
     """Configure the local part of a DRBD device.
 
     """
     """Configure the local part of a DRBD device.
 
     """
-    args = ["drbdsetup", cls._DevPath(minor), "disk",
+    args = ["drbdsetup", self._DevPath(minor), "disk",
             backend, meta, "0",
             backend, meta, "0",
-            "-d", "%sm" % size,
             "-e", "detach",
             "--create-device"]
             "-e", "detach",
             "--create-device"]
+    if size:
+      args.extend(["-d", "%sm" % size])
+
+    version = self._GetVersion(self._GetProcData())
+    vmaj = version["k_major"]
+    vmin = version["k_minor"]
+    vrel = version["k_point"]
+
+    barrier_args = \
+      self._ComputeDiskBarrierArgs(vmaj, vmin, vrel,
+                                   self.params[constants.LDP_BARRIERS],
+                                   self.params[constants.LDP_NO_META_FLUSH])
+    args.extend(barrier_args)
+
+    if self.params[constants.LDP_DISK_CUSTOM]:
+      args.extend(shlex.split(self.params[constants.LDP_DISK_CUSTOM]))
+
     result = utils.RunCmd(args)
     if result.failed:
       _ThrowError("drbd%d: can't attach local disk: %s", minor, result.output)
 
   @classmethod
     result = utils.RunCmd(args)
     if result.failed:
       _ThrowError("drbd%d: can't attach local disk: %s", minor, result.output)
 
   @classmethod
-  def _AssembleNet(cls, minor, net_info, protocol,
+  def _ComputeDiskBarrierArgs(cls, vmaj, vmin, vrel, disabled_barriers,
+      disable_meta_flush):
+    """Compute the DRBD command line parameters for disk barriers
+
+    Returns a list of the disk barrier parameters as requested via the
+    disabled_barriers and disable_meta_flush arguments, and according to the
+    supported ones in the DRBD version vmaj.vmin.vrel
+
+    If the desired option is unsupported, raises errors.BlockDeviceError.
+
+    """
+    disabled_barriers_set = frozenset(disabled_barriers)
+    if not disabled_barriers_set in constants.DRBD_VALID_BARRIER_OPT:
+      raise errors.BlockDeviceError("%s is not a valid option set for DRBD"
+                                    " barriers" % disabled_barriers)
+
+    args = []
+
+    # The following code assumes DRBD 8.x, with x < 4 and x != 1 (DRBD 8.1.x
+    # does not exist)
+    if not vmaj == 8 and vmin in (0, 2, 3):
+      raise errors.BlockDeviceError("Unsupported DRBD version: %d.%d.%d" %
+                                    (vmaj, vmin, vrel))
+
+    def _AppendOrRaise(option, min_version):
+      """Helper for DRBD options"""
+      if min_version is not None and vrel >= min_version:
+        args.append(option)
+      else:
+        raise errors.BlockDeviceError("Could not use the option %s as the"
+                                      " DRBD version %d.%d.%d does not support"
+                                      " it." % (option, vmaj, vmin, vrel))
+
+    # the minimum version for each feature is encoded via pairs of (minor
+    # version -> x) where x is version in which support for the option was
+    # introduced.
+    meta_flush_supported = disk_flush_supported = {
+      0: 12,
+      2: 7,
+      3: 0,
+      }
+
+    disk_drain_supported = {
+      2: 7,
+      3: 0,
+      }
+
+    disk_barriers_supported = {
+      3: 0,
+      }
+
+    # meta flushes
+    if disable_meta_flush:
+      _AppendOrRaise(cls._DISABLE_META_FLUSH_OPTION,
+                     meta_flush_supported.get(vmin, None))
+
+    # disk flushes
+    if constants.DRBD_B_DISK_FLUSH in disabled_barriers_set:
+      _AppendOrRaise(cls._DISABLE_FLUSH_OPTION,
+                     disk_flush_supported.get(vmin, None))
+
+    # disk drain
+    if constants.DRBD_B_DISK_DRAIN in disabled_barriers_set:
+      _AppendOrRaise(cls._DISABLE_DRAIN_OPTION,
+                     disk_drain_supported.get(vmin, None))
+
+    # disk barriers
+    if constants.DRBD_B_DISK_BARRIERS in disabled_barriers_set:
+      _AppendOrRaise(cls._DISABLE_DISK_OPTION,
+                     disk_barriers_supported.get(vmin, None))
+
+    return args
+
+  def _AssembleNet(self, minor, net_info, protocol,
                    dual_pri=False, hmac=None, secret=None):
     """Configure the network part of the device.
 
                    dual_pri=False, hmac=None, secret=None):
     """Configure the network part of the device.
 
@@ -1174,7 +1469,7 @@ class DRBD8(BaseDRBD):
     if None in net_info:
       # we don't want network connection and actually want to make
       # sure its shutdown
     if None in net_info:
       # we don't want network connection and actually want to make
       # sure its shutdown
-      cls._ShutdownNet(minor)
+      self._ShutdownNet(minor)
       return
 
     # Workaround for a race condition. When DRBD is doing its dance to
       return
 
     # Workaround for a race condition. When DRBD is doing its dance to
@@ -1183,10 +1478,27 @@ class DRBD8(BaseDRBD):
     # sync speed only after setting up both sides can race with DRBD
     # connecting, hence we set it here before telling DRBD anything
     # about its peer.
     # sync speed only after setting up both sides can race with DRBD
     # connecting, hence we set it here before telling DRBD anything
     # about its peer.
-    cls._SetMinorSyncSpeed(minor, constants.SYNC_SPEED)
+    sync_errors = self._SetMinorSyncParams(minor, self.params)
+    if sync_errors:
+      _ThrowError("drbd%d: can't set the synchronization parameters: %s" %
+                  (minor, utils.CommaJoin(sync_errors)))
+
+    if netutils.IP6Address.IsValid(lhost):
+      if not netutils.IP6Address.IsValid(rhost):
+        _ThrowError("drbd%d: can't connect ip %s to ip %s" %
+                    (minor, lhost, rhost))
+      family = "ipv6"
+    elif netutils.IP4Address.IsValid(lhost):
+      if not netutils.IP4Address.IsValid(rhost):
+        _ThrowError("drbd%d: can't connect ip %s to ip %s" %
+                    (minor, lhost, rhost))
+      family = "ipv4"
+    else:
+      _ThrowError("drbd%d: Invalid ip %s" % (minor, lhost))
 
 
-    args = ["drbdsetup", cls._DevPath(minor), "net",
-            "%s:%s" % (lhost, lport), "%s:%s" % (rhost, rport), protocol,
+    args = ["drbdsetup", self._DevPath(minor), "net",
+            "%s:%s:%s" % (family, lhost, lport),
+            "%s:%s:%s" % (family, rhost, rport), protocol,
             "-A", "discard-zero-changes",
             "-B", "consensus",
             "--create-device",
             "-A", "discard-zero-changes",
             "-B", "consensus",
             "--create-device",
@@ -1195,25 +1507,27 @@ class DRBD8(BaseDRBD):
       args.append("-m")
     if hmac and secret:
       args.extend(["-a", hmac, "-x", secret])
       args.append("-m")
     if hmac and secret:
       args.extend(["-a", hmac, "-x", secret])
+
+    if self.params[constants.LDP_NET_CUSTOM]:
+      args.extend(shlex.split(self.params[constants.LDP_NET_CUSTOM]))
+
     result = utils.RunCmd(args)
     if result.failed:
       _ThrowError("drbd%d: can't setup network: %s - %s",
                   minor, result.fail_reason, result.output)
 
     result = utils.RunCmd(args)
     if result.failed:
       _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:
-      info = cls._GetDevInfo(cls._GetShowData(minor))
+    def _CheckNetworkConfig():
+      info = self._GetDevInfo(self._GetShowData(minor))
       if not "local_addr" in info or not "remote_addr" in info:
       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)):
       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):
       _ThrowError("drbd%d: timeout while configuring network", minor)
 
   def AddChildren(self, devices):
@@ -1267,40 +1581,104 @@ class DRBD8(BaseDRBD):
     self._children = []
 
   @classmethod
     self._children = []
 
   @classmethod
-  def _SetMinorSyncSpeed(cls, minor, kbytes):
-    """Set the speed of the DRBD syncer.
+  def _SetMinorSyncParams(cls, minor, params):
+    """Set the parameters of the DRBD syncer.
 
     This is the low-level implementation.
 
     @type minor: int
     @param minor: the drbd minor whose settings we change
 
     This is the low-level implementation.
 
     @type minor: int
     @param minor: the drbd minor whose settings we change
-    @type kbytes: int
-    @param kbytes: the speed in kbytes/second
-    @rtype: boolean
-    @return: the success of the operation
+    @type params: dict
+    @param params: LD level disk parameters related to the synchronization
+    @rtype: list
+    @return: a list of error messages
+
+    """
+
+    args = ["drbdsetup", cls._DevPath(minor), "syncer"]
+    if params[constants.LDP_DYNAMIC_RESYNC]:
+      version = cls._GetVersion(cls._GetProcData())
+      vmin = version["k_minor"]
+      vrel = version["k_point"]
+
+      # By definition we are using 8.x, so just check the rest of the version
+      # number
+      if vmin != 3 or vrel < 9:
+        msg = ("The current DRBD version (8.%d.%d) does not support the "
+               "dynamic resync speed controller" % (vmin, vrel))
+        logging.error(msg)
+        return [msg]
+
+      if params[constants.LDP_PLAN_AHEAD] == 0:
+        msg = ("A value of 0 for c-plan-ahead disables the dynamic sync speed"
+               " controller at DRBD level. If you want to disable it, please"
+               " set the dynamic-resync disk parameter to False.")
+        logging.error(msg)
+        return [msg]
+
+      # add the c-* parameters to args
+      args.extend(["--c-plan-ahead", params[constants.LDP_PLAN_AHEAD],
+                   "--c-fill-target", params[constants.LDP_FILL_TARGET],
+                   "--c-delay-target", params[constants.LDP_DELAY_TARGET],
+                   "--c-max-rate", params[constants.LDP_MAX_RATE],
+                   "--c-min-rate", params[constants.LDP_MIN_RATE],
+                  ])
 
 
-    """
-    result = utils.RunCmd(["drbdsetup", cls._DevPath(minor), "syncer",
-                           "-r", "%d" % kbytes, "--create-device"])
+    else:
+      args.extend(["-r", "%d" % params[constants.LDP_RESYNC_RATE]])
+
+    args.append("--create-device")
+    result = utils.RunCmd(args)
     if result.failed:
     if result.failed:
-      logging.error("Can't change syncer rate: %s - %s",
-                    result.fail_reason, result.output)
-    return not result.failed
+      msg = ("Can't change syncer rate: %s - %s" %
+             (result.fail_reason, result.output))
+      logging.error(msg)
+      return msg
 
 
-  def SetSyncSpeed(self, kbytes):
-    """Set the speed of the DRBD syncer.
+    return []
 
 
-    @type kbytes: int
-    @param kbytes: the speed in kbytes/second
-    @rtype: boolean
+  def SetSyncParams(self, params):
+    """Set the synchronization parameters of the DRBD syncer.
+
+    @type params: dict
+    @param params: LD level disk parameters related to the synchronization
+    @rtype: list
+    @return: a list of error messages, emitted both by the current node and by
+    children. An empty list means no errors
+
+    """
+    if self.minor is None:
+      err = "Not attached during SetSyncParams"
+      logging.info(err)
+      return [err]
+
+    children_result = super(DRBD8, self).SetSyncParams(params)
+    children_result.extend(self._SetMinorSyncParams(self.minor, params))
+    return children_result
+
+  def PauseResumeSync(self, pause):
+    """Pauses or resumes the sync of a DRBD device.
+
+    @param pause: Wether to pause or resume
     @return: the success of the operation
 
     """
     if self.minor is None:
     @return: the success of the operation
 
     """
     if self.minor is None:
-      logging.info("Not attached during SetSyncSpeed")
+      logging.info("Not attached during PauseSync")
       return False
       return False
-    children_result = super(DRBD8, self).SetSyncSpeed(kbytes)
-    return self._SetMinorSyncSpeed(self.minor, kbytes) and children_result
+
+    children_result = super(DRBD8, self).PauseResumeSync(pause)
+
+    if pause:
+      cmd = "pause-sync"
+    else:
+      cmd = "resume-sync"
+
+    result = utils.RunCmd(["drbdsetup", self.dev_path, cmd])
+    if result.failed:
+      logging.error("Can't %s: %s - %s", cmd,
+                    result.fail_reason, result.output)
+    return not result.failed and children_result
 
   def GetProcStatus(self):
     """Return device data from /proc.
 
   def GetProcStatus(self):
     """Return device data from /proc.
@@ -1335,16 +1713,22 @@ class DRBD8(BaseDRBD):
       _ThrowError("drbd%d: can't Attach() in GetSyncStatus", self._aminor)
 
     stats = self.GetProcStatus()
       _ThrowError("drbd%d: can't Attach() in GetSyncStatus", self._aminor)
 
     stats = self.GetProcStatus()
-    ldisk_degraded = not stats.is_disk_uptodate
-    is_degraded = not stats.is_connected
+    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,
 
     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 or ldisk_degraded),
-                                  ldisk_degraded=ldisk_degraded)
+                                  is_degraded=is_degraded,
+                                  ldisk_status=ldisk_status)
 
   def Open(self, force=False):
     """Make the local state primary.
 
   def Open(self, force=False):
     """Make the local state primary.
@@ -1400,31 +1784,42 @@ class DRBD8(BaseDRBD):
       _ThrowError("drbd%d: DRBD disk missing network info in"
                   " DisconnectNet()", self.minor)
 
       _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"
         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)
 
       _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)
 
       logging.info("drbd%d: DisconnectNet: detach took %.3f seconds",
                    self.minor, reconfig_time)
 
@@ -1482,6 +1877,7 @@ class DRBD8(BaseDRBD):
       - if we have a configured device, we try to ensure that it matches
         our config
       - if not, we create it from zero
       - if we have a configured device, we try to ensure that it matches
         our config
       - if not, we create it from zero
+      - anyway, set the device parameters
 
     """
     super(DRBD8, self).Assemble()
 
     """
     super(DRBD8, self).Assemble()
@@ -1495,6 +1891,11 @@ class DRBD8(BaseDRBD):
       # the device
       self._SlowAssemble()
 
       # the device
       self._SlowAssemble()
 
+    sync_errors = self.SetSyncParams(self.params)
+    if sync_errors:
+      _ThrowError("drbd%d: can't set the synchronization parameters: %s" %
+                  (self.minor, utils.CommaJoin(sync_errors)))
+
   def _SlowAssemble(self):
     """Assembles the DRBD device from a (partially) configured device.
 
   def _SlowAssemble(self):
     """Assembles the DRBD device from a (partially) configured device.
 
@@ -1503,6 +1904,8 @@ class DRBD8(BaseDRBD):
     the attach if can return success.
 
     """
     the attach if can return success.
 
     """
+    # TODO: Rewrite to not use a for loop just because there is 'break'
+    # pylint: disable=W0631
     net_data = (self._lhost, self._lport, self._rhost, self._rport)
     for minor in (self._aminor,):
       info = self._GetDevInfo(self._GetShowData(minor))
     net_data = (self._lhost, self._lport, self._rhost, self._rport)
     for minor in (self._aminor,):
       info = self._GetDevInfo(self._GetShowData(minor))
@@ -1639,7 +2042,7 @@ class DRBD8(BaseDRBD):
     self.Shutdown()
 
   @classmethod
     self.Shutdown()
 
   @classmethod
-  def Create(cls, unique_id, children, size):
+  def Create(cls, unique_id, children, size, params):
     """Create a new DRBD8 device.
 
     Since DRBD devices are not created per se, just assembled, this
     """Create a new DRBD8 device.
 
     Since DRBD devices are not created per se, just assembled, this
@@ -1665,9 +2068,9 @@ class DRBD8(BaseDRBD):
                   aminor, meta)
     cls._CheckMetaSize(meta.dev_path)
     cls._InitMeta(aminor, meta.dev_path)
                   aminor, meta)
     cls._CheckMetaSize(meta.dev_path)
     cls._InitMeta(aminor, meta.dev_path)
-    return cls(unique_id, children, size)
+    return cls(unique_id, children, size, params)
 
 
-  def Grow(self, amount):
+  def Grow(self, amount, dryrun):
     """Resize the DRBD device and its backing storage.
 
     """
     """Resize the DRBD device and its backing storage.
 
     """
@@ -1675,7 +2078,10 @@ class DRBD8(BaseDRBD):
       _ThrowError("drbd%d: Grow called while not attached", self._aminor)
     if len(self._children) != 2 or None in self._children:
       _ThrowError("drbd%d: cannot grow diskless device", self.minor)
       _ThrowError("drbd%d: Grow called while not attached", self._aminor)
     if len(self._children) != 2 or None in self._children:
       _ThrowError("drbd%d: cannot grow diskless device", self.minor)
-    self._children[0].Grow(amount)
+    self._children[0].Grow(amount, dryrun)
+    if dryrun:
+      # DRBD does not support dry-run mode, so we'll return here
+      return
     result = utils.RunCmd(["drbdsetup", self.dev_path, "resize", "-s",
                            "%dm" % (self.size + amount)])
     if result.failed:
     result = utils.RunCmd(["drbdsetup", self.dev_path, "resize", "-s",
                            "%dm" % (self.size + amount)])
     if result.failed:
@@ -1690,13 +2096,13 @@ class FileStorage(BlockDev):
   The unique_id for the file device is a (file_driver, file_path) tuple.
 
   """
   The unique_id for the file device is a (file_driver, file_path) tuple.
 
   """
-  def __init__(self, unique_id, children, size):
+  def __init__(self, unique_id, children, size, params):
     """Initalizes a file device backend.
 
     """
     if children:
       raise errors.BlockDeviceError("Invalid setup for file device")
     """Initalizes a file device backend.
 
     """
     if children:
       raise errors.BlockDeviceError("Invalid setup for file device")
-    super(FileStorage, self).__init__(unique_id, children, size)
+    super(FileStorage, self).__init__(unique_id, children, size, params)
     if not isinstance(unique_id, (tuple, list)) or len(unique_id) != 2:
       raise ValueError("Invalid configuration data %s" % str(unique_id))
     self.driver = unique_id[0]
     if not isinstance(unique_id, (tuple, list)) or len(unique_id) != 2:
       raise ValueError("Invalid configuration data %s" % str(unique_id))
     self.driver = unique_id[0]
@@ -1750,6 +2156,34 @@ class FileStorage(BlockDev):
       if err.errno != errno.ENOENT:
         _ThrowError("Can't remove file '%s': %s", self.dev_path, err)
 
       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, dryrun):
+    """Grow the file
+
+    @param amount: the amount (in mebibytes) to grow with
+
+    """
+    # Check that the file exists
+    self.Assemble()
+    current_size = self.GetActualSize()
+    new_size = current_size + amount * 1024 * 1024
+    assert new_size > current_size, "Cannot Grow with a negative amount"
+    # We can't really simulate the growth
+    if dryrun:
+      return
+    try:
+      f = open(self.dev_path, "a+")
+      f.truncate(new_size)
+      f.close()
+    except EnvironmentError, err:
+      _ThrowError("Error in file growth: %", str(err))
+
   def Attach(self):
     """Attach to an existing file.
 
   def Attach(self):
     """Attach to an existing file.
 
@@ -1762,8 +2196,21 @@ class FileStorage(BlockDev):
     self.attached = os.path.exists(self.dev_path)
     return self.attached
 
     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
   @classmethod
-  def Create(cls, unique_id, children, size):
+  def Create(cls, unique_id, children, size, params):
     """Create a new file.
 
     @param size: the size of file in MiB
     """Create a new file.
 
     @param size: the size of file in MiB
@@ -1775,59 +2222,495 @@ class FileStorage(BlockDev):
     if not isinstance(unique_id, (tuple, list)) or len(unique_id) != 2:
       raise ValueError("Invalid configuration data %s" % str(unique_id))
     dev_path = unique_id[1]
     if not isinstance(unique_id, (tuple, list)) or len(unique_id) != 2:
       raise ValueError("Invalid configuration data %s" % str(unique_id))
     dev_path = unique_id[1]
-    if os.path.exists(dev_path):
-      _ThrowError("File already existing: %s", dev_path)
     try:
     try:
-      f = open(dev_path, 'w')
+      fd = os.open(dev_path, os.O_RDWR | os.O_CREAT | os.O_EXCL)
+      f = os.fdopen(fd, "w")
       f.truncate(size * 1024 * 1024)
       f.close()
       f.truncate(size * 1024 * 1024)
       f.close()
-    except IOError, err:
+    except EnvironmentError, err:
+      if err.errno == errno.EEXIST:
+        _ThrowError("File already existing: %s", dev_path)
       _ThrowError("Error in file creation: %", str(err))
 
       _ThrowError("Error in file creation: %", str(err))
 
-    return FileStorage(unique_id, children, size)
+    return FileStorage(unique_id, children, size, params)
+
+
+class PersistentBlockDevice(BlockDev):
+  """A block device with persistent node
+
+  May be either directly attached, or exposed through DM (e.g. dm-multipath).
+  udev helpers are probably required to give persistent, human-friendly
+  names.
+
+  For the time being, pathnames are required to lie under /dev.
+
+  """
+  def __init__(self, unique_id, children, size, params):
+    """Attaches to a static block device.
+
+    The unique_id is a path under /dev.
+
+    """
+    super(PersistentBlockDevice, self).__init__(unique_id, children, size,
+                                                params)
+    if not isinstance(unique_id, (tuple, list)) or len(unique_id) != 2:
+      raise ValueError("Invalid configuration data %s" % str(unique_id))
+    self.dev_path = unique_id[1]
+    if not os.path.realpath(self.dev_path).startswith("/dev/"):
+      raise ValueError("Full path '%s' lies outside /dev" %
+                              os.path.realpath(self.dev_path))
+    # TODO: this is just a safety guard checking that we only deal with devices
+    # we know how to handle. In the future this will be integrated with
+    # external storage backends and possible values will probably be collected
+    # from the cluster configuration.
+    if unique_id[0] != constants.BLOCKDEV_DRIVER_MANUAL:
+      raise ValueError("Got persistent block device of invalid type: %s" %
+                       unique_id[0])
+
+    self.major = self.minor = None
+    self.Attach()
+
+  @classmethod
+  def Create(cls, unique_id, children, size, params):
+    """Create a new device
+
+    This is a noop, we only return a PersistentBlockDevice instance
+
+    """
+    return PersistentBlockDevice(unique_id, children, 0, params)
+
+  def Remove(self):
+    """Remove a device
+
+    This is a noop
+
+    """
+    pass
+
+  def Rename(self, new_id):
+    """Rename this device.
+
+    """
+    _ThrowError("Rename is not supported for PersistentBlockDev storage")
+
+  def Attach(self):
+    """Attach to an existing block device.
+
+
+    """
+    self.attached = False
+    try:
+      st = os.stat(self.dev_path)
+    except OSError, err:
+      logging.error("Error stat()'ing %s: %s", self.dev_path, str(err))
+      return False
+
+    if not stat.S_ISBLK(st.st_mode):
+      logging.error("%s is not a block device", self.dev_path)
+      return False
+
+    self.major = os.major(st.st_rdev)
+    self.minor = os.minor(st.st_rdev)
+    self.attached = True
+
+    return True
+
+  def Assemble(self):
+    """Assemble the device.
+
+    """
+    pass
+
+  def Shutdown(self):
+    """Shutdown the device.
+
+    """
+    pass
+
+  def Open(self, force=False):
+    """Make the device ready for I/O.
+
+    """
+    pass
+
+  def Close(self):
+    """Notifies that the device will no longer be used for I/O.
+
+    """
+    pass
+
+  def Grow(self, amount, dryrun):
+    """Grow the logical volume.
+
+    """
+    _ThrowError("Grow is not supported for PersistentBlockDev storage")
+
+
+class RADOSBlockDevice(BlockDev):
+  """A RADOS Block Device (rbd).
+
+  This class implements the RADOS Block Device for the backend. You need
+  the rbd kernel driver, the RADOS Tools and a working RADOS cluster for
+  this to be functional.
+
+  """
+  def __init__(self, unique_id, children, size, params):
+    """Attaches to an rbd device.
+
+    """
+    super(RADOSBlockDevice, self).__init__(unique_id, children, size, params)
+    if not isinstance(unique_id, (tuple, list)) or len(unique_id) != 2:
+      raise ValueError("Invalid configuration data %s" % str(unique_id))
+
+    self.driver, self.rbd_name = unique_id
+
+    self.major = self.minor = None
+    self.Attach()
+
+  @classmethod
+  def Create(cls, unique_id, children, size, params):
+    """Create a new rbd device.
+
+    Provision a new rbd volume inside a RADOS pool.
+
+    """
+    if not isinstance(unique_id, (tuple, list)) or len(unique_id) != 2:
+      raise errors.ProgrammerError("Invalid configuration data %s" %
+                                   str(unique_id))
+    rbd_pool = params[constants.LDP_POOL]
+    rbd_name = unique_id[1]
+
+    # Provision a new rbd volume (Image) inside the RADOS cluster.
+    cmd = [constants.RBD_CMD, "create", "-p", rbd_pool,
+           rbd_name, "--size", "%s" % size]
+    result = utils.RunCmd(cmd)
+    if result.failed:
+      _ThrowError("rbd creation failed (%s): %s",
+                  result.fail_reason, result.output)
+
+    return RADOSBlockDevice(unique_id, children, size, params)
+
+  def Remove(self):
+    """Remove the rbd device.
+
+    """
+    rbd_pool = self.params[constants.LDP_POOL]
+    rbd_name = self.unique_id[1]
+
+    if not self.minor and not self.Attach():
+      # The rbd device doesn't exist.
+      return
+
+    # First shutdown the device (remove mappings).
+    self.Shutdown()
+
+    # Remove the actual Volume (Image) from the RADOS cluster.
+    cmd = [constants.RBD_CMD, "rm", "-p", rbd_pool, rbd_name]
+    result = utils.RunCmd(cmd)
+    if result.failed:
+      _ThrowError("Can't remove Volume from cluster with rbd rm: %s - %s",
+                  result.fail_reason, result.output)
+
+  def Rename(self, new_id):
+    """Rename this device.
+
+    """
+    pass
+
+  def Attach(self):
+    """Attach to an existing rbd device.
+
+    This method maps the rbd volume that matches our name with
+    an rbd device and then attaches to this device.
+
+    """
+    self.attached = False
+
+    # Map the rbd volume to a block device under /dev
+    self.dev_path = self._MapVolumeToBlockdev(self.unique_id)
+
+    try:
+      st = os.stat(self.dev_path)
+    except OSError, err:
+      logging.error("Error stat()'ing %s: %s", self.dev_path, str(err))
+      return False
+
+    if not stat.S_ISBLK(st.st_mode):
+      logging.error("%s is not a block device", self.dev_path)
+      return False
+
+    self.major = os.major(st.st_rdev)
+    self.minor = os.minor(st.st_rdev)
+    self.attached = True
+
+    return True
+
+  def _MapVolumeToBlockdev(self, unique_id):
+    """Maps existing rbd volumes to block devices.
+
+    This method should be idempotent if the mapping already exists.
+
+    @rtype: string
+    @return: the block device path that corresponds to the volume
+
+    """
+    pool = self.params[constants.LDP_POOL]
+    name = unique_id[1]
+
+    # Check if the mapping already exists.
+    showmap_cmd = [constants.RBD_CMD, "showmapped", "-p", pool]
+    result = utils.RunCmd(showmap_cmd)
+    if result.failed:
+      _ThrowError("rbd showmapped failed (%s): %s",
+                  result.fail_reason, result.output)
+
+    rbd_dev = self._ParseRbdShowmappedOutput(result.output, name)
+
+    if rbd_dev:
+      # The mapping exists. Return it.
+      return rbd_dev
+
+    # The mapping doesn't exist. Create it.
+    map_cmd = [constants.RBD_CMD, "map", "-p", pool, name]
+    result = utils.RunCmd(map_cmd)
+    if result.failed:
+      _ThrowError("rbd map failed (%s): %s",
+                  result.fail_reason, result.output)
+
+    # Find the corresponding rbd device.
+    showmap_cmd = [constants.RBD_CMD, "showmapped", "-p", pool]
+    result = utils.RunCmd(showmap_cmd)
+    if result.failed:
+      _ThrowError("rbd map succeeded, but showmapped failed (%s): %s",
+                  result.fail_reason, result.output)
+
+    rbd_dev = self._ParseRbdShowmappedOutput(result.output, name)
+
+    if not rbd_dev:
+      _ThrowError("rbd map succeeded, but could not find the rbd block"
+                  " device in output of showmapped, for volume: %s", name)
+
+    # The device was successfully mapped. Return it.
+    return rbd_dev
+
+  @staticmethod
+  def _ParseRbdShowmappedOutput(output, volume_name):
+    """Parse the output of `rbd showmapped'.
+
+    This method parses the output of `rbd showmapped' and returns
+    the rbd block device path (e.g. /dev/rbd0) that matches the
+    given rbd volume.
+
+    @type output: string
+    @param output: the whole output of `rbd showmapped'
+    @type volume_name: string
+    @param volume_name: the name of the volume whose device we search for
+    @rtype: string or None
+    @return: block device path if the volume is mapped, else None
+
+    """
+    allfields = 5
+    volumefield = 2
+    devicefield = 4
+
+    field_sep = "\t"
+
+    lines = output.splitlines()
+    splitted_lines = map(lambda l: l.split(field_sep), lines)
+
+    # Check empty output.
+    if not splitted_lines:
+      _ThrowError("rbd showmapped returned empty output")
+
+    # Check showmapped header line, to determine number of fields.
+    field_cnt = len(splitted_lines[0])
+    if field_cnt != allfields:
+      _ThrowError("Cannot parse rbd showmapped output because its format"
+                  " seems to have changed; expected %s fields, found %s",
+                  allfields, field_cnt)
+
+    matched_lines = \
+      filter(lambda l: len(l) == allfields and l[volumefield] == volume_name,
+             splitted_lines)
+
+    if len(matched_lines) > 1:
+      _ThrowError("The rbd volume %s is mapped more than once."
+                  " This shouldn't happen, try to unmap the extra"
+                  " devices manually.", volume_name)
+
+    if matched_lines:
+      # rbd block device found. Return it.
+      rbd_dev = matched_lines[0][devicefield]
+      return rbd_dev
+
+    # The given volume is not mapped.
+    return None
+
+  def Assemble(self):
+    """Assemble the device.
+
+    """
+    pass
+
+  def Shutdown(self):
+    """Shutdown the device.
+
+    """
+    if not self.minor and not self.Attach():
+      # The rbd device doesn't exist.
+      return
+
+    # Unmap the block device from the Volume.
+    self._UnmapVolumeFromBlockdev(self.unique_id)
+
+    self.minor = None
+    self.dev_path = None
+
+  def _UnmapVolumeFromBlockdev(self, unique_id):
+    """Unmaps the rbd device from the Volume it is mapped.
+
+    Unmaps the rbd device from the Volume it was previously mapped to.
+    This method should be idempotent if the Volume isn't mapped.
+
+    """
+    pool = self.params[constants.LDP_POOL]
+    name = unique_id[1]
+
+    # Check if the mapping already exists.
+    showmap_cmd = [constants.RBD_CMD, "showmapped", "-p", pool]
+    result = utils.RunCmd(showmap_cmd)
+    if result.failed:
+      _ThrowError("rbd showmapped failed [during unmap](%s): %s",
+                  result.fail_reason, result.output)
+
+    rbd_dev = self._ParseRbdShowmappedOutput(result.output, name)
+
+    if rbd_dev:
+      # The mapping exists. Unmap the rbd device.
+      unmap_cmd = [constants.RBD_CMD, "unmap", "%s" % rbd_dev]
+      result = utils.RunCmd(unmap_cmd)
+      if result.failed:
+        _ThrowError("rbd unmap failed (%s): %s",
+                    result.fail_reason, result.output)
+
+  def Open(self, force=False):
+    """Make the device ready for I/O.
+
+    """
+    pass
+
+  def Close(self):
+    """Notifies that the device will no longer be used for I/O.
+
+    """
+    pass
+
+  def Grow(self, amount, dryrun):
+    """Grow the Volume.
+
+    @type amount: integer
+    @param amount: the amount (in mebibytes) to grow with
+    @type dryrun: boolean
+    @param dryrun: whether to execute the operation in simulation mode
+        only, without actually increasing the size
+
+    """
+    if not self.Attach():
+      _ThrowError("Can't attach to rbd device during Grow()")
+
+    if dryrun:
+      # the rbd tool does not support dry runs of resize operations.
+      # Since rbd volumes are thinly provisioned, we assume
+      # there is always enough free space for the operation.
+      return
+
+    rbd_pool = self.params[constants.LDP_POOL]
+    rbd_name = self.unique_id[1]
+    new_size = self.size + amount
+
+    # Resize the rbd volume (Image) inside the RADOS cluster.
+    cmd = [constants.RBD_CMD, "resize", "-p", rbd_pool,
+           rbd_name, "--size", "%s" % new_size]
+    result = utils.RunCmd(cmd)
+    if result.failed:
+      _ThrowError("rbd resize failed (%s): %s",
+                  result.fail_reason, result.output)
 
 
 DEV_MAP = {
   constants.LD_LV: LogicalVolume,
   constants.LD_DRBD8: DRBD8,
 
 
 DEV_MAP = {
   constants.LD_LV: LogicalVolume,
   constants.LD_DRBD8: DRBD8,
-  constants.LD_FILE: FileStorage,
+  constants.LD_BLOCKDEV: PersistentBlockDevice,
+  constants.LD_RBD: RADOSBlockDevice,
   }
 
   }
 
+if constants.ENABLE_FILE_STORAGE or constants.ENABLE_SHARED_FILE_STORAGE:
+  DEV_MAP[constants.LD_FILE] = FileStorage
+
 
 
-def FindDevice(dev_type, unique_id, children, size):
+def _VerifyDiskType(dev_type):
+  if dev_type not in DEV_MAP:
+    raise errors.ProgrammerError("Invalid block device type '%s'" % dev_type)
+
+
+def FindDevice(disk, children):
   """Search for an existing, assembled device.
 
   This will succeed only if the device exists and is assembled, but it
   does not do any actions in order to activate the device.
 
   """Search for an existing, assembled device.
 
   This will succeed only if the device exists and is assembled, but it
   does not do any actions in order to activate the device.
 
+  @type disk: L{objects.Disk}
+  @param disk: the disk object to find
+  @type children: list of L{bdev.BlockDev}
+  @param children: the list of block devices that are children of the device
+                  represented by the disk parameter
+
   """
   """
-  if dev_type not in DEV_MAP:
-    raise errors.ProgrammerError("Invalid block device type '%s'" % dev_type)
-  device = DEV_MAP[dev_type](unique_id, children, size)
+  _VerifyDiskType(disk.dev_type)
+  dev_params = objects.FillDict(constants.DISK_LD_DEFAULTS[disk.dev_type],
+                                disk.params)
+  device = DEV_MAP[disk.dev_type](disk.physical_id, children, disk.size,
+                                  dev_params)
   if not device.attached:
     return None
   return device
 
 
   if not device.attached:
     return None
   return device
 
 
-def Assemble(dev_type, unique_id, children, size):
+def Assemble(disk, children):
   """Try to attach or assemble an existing device.
 
   This will attach to assemble the device, as needed, to bring it
   fully up. It must be safe to run on already-assembled devices.
 
   """Try to attach or assemble an existing device.
 
   This will attach to assemble the device, as needed, to bring it
   fully up. It must be safe to run on already-assembled devices.
 
+  @type disk: L{objects.Disk}
+  @param disk: the disk object to assemble
+  @type children: list of L{bdev.BlockDev}
+  @param children: the list of block devices that are children of the device
+                  represented by the disk parameter
+
   """
   """
-  if dev_type not in DEV_MAP:
-    raise errors.ProgrammerError("Invalid block device type '%s'" % dev_type)
-  device = DEV_MAP[dev_type](unique_id, children, size)
+  _VerifyDiskType(disk.dev_type)
+  dev_params = objects.FillDict(constants.DISK_LD_DEFAULTS[disk.dev_type],
+                                disk.params)
+  device = DEV_MAP[disk.dev_type](disk.physical_id, children, disk.size,
+                                  dev_params)
   device.Assemble()
   return device
 
 
   device.Assemble()
   return device
 
 
-def Create(dev_type, unique_id, children, size):
+def Create(disk, children):
   """Create a device.
 
   """Create a device.
 
+  @type disk: L{objects.Disk}
+  @param disk: the disk object to create
+  @type children: list of L{bdev.BlockDev}
+  @param children: the list of block devices that are children of the device
+                  represented by the disk parameter
+
   """
   """
-  if dev_type not in DEV_MAP:
-    raise errors.ProgrammerError("Invalid block device type '%s'" % dev_type)
-  device = DEV_MAP[dev_type].Create(unique_id, children, size)
+  _VerifyDiskType(disk.dev_type)
+  dev_params = objects.FillDict(constants.DISK_LD_DEFAULTS[disk.dev_type],
+                                disk.params)
+  device = DEV_MAP[disk.dev_type].Create(disk.physical_id, children, disk.size,
+                                         dev_params)
   return device
   return device