Add RBD connector

This connector allows RBD volumes to be accessed for I/O operations such
as volume migration.  The intent is to slowly move cinder's RBD code
that could be shared into this connector so that we may eventually share
with nova.

Blueprint: generic-volume-migration
Blueprint: add-rbd-support-to-cinder-brick
Change-Id: I9de951d01d27b983de724e214b4b89fd55f19aa8

os_brick/initiator/connector.py

@@ -41,6 +41,7 @@ from os_brick import exception
 from os_brick import executor
 from os_brick.initiator import host_driver
 from os_brick.initiator import linuxfc
+from os_brick.initiator import linuxrbd
 from os_brick.initiator import linuxscsi
 from os_brick.remotefs import remotefs
 from os_brick.i18n import _, _LE, _LW
@@ -201,6 +202,12 @@ class InitiatorConnector(executor.Executor):
                                  execute=execute,
                                  device_scan_attempts=device_scan_attempts,
                                  *args, **kwargs)
+        elif protocol == "RBD":
+            return RBDConnector(root_helper=root_helper,
+                                driver=driver,
+                                execute=execute,
+                                device_scan_attempts=device_scan_attempts,
+                                *args, **kwargs)
         else:
             msg = (_("Invalid InitiatorConnector protocol "
                      "specified %(protocol)s") %
@@ -1204,6 +1211,62 @@ class RemoteFsConnector(InitiatorConnector):
         """No need to do anything to disconnect a volume in a filesystem."""
 
 
+class RBDConnector(InitiatorConnector):
+    """"Connector class to attach/detach RBD volumes."""
+
+    def __init__(self, root_helper, driver=None,
+                 execute=putils.execute, use_multipath=False,
+                 device_scan_attempts=DEVICE_SCAN_ATTEMPTS_DEFAULT,
+                 *args, **kwargs):
+
+        super(RBDConnector, self).__init__(root_helper, driver=driver,
+                                           execute=execute,
+                                           device_scan_attempts=
+                                           device_scan_attempts,
+                                           *args, **kwargs)
+
+    def connect_volume(self, connection_properties):
+        """Connect to a volume."""
+        try:
+            user = connection_properties['auth_username']
+            pool, volume = connection_properties['name'].split('/')
+        except IndexError:
+            msg = _("Connect volume failed, malformed connection properties")
+            raise exception.BrickException(msg=msg)
+
+        rbd_client = linuxrbd.RBDClient(user, pool)
+        rbd_volume = linuxrbd.RBDVolume(rbd_client, volume)
+        rbd_handle = linuxrbd.RBDVolumeIOWrapper(rbd_volume)
+
+        return {'path': rbd_handle}
+
+    def disconnect_volume(self, connection_properties, device_info):
+        """Disconnect a volume."""
+        rbd_handle = device_info.get('path', None)
+        if rbd_handle is not None:
+            rbd_handle.close()
+
+    def check_valid_device(self, path, run_as_root=True):
+        """Verify an existing RBD handle is connected and valid."""
+        rbd_handle = path
+
+        if rbd_handle is None:
+            return False
+
+        original_offset = rbd_handle.tell()
+
+        try:
+            rbd_handle.read(4096)
+        except Exception as e:
+            LOG.error(_LE("Failed to access RBD device handle: %(error)s"),
+                      {"error": e})
+            return False
+        finally:
+            rbd_handle.seek(original_offset, 0)
+
+        return True
+
+
 class LocalConnector(InitiatorConnector):
     """"Connector class to attach/detach File System backed volumes."""
 

os_brick/initiator/linuxrbd.py

@@ -0,0 +1,192 @@
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not
+# use this file except in compliance with the License. You may obtain a copy of
+# the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations under
+# the License.
+
+""" Generic RBD connection utilities."""
+
+import io
+from oslo_log import log as logging
+from oslo_utils import encodeutils
+
+from os_brick import exception
+from os_brick.i18n import _, _LE, _LW
+
+try:
+    import rados
+    import rbd
+except ImportError:
+    rados = None
+    rbd = None
+
+
+LOG = logging.getLogger(__name__)
+
+
+class RBDClient(object):
+
+    def __init__(self, user, pool, *args, **kwargs):
+
+        self.rbd_user = user
+        self.rbd_pool = pool
+
+        for attr in ['rbd_user', 'rbd_pool']:
+            val = getattr(self, attr)
+            if val is not None:
+                setattr(self, attr, encodeutils.safe_encode(val))
+
+        # allow these to be overridden for testing
+        self.rados = kwargs.get('rados', rados)
+        self.rbd = kwargs.get('rbd', rbd)
+
+        if self.rados is None:
+            raise exception.InvalidParameterValue(
+                err=_('rados module required'))
+        if self.rbd is None:
+            raise exception.InvalidParameterValue(
+                err=_('rbd module required'))
+
+        self.rbd_conf = kwargs.get('conffile', '/etc/ceph/ceph.conf')
+
+        self.client, self.ioctx = self.connect()
+
+    def connect(self):
+        client = self.rados.Rados(rados_id=self.rbd_user,
+                                  conffile=self.rbd_conf)
+
+        try:
+            client.connect()
+            ioctx = client.open_ioctx(self.rbd_pool)
+            return client, ioctx
+        except self.rados.Error:
+            msg = _("Error connecting to ceph cluster.")
+            LOG.exception(msg)
+            # shutdown cannot raise an exception
+            client.shutdown()
+            raise exception.BrickException(message=msg)
+
+    def disconnect(self):
+        # closing an ioctx cannot raise an exception
+        self.ioctx.close()
+        self.client.shutdown()
+
+
+class RBDVolume(object):
+    """Context manager for dealing with an existing rbd volume."""
+
+    def __init__(self, client, name, snapshot=None, read_only=False):
+        if snapshot is not None:
+            snapshot = encodeutils.safe_encode(snapshot)
+
+        try:
+            self.image = client.rbd.Image(client.ioctx,
+                                          encodeutils.safe_encode(name),
+                                          snapshot=snapshot,
+                                          read_only=read_only)
+        except client.rbd.Error:
+            LOG.exception(_LE("error opening rbd image %s"), name)
+            client.disconnect()
+            raise
+
+        self.client = client
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, type_, value, traceback):
+        try:
+            self.image.close()
+        finally:
+            self.client.disconnect()
+
+    def __getattr__(self, attrib):
+        return getattr(self.image, attrib)
+
+
+class RBDVolumeIOWrapper(io.RawIOBase):
+    """Enables LibRBD.Image objects to be treated as Python IO objects.
+
+    Calling unimplemented interfaces will raise IOError.
+    """
+
+    def __init__(self, rbd_volume):
+        super(RBDVolumeIOWrapper, self).__init__()
+        self._rbd_volume = rbd_volume
+        self._offset = 0
+
+    def _inc_offset(self, length):
+        self._offset += length
+
+    def read(self, length=None):
+        offset = self._offset
+        total = self._rbd_volume.image.size()
+
+        # NOTE(dosaboy): posix files do not barf if you read beyond their
+        # length (they just return nothing) but rbd images do so we need to
+        # return empty string if we have reached the end of the image.
+        if (offset >= total):
+            return ''
+
+        if length is None:
+            length = total
+
+        if (offset + length) > total:
+            length = total - offset
+
+        self._inc_offset(length)
+        return self._rbd_volume.image.read(int(offset), int(length))
+
+    def write(self, data):
+        self._rbd_volume.image.write(data, self._offset)
+        self._inc_offset(len(data))
+
+    def seekable(self):
+        return True
+
+    def seek(self, offset, whence=0):
+        if whence == 0:
+            new_offset = offset
+        elif whence == 1:
+            new_offset = self._offset + offset
+        elif whence == 2:
+            new_offset = self._rbd_volume.image.size()
+            new_offset += offset
+        else:
+            raise IOError(_("Invalid argument - whence=%s not supported") %
+                          (whence))
+
+        if (new_offset < 0):
+            raise IOError(_("Invalid argument"))
+
+        self._offset = new_offset
+
+    def tell(self):
+        return self._offset
+
+    def flush(self):
+        try:
+            self._rbd_volume.image.flush()
+        except AttributeError:
+            LOG.warning(_LW("flush() not supported in this version of librbd"))
+
+    def fileno(self):
+        """RBD does not have support for fileno() so we raise IOError.
+
+        Raising IOError is recommended way to notify caller that interface is
+        not supported - see http://docs.python.org/2/library/io.html#io.IOBase
+        """
+        raise IOError(_("fileno() not supported by RBD()"))
+
+    # NOTE(dosaboy): if IO object is not closed explicitly, Python auto closes
+    # it which, if this is not overridden, calls flush() prior to close which
+    # in this case is unwanted since the rbd image may have been closed prior
+    # to the autoclean - currently triggering a segfault in librbd.
+    def close(self):
+        pass

os_brick/tests/initiator/test_connector.py

@@ -29,6 +29,7 @@ from os_brick.i18n import _LE
 from os_brick.initiator import connector
 from os_brick.initiator import host_driver
 from os_brick.initiator import linuxfc
+from os_brick.initiator import linuxrbd
 from os_brick.initiator import linuxscsi
 from os_brick.openstack.common import loopingcall
 from os_brick.remotefs import remotefs
@@ -1496,3 +1497,55 @@ Request Succeeded
         self.assertRaises(exception.BrickException,
                           self.connector.disconnect_volume,
                           cprops, None)
+
+
+class RBDConnectorTestCase(ConnectorTestCase):
+
+    def setUp(self):
+        super(RBDConnectorTestCase, self).setUp()
+
+        self.user = 'fake_user'
+        self.pool = 'fake_pool'
+        self.volume = 'fake_volume'
+
+        self.connection_properties = {
+            'auth_username': self.user,
+            'name': '%s/%s' % (self.pool, self.volume),
+        }
+
+    @mock.patch('os_brick.initiator.linuxrbd.rbd')
+    @mock.patch('os_brick.initiator.linuxrbd.rados')
+    def test_connect_volume(self, mock_rados, mock_rbd):
+        """Test the connect volume case."""
+        rbd = connector.RBDConnector(None)
+        device_info = rbd.connect_volume(self.connection_properties)
+
+        # Ensure rados is instantiated correctly
+        mock_rados.Rados.assert_called_once_with(
+            rados_id=self.user,
+            conffile='/etc/ceph/ceph.conf')
+
+        # Ensure correct calls to connect to cluster
+        mock_rados.Rados.return_value.connect.assert_called_once()
+        mock_rados.Rados.return_value.open_ioctx.assert_called_once_with(
+            self.pool)
+
+        # Ensure rbd image is instantiated correctly
+        mock_rbd.Image.assert_called_once_with(
+            mock_rados.Rados.return_value.open_ioctx.return_value,
+            self.volume, read_only=False, snapshot=None)
+
+        # Ensure expected object is returned correctly
+        self.assertTrue(isinstance(device_info['path'],
+                                   linuxrbd.RBDVolumeIOWrapper))
+
+    @mock.patch('os_brick.initiator.linuxrbd.rbd')
+    @mock.patch('os_brick.initiator.linuxrbd.rados')
+    @mock.patch.object(linuxrbd.RBDVolumeIOWrapper, 'close')
+    def test_disconnect_volume(self, volume_close, mock_rados, mock_rbd):
+        """Test the disconnect volume case."""
+        rbd = connector.RBDConnector(None)
+        device_info = rbd.connect_volume(self.connection_properties)
+        rbd.disconnect_volume(self.connection_properties, device_info)
+
+        volume_close.assert_called_once()

os_brick/tests/initiator/test_linuxrbd.py

@@ -0,0 +1,119 @@
+# Licensed under the Apache License, Version 2.0 (the "License"); you may not
+# use this file except in compliance with the License. You may obtain a copy of
+# the License at
+#
+#      http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+# License for the specific language governing permissions and limitations under
+# the License.
+
+import mock
+
+from os_brick.initiator import linuxrbd
+from os_brick.tests import base
+
+
+class RBDVolumeIOWrapperTestCase(base.TestCase):
+
+    def setUp(self):
+        super(RBDVolumeIOWrapperTestCase, self).setUp()
+        self.mock_volume = mock.Mock()
+        self.mock_volume_wrapper = \
+            linuxrbd.RBDVolumeIOWrapper(self.mock_volume)
+        self.data_length = 1024
+        self.full_data = 'abcd' * 256
+
+    def test_init(self):
+        self.assertEqual(self.mock_volume,
+                         self.mock_volume_wrapper._rbd_volume)
+        self.assertEqual(0, self.mock_volume_wrapper._offset)
+
+    def test_inc_offset(self):
+        self.mock_volume_wrapper._inc_offset(10)
+        self.mock_volume_wrapper._inc_offset(10)
+        self.assertEqual(20, self.mock_volume_wrapper._offset)
+
+    def test_read(self):
+
+        def mock_read(offset, length):
+            return self.full_data[offset:length]
+
+        self.mock_volume.image.read.side_effect = mock_read
+        self.mock_volume.image.size.return_value = self.data_length
+
+        data = self.mock_volume_wrapper.read()
+        self.assertEqual(self.full_data, data)
+
+        data = self.mock_volume_wrapper.read()
+        self.assertEqual('', data)
+
+        self.mock_volume_wrapper.seek(0)
+        data = self.mock_volume_wrapper.read()
+        self.assertEqual(self.full_data, data)
+
+        self.mock_volume_wrapper.seek(0)
+        data = self.mock_volume_wrapper.read(10)
+        self.assertEqual(self.full_data[:10], data)
+
+    def test_write(self):
+        self.mock_volume_wrapper.write(self.full_data)
+        self.assertEqual(1024, self.mock_volume_wrapper._offset)
+
+    def test_seekable(self):
+        self.assertTrue(self.mock_volume_wrapper.seekable)
+
+    def test_seek(self):
+        self.assertEqual(0, self.mock_volume_wrapper._offset)
+        self.mock_volume_wrapper.seek(10)
+        self.assertEqual(10, self.mock_volume_wrapper._offset)
+        self.mock_volume_wrapper.seek(10)
+        self.assertEqual(10, self.mock_volume_wrapper._offset)
+        self.mock_volume_wrapper.seek(10, 1)
+        self.assertEqual(20, self.mock_volume_wrapper._offset)
+
+        self.mock_volume_wrapper.seek(0)
+        self.mock_volume_wrapper.write(self.full_data)
+        self.mock_volume.image.size.return_value = self.data_length
+        self.mock_volume_wrapper.seek(0)
+        self.assertEqual(0, self.mock_volume_wrapper._offset)
+
+        self.mock_volume_wrapper.seek(10, 2)
+        self.assertEqual(self.data_length + 10,
+                         self.mock_volume_wrapper._offset)
+        self.mock_volume_wrapper.seek(-10, 2)
+        self.assertEqual(self.data_length - 10,
+                         self.mock_volume_wrapper._offset)
+
+        # test exceptions.
+        self.assertRaises(IOError, self.mock_volume_wrapper.seek, 0, 3)
+        self.assertRaises(IOError, self.mock_volume_wrapper.seek, -1)
+        # offset should not have been changed by any of the previous
+        # operations.
+        self.assertEqual(self.data_length - 10,
+                         self.mock_volume_wrapper._offset)
+
+    def test_tell(self):
+        self.assertEqual(0, self.mock_volume_wrapper.tell())
+        self.mock_volume_wrapper._inc_offset(10)
+        self.assertEqual(10, self.mock_volume_wrapper.tell())
+
+    def test_flush(self):
+        with mock.patch.object(linuxrbd, 'LOG') as mock_logger:
+            self.mock_volume.image.flush = mock.Mock()
+            self.mock_volume_wrapper.flush()
+            self.mock_volume.image.flush.assert_called_once()
+            self.mock_volume.image.flush.reset_mock()
+            # this should be caught and logged silently.
+            self.mock_volume.image.flush.side_effect = AttributeError
+            self.mock_volume_wrapper.flush()
+            self.mock_volume.image.flush.assert_called_once()
+            mock_logger.warning.assert_called_once()
+
+    def test_fileno(self):
+        self.assertRaises(IOError, self.mock_volume_wrapper.fileno)
+
+    def test_close(self):
+        self.mock_volume_wrapper.close()