Implement Cinder attach workflow

The attach workflow is as following:
* reserve_volume in Cinder
* initialize_connection in Cinder
* connect_volume in os_brick
* attach in Cinder
If the workflow executes sucessfully, the Cinder volume
is attached to the compute host as a device. Zun will
mount the device to the filesystem in order to use it.

The detach workflow is as following:
* begin_detach in Cinder
* disconnect_volume in os_brick
* terminate_connection in Cinder
* detach in Cinder

Partial-Implements: blueprint direct-cinder-integration
Change-Id: Ib47e76d14b3dc036f903a3988bb138288cdb601e

devstack/lib/zun

@@ -181,6 +181,7 @@ function create_zun_conf {
         iniset $ZUN_CONF DEFAULT db_type sql
     fi
     iniset $ZUN_CONF DEFAULT debug "$ENABLE_DEBUG_LOG_LEVEL"
+    iniset $ZUN_CONF DEFAULT my_ip "$HOST_IP"
     iniset $ZUN_CONF oslo_messaging_rabbit rabbit_userid $RABBIT_USERID
     iniset $ZUN_CONF oslo_messaging_rabbit rabbit_password $RABBIT_PASSWORD
     iniset $ZUN_CONF oslo_messaging_rabbit rabbit_host $RABBIT_HOST

requirements.txt

@@ -12,6 +12,7 @@ python-etcd>=0.4.3 # MIT License
 python-glanceclient>=2.8.0 # Apache-2.0
 python-neutronclient>=6.3.0 # Apache-2.0
 python-novaclient>=9.1.0 # Apache-2.0
+python-cinderclient>=3.2.0  # Apache-2.0
 oslo.i18n>=3.15.3 # Apache-2.0
 oslo.log>=3.30.0 # Apache-2.0
 oslo.concurrency>=3.20.0 # Apache-2.0
@@ -26,6 +27,7 @@ oslo.context!=2.19.1,>=2.14.0 # Apache-2.0
 oslo.utils>=3.28.0 # Apache-2.0
 oslo.db>=4.27.0 # Apache-2.0
 os-vif>=1.7.0 # Apache-2.0
+os-brick>=1.15.2  # Apache-2.0
 six>=1.9.0 # MIT
 WSME>=0.8.0 # MIT
 SQLAlchemy!=1.1.5,!=1.1.6,!=1.1.7,!=1.1.8,>=1.0.10 # MIT

zun/common/clients.py

@@ -12,6 +12,7 @@
 # License for the specific language governing permissions and limitations
 # under the License.
 
+from cinderclient import client as cinderclient
 from glanceclient import client as glanceclient
 from neutronclient.v2_0 import client as neutronclient
 from novaclient import client as novaclient
@@ -30,6 +31,7 @@ class OpenStackClients(object):
         self._glance = None
         self._nova = None
         self._neutron = None
+        self._cinder = None
 
     def url_for(self, **kwargs):
         return self.keystone().session.get_endpoint(**kwargs)
@@ -107,3 +109,21 @@ class OpenStackClients(object):
                                              endpoint_type=endpoint_type)
 
         return self._neutron
+
+    @exception.wrap_keystone_exception
+    def cinder(self):
+        if self._cinder:
+            return self._cinder
+
+        cinder_api_version = self._get_client_option('cinder', 'api_version')
+        endpoint_type = self._get_client_option('cinder', 'endpoint_type')
+        kwargs = {
+            'session': self.keystone().session,
+            'endpoint_type': endpoint_type,
+            'cacert': self._get_client_option('cinder', 'ca_file'),
+            'insecure': self._get_client_option('cinder', 'insecure')
+        }
+        self._cinder = cinderclient.Client(version=cinder_api_version,
+                                           **kwargs)
+
+        return self._cinder

zun/common/exception.py

@@ -374,6 +374,10 @@ class VolumeMappingNotFound(HTTPNotFound):
     message = _("Volume mapping %(volume_mapping)s could not be found.")
 
 
+class VolumeNotFound(HTTPNotFound):
+    message = _("Volume %(volume)s could not be found.")
+
+
 class ImageNotFound(Invalid):
     message = _("Image %(image)s could not be found.")
 
@@ -434,6 +438,14 @@ class PortInUse(Invalid):
     message = _("Port %(port)s is still in use.")
 
 
+class VolumeNotUsable(Invalid):
+    message = _("Volume %(volume)s not usable for the container.")
+
+
+class VolumeInUse(Invalid):
+    message = _("Volume %(volume)s is still in use.")
+
+
 class PortBindingFailed(Invalid):
     message = _("Binding failed for port %(port)s, please check neutron "
                 "logs for more information.")

zun/conf/__init__.py

@@ -15,12 +15,14 @@
 from oslo_config import cfg
 
 from zun.conf import api
+from zun.conf import cinder_client
 from zun.conf import compute
 from zun.conf import container_driver
 from zun.conf import database
 from zun.conf import docker
 from zun.conf import glance_client
 from zun.conf import image_driver
+from zun.conf import netconf
 from zun.conf import network
 from zun.conf import neutron_client
 from zun.conf import nova_client
@@ -30,6 +32,7 @@ from zun.conf import profiler
 from zun.conf import scheduler
 from zun.conf import services
 from zun.conf import ssl
+from zun.conf import volume
 from zun.conf import websocket_proxy
 from zun.conf import zun_client
 
@@ -53,3 +56,6 @@ neutron_client.register_opts(CONF)
 network.register_opts(CONF)
 websocket_proxy.register_opts(CONF)
 pci.register_opts(CONF)
+volume.register_opts(CONF)
+cinder_client.register_opts(CONF)
+netconf.register_opts(CONF)

zun/conf/cinder_client.py

@@ -0,0 +1,46 @@
+# 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.
+
+from oslo_config import cfg
+
+
+cinder_group = cfg.OptGroup(name='cinder_client',
+                            title='Options for the Cinder client')
+
+common_security_opts = [
+    cfg.StrOpt('ca_file',
+               help='Optional CA cert file to use in SSL connections.'),
+    cfg.BoolOpt('insecure',
+                default=False,
+                help="If set, then the server's certificate will not "
+                     "be verified.")]
+
+cinder_client_opts = [
+    cfg.StrOpt('endpoint_type',
+               default='publicURL',
+               help='Type of endpoint in Identity service catalog to use '
+                    'for communication with the OpenStack service.'),
+    cfg.StrOpt('api_version',
+               default='3',
+               help='Version of Cinder API to use in cinderclient.')]
+
+
+ALL_OPTS = (cinder_client_opts + common_security_opts)
+
+
+def register_opts(conf):
+    conf.register_group(cinder_group)
+    conf.register_opts(ALL_OPTS, group=cinder_group)
+
+
+def list_opts():
+    return {cinder_group: ALL_OPTS}

zun/conf/netconf.py

@@ -0,0 +1,56 @@
+# 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.
+
+from oslo_config import cfg
+from oslo_utils import netutils
+
+
+netconf_opts = [
+    cfg.StrOpt("my_ip",
+               default=netutils.get_my_ipv4(),
+               help="""
+The IP address which the host is using to connect to the management network.
+
+Possible values:
+
+* String with valid IP address. Default is IPv4 address of this host.
+
+Related options:
+
+* my_block_storage_ip
+"""),
+    cfg.StrOpt("my_block_storage_ip",
+               default="$my_ip",
+               help="""
+The IP address which is used to connect to the block storage network.
+
+Possible values:
+
+* String with valid IP address. Default is IP address of this host.
+
+Related options:
+
+* my_ip - if my_block_storage_ip is not set, then my_ip value is used.
+"""),
+]
+
+
+ALL_OPTS = (netconf_opts)
+
+
+def register_opts(conf):
+    conf.register_opts(ALL_OPTS)
+
+
+def list_opts():
+    return {"DEFAULT": ALL_OPTS}

zun/conf/volume.py

@@ -0,0 +1,49 @@
+# 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.
+
+from oslo_config import cfg
+
+
+volume_group = cfg.OptGroup(name='volume',
+                            title='Options for the container volume')
+
+volume_opts = [
+    cfg.StrOpt('driver',
+               default='cinder',
+               help='Defines which driver to use for container volume.'),
+    cfg.StrOpt('volume_dir',
+               default='$state_path/mnt',
+               help='At which the docker volume will create.'),
+    cfg.StrOpt('fstype',
+               default='ext4',
+               help='Default filesystem type for volume.'),
+    cfg.BoolOpt('use_multipath',
+                default=False,
+                help="""
+Use multipath connection of volume
+
+Volumes can be connected as multipath devices. This will provide high
+availability and fault tolerance.
+"""),
+]
+
+
+ALL_OPTS = (volume_opts)
+
+
+def register_opts(conf):
+    conf.register_group(volume_group)
+    conf.register_opts(ALL_OPTS, group=volume_group)
+
+
+def list_opts():
+    return {volume_group: ALL_OPTS}

zun/tests/unit/volume/test_cinder_api.py

@@ -0,0 +1,226 @@
+#    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 cinderclient import exceptions as cinder_exception
+from oslo_utils import timeutils
+
+import zun.conf
+from zun.tests import base
+from zun.volume import cinder_api
+
+
+CONF = zun.conf.CONF
+
+
+class FakeVolume(object):
+
+    def __init__(self, volume_id, size=1, attachments=None, multiattach=False):
+        self.id = volume_id
+        self.name = 'volume_name'
+        self.description = 'volume_description'
+        self.status = 'available'
+        self.created_at = timeutils.utcnow()
+        self.size = size
+        self.availability_zone = 'nova'
+        self.attachments = attachments or []
+        self.volume_type = 99
+        self.bootable = False
+        self.snapshot_id = 'snap_id_1'
+        self.metadata = {}
+        self.multiattach = multiattach
+
+
+class TestingException(Exception):
+    pass
+
+
+class CinderApiTestCase(base.TestCase):
+
+    @mock.patch('zun.common.clients.OpenStackClients.cinder')
+    def test_get(self, mock_cinderclient):
+        volume_id = 'volume_id1'
+        mock_volumes = mock.MagicMock()
+        mock_cinderclient.return_value = mock.MagicMock(volumes=mock_volumes)
+
+        self.api = cinder_api.CinderAPI(self.context)
+        self.api.get(volume_id)
+
+        mock_cinderclient.assert_called_once_with()
+        mock_volumes.get.assert_called_once_with(volume_id)
+
+    @mock.patch('zun.common.clients.OpenStackClients.cinder')
+    def test_reserve_volume(self, mock_cinderclient):
+        mock_volumes = mock.MagicMock()
+        mock_cinderclient.return_value = mock.MagicMock(volumes=mock_volumes)
+
+        self.api = cinder_api.CinderAPI(self.context)
+        self.api.reserve_volume('id1')
+
+        mock_cinderclient.assert_called_once_with()
+        mock_volumes.reserve.assert_called_once_with('id1')
+
+    @mock.patch('zun.common.clients.OpenStackClients.cinder')
+    def test_unreserve_volume(self, mock_cinderclient):
+        mock_volumes = mock.MagicMock()
+        mock_cinderclient.return_value = mock.MagicMock(volumes=mock_volumes)
+
+        self.api = cinder_api.CinderAPI(self.context)
+        self.api.unreserve_volume('id1')
+
+        mock_cinderclient.assert_called_once_with()
+        mock_volumes.unreserve.assert_called_once_with('id1')
+
+    @mock.patch('zun.common.clients.OpenStackClients.cinder')
+    def test_begin_detaching(self, mock_cinderclient):
+        mock_volumes = mock.MagicMock()
+        mock_cinderclient.return_value = mock.MagicMock(volumes=mock_volumes)
+
+        self.api = cinder_api.CinderAPI(self.context)
+        self.api.begin_detaching('id1')
+
+        mock_cinderclient.assert_called_once_with()
+        mock_volumes.begin_detaching.assert_called_once_with('id1')
+
+    @mock.patch('zun.common.clients.OpenStackClients.cinder')
+    def test_roll_detaching(self, mock_cinderclient):
+        mock_volumes = mock.MagicMock()
+        mock_cinderclient.return_value = mock.MagicMock(volumes=mock_volumes)
+
+        self.api = cinder_api.CinderAPI(self.context)
+        self.api.roll_detaching('id1')
+
+        mock_cinderclient.assert_called_once_with()
+        mock_volumes.roll_detaching.assert_called_once_with('id1')
+
+    @mock.patch('zun.common.clients.OpenStackClients.cinder')
+    def test_attach(self, mock_cinderclient):
+        mock_volumes = mock.MagicMock()
+        mock_cinderclient.return_value = mock.MagicMock(volumes=mock_volumes)
+
+        self.api = cinder_api.CinderAPI(self.context)
+        self.api.attach('id1', 'point', 'host')
+
+        mock_cinderclient.assert_called_once_with()
+        mock_volumes.attach.assert_called_once_with(
+            volume='id1', mountpoint='point', host_name='host',
+            instance_uuid=None)
+
+    @mock.patch('zun.common.clients.OpenStackClients.cinder')
+    def test_detach(self, mock_cinderclient):
+        attachment = {'host_name': 'fake_host',
+                      'attachment_id': 'fakeid'}
+
+        mock_volumes = mock.MagicMock()
+        mock_cinderclient.return_value = mock.MagicMock(volumes=mock_volumes)
+        mock_cinderclient.return_value.volumes.get.return_value = \
+            FakeVolume('id1', attachments=[attachment])
+
+        self.api = cinder_api.CinderAPI(self.context)
+        self.api.detach('id1')
+
+        mock_cinderclient.assert_called_with()
+        mock_volumes.detach.assert_called_once_with('id1', None)
+
+    @mock.patch('zun.common.clients.OpenStackClients.cinder')
+    def test_detach_multiattach(self, mock_cinderclient):
+        attachment = {'host_name': CONF.host,
+                      'attachment_id': 'fakeid'}
+
+        mock_volumes = mock.MagicMock()
+        mock_cinderclient.return_value = mock.MagicMock(volumes=mock_volumes)
+        mock_cinderclient.return_value.volumes.get.return_value = \
+            FakeVolume('id1', attachments=[attachment], multiattach=True)
+
+        self.api = cinder_api.CinderAPI(self.context)
+        self.api.detach('id1')
+
+        mock_cinderclient.assert_called_with()
+        mock_volumes.detach.assert_called_once_with('id1', 'fakeid')
+
+    @mock.patch('zun.common.clients.OpenStackClients.cinder')
+    def test_initialize_connection(self, mock_cinderclient):
+        connection_info = {'foo': 'bar'}
+        mock_cinderclient.return_value.volumes. \
+            initialize_connection.return_value = connection_info
+
+        volume_id = 'fake_vid'
+        connector = {'host': 'fakehost1'}
+        self.api = cinder_api.CinderAPI(self.context)
+        actual = self.api.initialize_connection(volume_id, connector)
+
+        expected = connection_info
+        self.assertEqual(expected, actual)
+
+        mock_cinderclient.return_value.volumes. \
+            initialize_connection.assert_called_once_with(volume_id, connector)
+
+    @mock.patch('zun.volume.cinder_api.LOG')
+    @mock.patch('zun.common.clients.OpenStackClients.cinder')
+    def test_initialize_connection_exception_no_code(
+            self, mock_cinderclient, mock_log):
+        mock_cinderclient.return_value.volumes. \
+            initialize_connection.side_effect = (
+                cinder_exception.ClientException(500, "500"))
+        mock_cinderclient.return_value.volumes. \
+            terminate_connection.side_effect = (TestingException)
+
+        connector = {'host': 'fakehost1'}
+        self.api = cinder_api.CinderAPI(self.context)
+        self.assertRaises(cinder_exception.ClientException,
+                          self.api.initialize_connection,
+                          'id1',
+                          connector)
+        self.assertIsNone(mock_log.error.call_args_list[1][0][1]['code'])
+
+    @mock.patch('zun.common.clients.OpenStackClients.cinder')
+    def test_initialize_connection_rollback(self, mock_cinderclient):
+        mock_cinderclient.return_value.volumes.\
+            initialize_connection.side_effect = (
+                cinder_exception.ClientException(500, "500"))
+
+        connector = {'host': 'host1'}
+        self.api = cinder_api.CinderAPI(self.context)
+        ex = self.assertRaises(cinder_exception.ClientException,
+                               self.api.initialize_connection,
+                               'id1',
+                               connector)
+        self.assertEqual(500, ex.code)
+        mock_cinderclient.return_value.volumes.\
+            terminate_connection.assert_called_once_with('id1', connector)
+
+    @mock.patch('zun.common.clients.OpenStackClients.cinder')
+    def test_initialize_connection_no_rollback(self, mock_cinderclient):
+        mock_cinderclient.return_value.volumes.\
+            initialize_connection.side_effect = TestingException
+
+        connector = {'host': 'host1'}
+        self.api = cinder_api.CinderAPI(self.context)
+        self.assertRaises(TestingException,
+                          self.api.initialize_connection,
+                          'id1',
+                          connector)
+        self.assertFalse(mock_cinderclient.return_value.volumes.
+                         terminate_connection.called)
+
+    @mock.patch('zun.common.clients.OpenStackClients.cinder')
+    def test_terminate_connection(self, mock_cinderclient):
+        mock_volumes = mock.MagicMock()
+        mock_cinderclient.return_value = mock.MagicMock(volumes=mock_volumes)
+
+        self.api = cinder_api.CinderAPI(self.context)
+        self.api.terminate_connection('id1', 'connector')
+
+        mock_cinderclient.assert_called_once_with()
+        mock_volumes.terminate_connection.assert_called_once_with('id1',
+                                                                  'connector')

zun/tests/unit/volume/test_cinder_workflow.py

@@ -0,0 +1,268 @@
+#    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 cinderclient import exceptions as cinder_exception
+from os_brick import exception as os_brick_exception
+from oslo_serialization import jsonutils
+
+import zun.conf
+from zun.tests import base
+from zun.volume import cinder_workflow
+
+
+CONF = zun.conf.CONF
+
+
+class CinderWorkflowTestCase(base.TestCase):
+
+    def setUp(self):
+        super(CinderWorkflowTestCase, self).setUp()
+        self.fake_volume_id = 'fake-volume-id-1'
+        self.fake_conn_prprts = {
+            'ip': '10.3.4.5',
+            'host': 'fakehost1'
+        }
+        self.fake_device_info = {
+            'path': '/foo'
+        }
+        self.fake_conn_info = {
+            'driver_volume_type': 'fake',
+            'data': {},
+        }
+
+    @mock.patch('zun.volume.cinder_workflow.get_volume_connector')
+    @mock.patch('zun.volume.cinder_workflow.get_volume_connector_properties')
+    @mock.patch('zun.volume.cinder_api.CinderAPI')
+    def test_attach_volume(self,
+                           mock_cinder_api_cls,
+                           mock_get_connector_prprts,
+                           mock_get_volume_connector):
+        mock_cinder_api, mock_connector = self._test_attach_volume(
+            mock_cinder_api_cls, mock_get_connector_prprts,
+            mock_get_volume_connector)
+
+        mock_cinder_api.reserve_volume.assert_called_once_with(
+            self.fake_volume_id)
+        mock_cinder_api.initialize_connection.assert_called_once_with(
+            self.fake_volume_id, self.fake_conn_prprts)
+        mock_connector.connect_volume.assert_called_once_with(
+            self.fake_conn_info['data'])
+        mock_cinder_api.attach.assert_called_once_with(
+            volume_id=self.fake_volume_id,
+            mountpoint=self.fake_device_info['path'],
+            hostname=CONF.host)
+        mock_connector.disconnect_volume.assert_not_called()
+        mock_cinder_api.terminate_connection.assert_not_called()
+        mock_cinder_api.detach.assert_not_called()
+        mock_cinder_api.unreserve_volume.assert_not_called()
+
+    @mock.patch('zun.volume.cinder_workflow.get_volume_connector')
+    @mock.patch('zun.volume.cinder_workflow.get_volume_connector_properties')
+    @mock.patch('zun.volume.cinder_api.CinderAPI')
+    def test_attach_volume_fail_reserve_volume(
+            self, mock_cinder_api_cls, mock_get_connector_prprts,
+            mock_get_volume_connector):
+        mock_cinder_api, mock_connector = self._test_attach_volume(
+            mock_cinder_api_cls, mock_get_connector_prprts,
+            mock_get_volume_connector, fail_reserve=True)
+
+        mock_cinder_api.reserve_volume.assert_called_once_with(
+            self.fake_volume_id)
+        mock_cinder_api.initialize_connection.assert_not_called()
+        mock_connector.connect_volume.assert_not_called()
+        mock_cinder_api.attach.assert_not_called()
+        mock_connector.disconnect_volume.assert_not_called()
+        mock_cinder_api.terminate_connection.assert_not_called()
+        mock_cinder_api.detach.assert_not_called()
+        mock_cinder_api.unreserve_volume.assert_called_once_with(
+            self.fake_volume_id)
+
+    @mock.patch('zun.volume.cinder_workflow.get_volume_connector')
+    @mock.patch('zun.volume.cinder_workflow.get_volume_connector_properties')
+    @mock.patch('zun.volume.cinder_api.CinderAPI')
+    def test_attach_volume_fail_initialize_connection(
+            self, mock_cinder_api_cls, mock_get_connector_prprts,
+            mock_get_volume_connector):
+        mock_cinder_api, mock_connector = self._test_attach_volume(
+            mock_cinder_api_cls, mock_get_connector_prprts,
+            mock_get_volume_connector, fail_init=True)
+
+        mock_cinder_api.reserve_volume.assert_called_once_with(
+            self.fake_volume_id)
+        mock_cinder_api.initialize_connection.assert_called_once_with(
+            self.fake_volume_id, self.fake_conn_prprts)
+        mock_connector.connect_volume.assert_not_called()
+        mock_cinder_api.attach.assert_not_called()
+        mock_connector.disconnect_volume.assert_not_called()
+        mock_cinder_api.terminate_connection.assert_not_called()
+        mock_cinder_api.detach.assert_not_called()
+        mock_cinder_api.unreserve_volume.assert_called_once_with(
+            self.fake_volume_id)
+
+    @mock.patch('zun.volume.cinder_workflow.get_volume_connector')
+    @mock.patch('zun.volume.cinder_workflow.get_volume_connector_properties')
+    @mock.patch('zun.volume.cinder_api.CinderAPI')
+    def test_attach_volume_fail_connect_volume(
+            self, mock_cinder_api_cls, mock_get_connector_prprts,
+            mock_get_volume_connector):
+        mock_cinder_api, mock_connector = self._test_attach_volume(
+            mock_cinder_api_cls, mock_get_connector_prprts,
+            mock_get_volume_connector, fail_connect=True)
+
+        mock_cinder_api.reserve_volume.assert_called_once_with(
+            self.fake_volume_id)
+        mock_cinder_api.initialize_connection.assert_called_once_with(
+            self.fake_volume_id, self.fake_conn_prprts)
+        mock_connector.connect_volume.assert_called_once_with(
+            self.fake_conn_info['data'])
+        mock_cinder_api.attach.assert_not_called()
+        mock_connector.disconnect_volume.assert_not_called()
+        mock_cinder_api.terminate_connection.assert_called_once_with(
+            self.fake_volume_id, self.fake_conn_prprts)
+        mock_cinder_api.detach.assert_not_called()
+        mock_cinder_api.unreserve_volume.assert_called_once_with(
+            self.fake_volume_id)
+
+    @mock.patch('zun.volume.cinder_workflow.get_volume_connector')
+    @mock.patch('zun.volume.cinder_workflow.get_volume_connector_properties')
+    @mock.patch('zun.volume.cinder_api.CinderAPI')
+    def test_attach_volume_fail_attach(self,
+                                       mock_cinder_api_cls,
+                                       mock_get_connector_prprts,
+                                       mock_get_volume_connector):
+        mock_cinder_api, mock_connector = self._test_attach_volume(
+            mock_cinder_api_cls, mock_get_connector_prprts,
+            mock_get_volume_connector, fail_attach=True)
+
+        mock_cinder_api.reserve_volume.assert_called_once_with(
+            self.fake_volume_id)
+        mock_cinder_api.initialize_connection.assert_called_once_with(
+            self.fake_volume_id, self.fake_conn_prprts)
+        mock_connector.connect_volume.assert_called_once_with(
+            self.fake_conn_info['data'])
+        mock_cinder_api.attach.assert_called_once_with(
+            volume_id=self.fake_volume_id,
+            mountpoint=self.fake_device_info['path'],
+            hostname=CONF.host)
+        mock_connector.disconnect_volume.assert_called_once_with(
+            self.fake_conn_info['data'], None)
+        mock_cinder_api.terminate_connection.assert_called_once_with(
+            self.fake_volume_id, self.fake_conn_prprts)
+        mock_cinder_api.detach.assert_called_once_with(
+            self.fake_volume_id)
+        mock_cinder_api.unreserve_volume.assert_called_once_with(
+            self.fake_volume_id)
+
+    def _test_attach_volume(self,
+                            mock_cinder_api_cls,
+                            mock_get_connector_prprts,
+                            mock_get_volume_connector,
+                            fail_reserve=False, fail_init=False,
+                            fail_connect=False, fail_attach=False):
+        volume = mock.MagicMock()
+        volume.volume_id = self.fake_volume_id
+        mock_cinder_api = mock.MagicMock()
+        mock_cinder_api_cls.return_value = mock_cinder_api
+        mock_connector = mock.MagicMock()
+        mock_get_connector_prprts.return_value = self.fake_conn_prprts
+        mock_get_volume_connector.return_value = mock_connector
+        mock_cinder_api.initialize_connection.return_value = \
+            self.fake_conn_info
+        mock_connector.connect_volume.return_value = self.fake_device_info
+        cinder = cinder_workflow.CinderWorkflow(self.context)
+
+        if fail_reserve:
+            mock_cinder_api.reserve_volume.side_effect = \
+                cinder_exception.ClientException(400)
+            self.assertRaises(cinder_exception.ClientException,
+                              cinder.attach_volume, volume)
+        elif fail_init:
+            mock_cinder_api.initialize_connection.side_effect = \
+                cinder_exception.ClientException(400)
+            self.assertRaises(cinder_exception.ClientException,
+                              cinder.attach_volume, volume)
+        elif fail_connect:
+            mock_connector.connect_volume.side_effect = \
+                os_brick_exception.BrickException()
+            self.assertRaises(os_brick_exception.BrickException,
+                              cinder.attach_volume, volume)
+        elif fail_attach:
+            mock_cinder_api.attach.side_effect = \
+                cinder_exception.ClientException(400)
+            self.assertRaises(cinder_exception.ClientException,
+                              cinder.attach_volume, volume)
+        else:
+            device_path = cinder.attach_volume(volume)
+            self.assertEqual('/foo', device_path)
+
+        return mock_cinder_api, mock_connector
+
+    @mock.patch('zun.volume.cinder_workflow.get_volume_connector')
+    @mock.patch('zun.volume.cinder_workflow.get_volume_connector_properties')
+    @mock.patch('zun.volume.cinder_api.CinderAPI')
+    def test_detach_volume(self,
+                           mock_cinder_api_cls,
+                           mock_get_connector_prprts,
+                           mock_get_volume_connector):
+        volume = mock.MagicMock()
+        volume.volume_id = self.fake_volume_id
+        volume.connection_info = jsonutils.dumps(self.fake_conn_info)
+        mock_cinder_api = mock.MagicMock()
+        mock_cinder_api_cls.return_value = mock_cinder_api
+        mock_connector = mock.MagicMock()
+        mock_get_connector_prprts.return_value = self.fake_conn_prprts
+        mock_get_volume_connector.return_value = mock_connector
+
+        cinder = cinder_workflow.CinderWorkflow(self.context)
+        cinder.detach_volume(volume)
+
+        mock_cinder_api.begin_detaching.assert_called_once_with(
+            self.fake_volume_id)
+        mock_connector.disconnect_volume.assert_called_once_with(
+            self.fake_conn_info['data'], None)
+        mock_cinder_api.terminate_connection.assert_called_once_with(
+            self.fake_volume_id, self.fake_conn_prprts)
+        mock_cinder_api.detach.assert_called_once_with(
+            self.fake_volume_id)
+        mock_cinder_api.roll_detaching.assert_not_called()
+
+    @mock.patch('zun.volume.cinder_workflow.get_volume_connector')
+    @mock.patch('zun.volume.cinder_workflow.get_volume_connector_properties')
+    @mock.patch('zun.volume.cinder_api.CinderAPI')
+    def test_detach_volume_fail_disconnect(
+            self, mock_cinder_api_cls, mock_get_connector_prprts,
+            mock_get_volume_connector):
+        volume = mock.MagicMock()
+        volume.volume_id = self.fake_volume_id
+        volume.connection_info = jsonutils.dumps(self.fake_conn_info)
+        mock_cinder_api = mock.MagicMock()
+        mock_cinder_api_cls.return_value = mock_cinder_api
+        mock_connector = mock.MagicMock()
+        mock_get_connector_prprts.return_value = self.fake_conn_prprts
+        mock_get_volume_connector.return_value = mock_connector
+        mock_connector.disconnect_volume.side_effect = \
+            os_brick_exception.BrickException()
+
+        cinder = cinder_workflow.CinderWorkflow(self.context)
+        self.assertRaises(os_brick_exception.BrickException,
+                          cinder.detach_volume, volume)
+
+        mock_cinder_api.begin_detaching.assert_called_once_with(
+            self.fake_volume_id)
+        mock_connector.disconnect_volume.assert_called_once_with(
+            self.fake_conn_info['data'], None)
+        mock_cinder_api.terminate_connection.assert_not_called()
+        mock_cinder_api.detach.assert_not_called()
+        mock_cinder_api.roll_detaching.assert_called_once_with(
+            self.fake_volume_id)

zun/volume/cinder_api.py

@@ -0,0 +1,138 @@
+#    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 six
+
+from cinderclient import exceptions as cinder_exception
+from oslo_log import log as logging
+from oslo_utils import excutils
+from oslo_utils import uuidutils
+
+from zun.common import clients
+from zun.common import exception
+from zun.common.i18n import _
+import zun.conf
+
+
+LOG = logging.getLogger(__name__)
+
+CONF = zun.conf.CONF
+
+
+class CinderAPI(object):
+
+    def __init__(self, context):
+        self.context = context
+        self.cinder = clients.OpenStackClients(self.context).cinder()
+
+    def __getattr__(self, key):
+        return getattr(self.cinder, key)
+
+    def get(self, volume_id):
+        return self.cinder.volumes.get(volume_id)
+
+    def search_volume(self, volume):
+        if uuidutils.is_uuid_like(volume):
+            volume = self.cinder.volumes.get(volume)
+            volumes = [volume]
+        else:
+            volumes = self.cinder.volumes.list(search_opts={'name': volume})
+
+        if len(volumes) == 0:
+            raise exception.VolumeNotFound(volume=volume)
+        elif len(volumes) > 1:
+            raise exception.Conflict(_(
+                'Multiple cinder volumes exist with same name. '
+                'Please use the uuid instead.'))
+
+        volume = volumes[0]
+        return volume
+
+    def ensure_volume_usable(self, volume):
+        # Make sure the container has access to the volume.
+        if hasattr(volume, 'os-vol-tenant-attr:tenant_id'):
+            project_id = self.context.project_id
+            if getattr(volume, 'os-vol-tenant-attr:tenant_id') != project_id:
+                raise exception.VolumeNotUsable(volume=volume.id)
+
+        if volume.attachments and not volume.multiattach:
+            raise exception.VolumeInUse(volume=volume.id)
+
+    def reserve_volume(self, volume_id):
+        return self.cinder.volumes.reserve(volume_id)
+
+    def unreserve_volume(self, volume_id):
+        return self.cinder.volumes.unreserve(volume_id)
+
+    def initialize_connection(self, volume_id, connector):
+        try:
+            connection_info = self.cinder.volumes.initialize_connection(
+                volume_id, connector)
+            return connection_info
+        except cinder_exception.ClientException as ex:
+            with excutils.save_and_reraise_exception():
+                LOG.error('Initialize connection failed for volume '
+                          '%(vol)s on host %(host)s. Error: %(msg)s '
+                          'Code: %(code)s. Attempting to terminate '
+                          'connection.',
+                          {'vol': volume_id,
+                           'host': connector.get('host'),
+                           'msg': six.text_type(ex),
+                           'code': ex.code})
+                try:
+                    self.terminate_connection(volume_id, connector)
+                except Exception as exc:
+                    LOG.error('Connection between volume %(vol)s and host '
+                              '%(host)s might have succeeded, but attempt '
+                              'to terminate connection has failed. '
+                              'Validate the connection and determine if '
+                              'manual cleanup is needed. Error: %(msg)s '
+                              'Code: %(code)s.',
+                              {'vol': volume_id,
+                               'host': connector.get('host'),
+                               'msg': six.text_type(exc),
+                               'code': (exc.code
+                                        if hasattr(exc, 'code') else None)})
+
+    def terminate_connection(self, volume_id, connector):
+        return self.cinder.volumes.terminate_connection(volume_id, connector)
+
+    def attach(self, volume_id, mountpoint, hostname):
+        return self.cinder.volumes.attach(volume=volume_id,
+                                          instance_uuid=None,
+                                          mountpoint=mountpoint,
+                                          host_name=hostname)
+
+    def detach(self, volume_id):
+        attachment_id = None
+        volume = self.get(volume_id)
+        attachments = volume.attachments or {}
+        for am in attachments:
+            if am['host_name'].lower() == CONF.host.lower():
+                attachment_id = am['attachment_id']
+                break
+
+        if attachment_id is None and volume.multiattach:
+            LOG.warning("attachment_id couldn't be retrieved for "
+                        "volume %(volume_id)s. The volume has the "
+                        "'multiattach' flag enabled, without the "
+                        "attachment_id Cinder most probably "
+                        "cannot perform the detach.",
+                        {'volume_id': volume_id})
+
+        return self.cinder.volumes.detach(volume_id, attachment_id)
+
+    def begin_detaching(self, volume_id):
+        self.cinder.volumes.begin_detaching(volume_id)
+
+    def roll_detaching(self, volume_id):
+        self.cinder.volumes.roll_detaching(volume_id)

zun/volume/cinder_workflow.py

@@ -0,0 +1,171 @@
+#    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.
+
+from cinderclient import exceptions as cinder_exception
+from os_brick import exception as os_brick_exception
+from os_brick.initiator import connector as brick_connector
+from oslo_log import log as logging
+from oslo_serialization import jsonutils
+from oslo_utils import excutils
+
+from zun.common import exception
+from zun.common.i18n import _
+from zun.common import utils
+import zun.conf
+from zun.volume import cinder_api as cinder
+
+
+LOG = logging.getLogger(__name__)
+
+CONF = zun.conf.CONF
+
+
+def get_volume_connector_properties():
+    """Wrapper to automatically set root_helper in brick calls.
+
+    :param multipath: A boolean indicating whether the connector can
+                      support multipath.
+    :param enforce_multipath: If True, it raises exception when multipath=True
+                              is specified but multipathd is not running.
+                              If False, it falls back to multipath=False
+                              when multipathd is not running.
+    """
+
+    root_helper = utils.get_root_helper()
+    return brick_connector.get_connector_properties(
+        root_helper,
+        CONF.my_block_storage_ip,
+        CONF.volume.use_multipath,
+        enforce_multipath=True,
+        host=CONF.host)
+
+
+def get_volume_connector(protocol, driver=None,
+                         use_multipath=False,
+                         device_scan_attempts=3,
+                         *args, **kwargs):
+    """Wrapper to get a brick connector object.
+
+    This automatically populates the required protocol as well
+    as the root_helper needed to execute commands.
+    """
+
+    root_helper = utils.get_root_helper()
+    if protocol.upper() == "RBD":
+        kwargs['do_local_attach'] = True
+    return brick_connector.InitiatorConnector.factory(
+        protocol, root_helper,
+        driver=driver,
+        use_multipath=use_multipath,
+        device_scan_attempts=device_scan_attempts,
+        *args, **kwargs)
+
+
+class CinderWorkflow(object):
+
+    def __init__(self, context):
+        self.context = context
+
+    def attach_volume(self, volume):
+        cinder_api = cinder.CinderAPI(self.context)
+        try:
+            return self._do_attach_volume(cinder_api, volume)
+        except Exception:
+            with excutils.save_and_reraise_exception():
+                LOG.exception("Failed to attach volume %(volume_id)s",
+                              {'volume_id': volume.volume_id})
+                cinder_api.unreserve_volume(volume.volume_id)
+
+    def _do_attach_volume(self, cinder_api, volume):
+        volume_id = volume.volume_id
+
+        cinder_api.reserve_volume(volume_id)
+        conn_info = cinder_api.initialize_connection(
+            volume_id,
+            get_volume_connector_properties())
+        LOG.info("Get connection information %s", conn_info)
+
+        try:
+            device_info = self._connect_volume(conn_info)
+            LOG.info("Get device_info after connect to "
+                     "volume %s", device_info)
+        except Exception:
+            with excutils.save_and_reraise_exception():
+                cinder_api.terminate_connection(
+                    volume_id, get_volume_connector_properties())
+
+        conn_info['data']['device_path'] = device_info['path']
+        mountpoint = device_info['path']
+        try:
+            volume.connection_info = jsonutils.dumps(conn_info)
+        except TypeError:
+            pass
+        # NOTE(hongbin): save connection_info in the database
+        # before calling cinder_api.attach because the volume status
+        # will go to 'in-use' then caller immediately try to detach
+        # the volume and connection_info is required for detach.
+        volume.save()
+
+        try:
+            cinder_api.attach(volume_id=volume_id,
+                              mountpoint=mountpoint,
+                              hostname=CONF.host)
+            LOG.info("Attach volume to this server successfully")
+        except Exception:
+            with excutils.save_and_reraise_exception():
+                try:
+                    self._disconnect_volume(conn_info)
+                except os_brick_exception.VolumeDeviceNotFound as exc:
+                    LOG.warning('Ignoring VolumeDeviceNotFound: %s', exc)
+
+                cinder_api.terminate_connection(
+                    volume_id, get_volume_connector_properties())
+
+                # Cinder-volume might have completed volume attach. So
+                # we should detach the volume. If the attach did not
+                # happen, the detach request will be ignored.
+                cinder_api.detach(volume_id)
+
+        return device_info['path']
+
+    def _connect_volume(self, conn_info):
+        protocol = conn_info['driver_volume_type']
+        connector = get_volume_connector(protocol)
+        device_info = connector.connect_volume(conn_info['data'])
+        return device_info
+
+    def _disconnect_volume(self, conn_info):
+        protocol = conn_info['driver_volume_type']
+        connector = get_volume_connector(protocol)
+        connector.disconnect_volume(conn_info['data'], None)
+
+    def detach_volume(self, volume):
+        volume_id = volume.volume_id
+        cinder_api = cinder.CinderAPI(self.context)
+
+        try:
+            cinder_api.begin_detaching(volume_id)
+        except cinder_exception.BadRequest as e:
+            raise exception.Invalid(_("Invalid volume: %s") % str(e))
+
+        conn_info = jsonutils.loads(volume.connection_info)
+        try:
+            self._disconnect_volume(conn_info)
+        except Exception:
+            with excutils.save_and_reraise_exception():
+                LOG.exception('Failed to disconnect volume %(volume_id)s',
+                              {'volume_id': volume_id})
+                cinder_api.roll_detaching(volume_id)
+
+        cinder_api.terminate_connection(
+            volume_id, get_volume_connector_properties())
+        cinder_api.detach(volume_id)