Check available capacity before creating resources

For some resources, due to their inheritance of the 'host' attribute
from the source where they are asked to create from, the 'CREATE'
request doesn't go through Cinder scheduler, thus missing the chance
to check if the storage backend still have capacity. Such resources
are snapshots, volume clones and potentially volume replication
clones. In some cases, current implementation is fine because those
new resource would end up being in 'error' state if backend is out
of capacity, but in the case where thin-provisioning is desired and
the backend has max over-subscription ratio set, this might cause
backend to serve more than it should if the backend doesn't enforce
simliar checks.

The solution in this change is simply passing all create requests to
scheduler even for those resource has known 'host'.

Alternative solution, which tries to add some scheduling logic into
volume manager is here: https://review.openstack.org/#/c/437677
The fact that this change almost duplicates scheduler code in a very
unclean way makes it ugly, hard to maintain hence less preferrable.

Change-Id: I3454255c8dae481c85f79896ce53fabdc2a50b4d
Closes-bug: #1537166

cinder/scheduler/filter_scheduler.py

@@ -126,7 +126,7 @@ class FilterScheduler(driver.Scheduler):
         reason_param = {'resource': 'volume',
                         'id': '??id missing??',
                         'backend': backend}
-        for resource in ['volume', 'group']:
+        for resource in ['volume', 'group', 'snapshot']:
             resource_id = request_spec.get('%s_id' % resource, None)
             if resource_id:
                 reason_param.update({'resource': resource, 'id': resource_id})

cinder/scheduler/manager.py

@@ -192,6 +192,29 @@ class SchedulerManager(manager.CleanableManager, manager.Manager):
         with flow_utils.DynamicLogListener(flow_engine, logger=LOG):
             flow_engine.run()
 
+    def create_snapshot(self, ctxt, volume, snapshot, backend,
+                        request_spec=None, filter_properties=None):
+        """Create snapshot for a volume.
+
+        The main purpose of this method is to check if target
+        backend (of volume and snapshot) has sufficient capacity
+        to host to-be-created snapshot.
+        """
+        self._wait_for_scheduler()
+
+        try:
+            tgt_backend = self.driver.backend_passes_filters(
+                ctxt, backend, request_spec, filter_properties)
+            tgt_backend.consume_from_volume(
+                {'size': request_spec['volume_properties']['size']})
+        except exception.NoValidBackend as ex:
+            self._set_snapshot_state_and_notify('create_snapshot',
+                                                snapshot, 'error',
+                                                ctxt, ex, request_spec)
+        else:
+            volume_rpcapi.VolumeAPI().create_snapshot(ctxt, volume,
+                                                      snapshot)
+
     def _do_cleanup(self, ctxt, vo_resource):
         # We can only receive cleanup requests for volumes, but we check anyway
         # We need to cleanup the volume status for cases where the scheduler
@@ -407,6 +430,28 @@ class SchedulerManager(manager.CleanableManager, manager.Manager):
                                             'scheduler.' + method,
                                             payload)
 
+    def _set_snapshot_state_and_notify(self, method, snapshot, state,
+                                       context, ex, request_spec,
+                                       msg=None):
+        if not msg:
+            msg = ("Failed to schedule_%(method)s: %(ex)s" %
+                   {'method': method, 'ex': six.text_type(ex)})
+        LOG.error(msg)
+
+        model_update = dict(status=state)
+        snapshot.update(model_update)
+        snapshot.save()
+
+        payload = dict(request_spec=request_spec,
+                       snapshot_id=snapshot.id,
+                       state=state,
+                       method=method,
+                       reason=ex)
+
+        rpc.get_notifier("scheduler").error(context,
+                                            'scheduler.' + method,
+                                            payload)
+
     @property
     def upgrading_cloud(self):
         min_version_str = self.sch_api.determine_rpc_version_cap()

cinder/scheduler/rpcapi.py

@@ -69,9 +69,10 @@ class SchedulerAPI(rpc.RPCAPI):
         3.6 - Removed create_consistencygroup method
         3.7 - Adds set_log_levels and get_log_levels
         3.8 - Addds ``valid_host_capacity`` method
+        3.9 - Adds create_snapshot method
     """
 
-    RPC_API_VERSION = '3.8'
+    RPC_API_VERSION = '3.9'
     RPC_DEFAULT_VERSION = '3.0'
     TOPIC = constants.SCHEDULER_TOPIC
     BINARY = 'cinder-scheduler'
@@ -109,6 +110,17 @@ class SchedulerAPI(rpc.RPCAPI):
         cctxt = self._get_cctxt()
         return cctxt.call(ctxt, 'validate_host_capacity', **msg_args)
 
+    @rpc.assert_min_rpc_version('3.9')
+    def create_snapshot(self, ctxt, volume, snapshot, backend,
+                        request_spec=None, filter_properties=None):
+        cctxt = self._get_cctxt()
+        msg_args = {'request_spec': request_spec,
+                    'filter_properties': filter_properties,
+                    'volume': volume,
+                    'snapshot': snapshot,
+                    'backend': backend}
+        return cctxt.cast(ctxt, 'create_snapshot', **msg_args)
+
     def migrate_volume(self, ctxt, volume, backend, force_copy=False,
                        request_spec=None, filter_properties=None):
         request_spec_p = jsonutils.to_primitive(request_spec)

cinder/tests/unit/api/v2/test_snapshot_metadata.py

@@ -28,6 +28,7 @@ from cinder import context
 import cinder.db
 from cinder import exception
 from cinder.objects import fields
+from cinder.scheduler import rpcapi as scheduler_rpcapi
 from cinder import test
 from cinder.tests.unit.api import fakes
 from cinder.tests.unit import fake_constants as fake
@@ -113,6 +114,7 @@ class SnapshotMetaDataTest(test.TestCase):
         super(SnapshotMetaDataTest, self).setUp()
         self.volume_api = cinder.volume.api.API()
         self.mock_object(volume.api.API, 'get', fake_get)
+        self.mock_object(scheduler_rpcapi.SchedulerAPI, 'create_snapshot')
         self.mock_object(cinder.db, 'snapshot_get', return_snapshot)
         self.mock_object(self.volume_api, 'update_snapshot_metadata')
 

cinder/tests/unit/api/v2/test_snapshots.py

@@ -29,6 +29,7 @@ from cinder import db
 from cinder import exception
 from cinder import objects
 from cinder.objects import fields
+from cinder.scheduler import rpcapi as scheduler_rpcapi
 from cinder import test
 from cinder.tests.unit.api import fakes
 from cinder.tests.unit.api.v2 import fakes as v2_fakes
@@ -84,6 +85,7 @@ def fake_snapshot_get_all(self, context, search_opts=None):
 class SnapshotApiTest(test.TestCase):
     def setUp(self):
         super(SnapshotApiTest, self).setUp()
+        self.mock_object(scheduler_rpcapi.SchedulerAPI, 'create_snapshot')
         self.controller = snapshots.SnapshotsController()
         self.ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
 

cinder/tests/unit/api/v3/test_snapshots.py

@@ -22,6 +22,7 @@ from cinder.api.v3 import snapshots
 from cinder import context
 from cinder import exception
 from cinder.objects import fields
+from cinder.scheduler import rpcapi as scheduler_rpcapi
 from cinder import test
 from cinder.tests.unit.api import fakes
 from cinder.tests.unit import fake_constants as fake
@@ -65,6 +66,7 @@ class SnapshotApiTest(test.TestCase):
     def setUp(self):
         super(SnapshotApiTest, self).setUp()
         self.mock_object(volume.api.API, 'get', fake_get)
+        self.mock_object(scheduler_rpcapi.SchedulerAPI, 'create_snapshot')
         self.controller = snapshots.SnapshotsController()
         self.ctx = context.RequestContext(fake.USER_ID, fake.PROJECT_ID, True)
 

cinder/tests/unit/scheduler/test_rpcapi.py

@@ -27,6 +27,7 @@ from cinder import objects
 from cinder.scheduler import rpcapi as scheduler_rpcapi
 from cinder import test
 from cinder.tests.unit import fake_constants
+from cinder.tests.unit import fake_snapshot
 from cinder.tests.unit import fake_volume
 
 
@@ -40,6 +41,8 @@ class SchedulerRPCAPITestCase(test.RPCAPITestCase):
         self.fake_volume = fake_volume.fake_volume_obj(
             self.context, expected_attrs=['metadata', 'admin_metadata',
                                           'glance_metadata'])
+        self.fake_snapshot = fake_snapshot.fake_snapshot_obj(
+            self.context)
         self.fake_rs_obj = objects.RequestSpec.from_primitives({})
         self.fake_rs_dict = {'volume_id': self.volume_id}
         self.fake_fp_dict = {'availability_zone': 'fake_az'}
@@ -71,6 +74,30 @@ class SchedulerRPCAPITestCase(test.RPCAPITestCase):
                            filter_properties=self.fake_fp_dict)
         create_worker_mock.assert_called_once()
 
+    @mock.patch('oslo_messaging.RPCClient.can_send_version',
+                return_value=True)
+    def test_create_snapshot(self, can_send_version_mock):
+        self._test_rpc_api('create_snapshot',
+                           rpc_method='cast',
+                           volume='fake_volume',
+                           snapshot='fake_snapshot',
+                           backend='fake_backend',
+                           request_spec={'snapshot_id': self.fake_snapshot.id},
+                           filter_properties=None)
+
+    @mock.patch('oslo_messaging.RPCClient.can_send_version',
+                return_value=False)
+    def test_create_snapshot_capped(self, can_send_version_mock):
+        self.assertRaises(exception.ServiceTooOld,
+                          self._test_rpc_api,
+                          'create_snapshot',
+                          rpc_method='cast',
+                          volume=self.fake_volume,
+                          snapshot=self.fake_snapshot,
+                          backend='fake_backend',
+                          request_spec=self.fake_rs_obj,
+                          version='3.5')
+
     @mock.patch('oslo_messaging.RPCClient.can_send_version', return_value=True)
     def test_notify_service_capabilities_backend(self, can_send_version_mock):
         """Test sending new backend by RPC instead of old host parameter."""

cinder/tests/unit/test_quota.py

@@ -34,6 +34,7 @@ from cinder import objects
 from cinder.objects import fields
 from cinder import quota
 from cinder import quota_utils
+from cinder.scheduler import rpcapi as scheduler_rpcapi
 from cinder import test
 from cinder.tests.unit import fake_constants as fake
 import cinder.tests.unit.image.fake
@@ -265,6 +266,7 @@ class QuotaIntegrationTestCase(test.TestCase):
             vol_ref.destroy()
 
     def test_no_snapshot_gb_quota_flag(self):
+        self.mock_object(scheduler_rpcapi.SchedulerAPI, 'create_snapshot')
         self.flags(quota_volumes=2,
                    quota_snapshots=2,
                    quota_gigabytes=20,

cinder/volume/api.py

@@ -809,8 +809,15 @@ class API(base.Base):
             context, volume, name,
             description, force, metadata, cgsnapshot_id,
             True, group_snapshot_id)
-        self.volume_rpcapi.create_snapshot(context, volume, snapshot)
-
+        # NOTE(tommylikehu): We only wrap the 'size' attribute here
+        # because only the volume's host is passed and only capacity is
+        # validated in the scheduler now.
+        kwargs = {'snapshot_id': snapshot.id,
+                  'volume_properties': objects.VolumeProperties(
+                      size=volume.size)}
+        self.scheduler_rpcapi.create_snapshot(context, volume, snapshot,
+                                              volume.service_topic_queue,
+                                              objects.RequestSpec(**kwargs))
         return snapshot
 
     def create_snapshot_in_db(self, context,