Add amphora delete API

This patch adds an amphora delete API. It can be used to delete
extra "spare" amphora after the feature has been disabled.

A followup patch will be required for the amphorav2 path as the
amphorav2 failover patch, which is required for the amphora delete
flow, has not yet merged.

Story: 2008014
Task: 40666

Change-Id: I32b6561c78c153a4b7e73b1a4b83e045fbe97fb6

api-ref/source/v2/amphora.inc

@@ -302,3 +302,46 @@ Response
 --------
 
 There is no body content for the response of a successful PUT request.
+
+Remove an Amphora
+=================
+
+.. rest_method:: DELETE /v2/octavia/amphorae/{amphora_id}
+
+Removes an amphora and its associated configuration.
+
+The API immediately purges any and all configuration data, depending on the
+configuration settings. You cannot recover it.
+
+**New in version 2.20**
+
+.. rest_status_code:: success ../http-status.yaml
+
+   - 204
+
+.. rest_status_code:: error ../http-status.yaml
+
+   - 400
+   - 401
+   - 403
+   - 404
+   - 409
+   - 500
+
+Request
+-------
+
+.. rest_parameters:: ../parameters.yaml
+
+   - amphora_id: path-amphora-id
+
+Curl Example
+------------
+
+.. literalinclude:: examples/amphora-delete-curl
+   :language: bash
+
+Response
+--------
+
+There is no body content for the response of a successful DELETE request.

api-ref/source/v2/examples/amphora-delete-curl

@@ -0,0 +1 @@
+curl -X DELETE -H "X-Auth-Token: <token>" http://198.51.100.10:9876/v2/octavia/amphorae/1a032adb-d6ac-4dbb-a04a-c1126bc547c7

octavia/api/root_controller.py

@@ -125,6 +125,9 @@ class RootController(object):
         self._add_a_version(versions, 'v2.19', 'v2', 'SUPPORTED',
                             '2020-05-12T00:00:00Z', host_url)
         # ALPN protocols
-        self._add_a_version(versions, 'v2.20', 'v2', 'CURRENT',
+        self._add_a_version(versions, 'v2.20', 'v2', 'SUPPORTED',
                             '2020-08-02T00:00:00Z', host_url)
+        # Amphora delete
+        self._add_a_version(versions, 'v2.21', 'v2', 'CURRENT',
+                            '2020-09-03T00:00:00Z', host_url)
         return {'versions': versions}

octavia/api/v2/controllers/amphora.py

@@ -19,6 +19,7 @@ import oslo_messaging as messaging
 from oslo_utils import excutils
 from pecan import expose as pecan_expose
 from pecan import request as pecan_request
+from sqlalchemy.orm import exc as sa_exception
 from wsme import types as wtypes
 from wsmeext import pecan as wsme_pecan
 
@@ -27,6 +28,7 @@ from octavia.api.v2.types import amphora as amp_types
 from octavia.common import constants
 from octavia.common import exceptions
 from octavia.common import rpc
+from octavia.db import api as db_api
 
 CONF = cfg.CONF
 LOG = logging.getLogger(__name__)
@@ -37,6 +39,11 @@ class AmphoraController(base.BaseController):
 
     def __init__(self):
         super().__init__()
+        topic = cfg.CONF.oslo_messaging.topic
+        self.target = messaging.Target(
+            namespace=constants.RPC_NAMESPACE_CONTROLLER_AGENT,
+            topic=topic, version="1.0", fanout=False)
+        self.client = rpc.get_client(self.target)
 
     @wsme_pecan.wsexpose(amp_types.AmphoraRootResponse, wtypes.text,
                          [wtypes.text], ignore_extra_args=True)
@@ -57,7 +64,7 @@ class AmphoraController(base.BaseController):
     @wsme_pecan.wsexpose(amp_types.AmphoraeRootResponse, [wtypes.text],
                          ignore_extra_args=True)
     def get_all(self, fields=None):
-        """Gets all health monitors."""
+        """Gets all amphorae."""
         pcontext = pecan_request.context
         context = pcontext.get('octavia_context')
 
@@ -74,6 +81,25 @@ class AmphoraController(base.BaseController):
         return amp_types.AmphoraeRootResponse(
             amphorae=result, amphorae_links=links)
 
+    @wsme_pecan.wsexpose(None, wtypes.text, status_code=204)
+    def delete(self, id):
+        """Deletes an amphora."""
+        context = pecan_request.context.get('octavia_context')
+
+        self._auth_validate_action(context, context.project_id,
+                                   constants.RBAC_DELETE)
+
+        with db_api.get_lock_session() as lock_session:
+            try:
+                self.repositories.amphora.test_and_set_status_for_delete(
+                    lock_session, id)
+            except sa_exception.NoResultFound as e:
+                raise exceptions.NotFound(resource='Amphora', id=id) from e
+
+            LOG.info("Sending delete amphora %s to the queue.", id)
+            payload = {constants.AMPHORA_ID: id}
+            self.client.cast({}, 'delete_amphora', **payload)
+
     @pecan_expose()
     def _lookup(self, amphora_id, *remainder):
         """Overridden pecan _lookup method for custom routing.

octavia/controller/queue/v1/endpoints.py

@@ -154,3 +154,7 @@ class Endpoints(object):
         LOG.info('Updating amphora \'%s\' agent configuration...',
                  amphora_id)
         self.worker.update_amphora_agent_config(amphora_id)
+
+    def delete_amphora(self, context, amphora_id):
+        LOG.info('Deleting amphora \'%s\'...', amphora_id)
+        self.worker.delete_amphora(amphora_id)

octavia/controller/worker/v1/controller_worker.py

@@ -112,6 +112,26 @@ class ControllerWorker(base_taskflow.BaseTaskFlowEngine):
         except Exception as e:
             LOG.error('Failed to create an amphora due to: {}'.format(str(e)))
 
+    def delete_amphora(self, amphora_id):
+        """Deletes an existing Amphora.
+
+        :param amphora_id: ID of the amphora to delete
+        :returns: None
+        :raises AmphoraNotFound: The referenced Amphora was not found
+        """
+        try:
+            amphora = self._amphora_repo.get(db_apis.get_session(),
+                                             id=amphora_id)
+            delete_amp_tf = self.taskflow_load(
+                self._amphora_flows.get_delete_amphora_flow(amphora))
+            with tf_logging.DynamicLoggingListener(delete_amp_tf, log=LOG):
+                delete_amp_tf.run()
+        except Exception as e:
+            LOG.error('Failed to delete a amphora {0} due to: {1}'.format(
+                amphora_id, str(e)))
+            return
+        LOG.info('Finished deleting amphora %s.', amphora_id)
+
     @tenacity.retry(
         retry=tenacity.retry_if_exception_type(db_exceptions.NoResultFound),
         wait=tenacity.wait_incrementing(

octavia/db/repositories.py

@@ -1492,6 +1492,27 @@ class AmphoraRepository(BaseRepository):
 
         return lb
 
+    def test_and_set_status_for_delete(self, lock_session, id):
+        """Tests and sets an amphora status.
+
+        Puts a lock on the amphora table to check the status of the
+        amphora. The status must be either AMPHORA_READY or ERROR to
+        successfuly update the amphora status.
+
+        :param lock_session: A Sql Alchemy database session.
+        :param id: id of Load Balancer
+        :raises ImmutableObject: The amphora is not in a state that can be
+                                 deleted.
+        :raises NoResultFound: The amphora was not found or already deleted.
+        :returns: None
+        """
+        amp = lock_session.query(self.model_class).with_for_update().filter_by(
+            id=id).filter(self.model_class.status != consts.DELETED).one()
+        if amp.status not in [consts.AMPHORA_READY, consts.ERROR]:
+            raise exceptions.ImmutableObject(resource=consts.AMPHORA, id=id)
+        amp.status = consts.PENDING_DELETE
+        lock_session.flush()
+
 
 class AmphoraBuildReqRepository(BaseRepository):
     model_class = models.AmphoraBuildRequest

octavia/policies/amphora.py

@@ -30,6 +30,13 @@ rules = [
         "Show Amphora details",
         [{'method': 'GET', 'path': '/v2/octavia/amphorae/{amphora_id}'}]
     ),
+    policy.DocumentedRuleDefault(
+        '{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_AMPHORA,
+                                    action=constants.RBAC_DELETE),
+        constants.RULE_API_ADMIN,
+        "Delete an Amphora",
+        [{'method': 'DELETE', 'path': '/v2/octavia/amphorae/{amphora_id}'}]
+    ),
     policy.DocumentedRuleDefault(
         '{rbac_obj}{action}'.format(rbac_obj=constants.RBAC_AMPHORA,
                                     action=constants.RBAC_PUT_CONFIG),

octavia/tests/functional/api/test_root_controller.py

@@ -45,7 +45,7 @@ class TestRootController(base_db_test.OctaviaDBTestBase):
     def test_api_versions(self):
         versions = self._get_versions_with_config()
         version_ids = tuple(v.get('id') for v in versions)
-        self.assertEqual(21, len(version_ids))
+        self.assertEqual(22, len(version_ids))
         self.assertIn('v2.0', version_ids)
         self.assertIn('v2.1', version_ids)
         self.assertIn('v2.2', version_ids)
@@ -67,6 +67,7 @@ class TestRootController(base_db_test.OctaviaDBTestBase):
         self.assertIn('v2.18', version_ids)
         self.assertIn('v2.19', version_ids)
         self.assertIn('v2.20', version_ids)
+        self.assertIn('v2.21', version_ids)
 
         # Each version should have a 'self' 'href' to the API version URL
         # [{u'rel': u'self', u'href': u'http://localhost/v2'}]

octavia/tests/functional/api/v2/test_amphora.py

@@ -127,6 +127,102 @@ class TestAmphora(base.BaseAPITest):
             amphora_id=self.amp_id)).json.get(self.root_tag)
         self._assert_amp_equal(self.amp_args, response)
 
+    @mock.patch('oslo_messaging.RPCClient.cast')
+    def test_delete(self, mock_cast):
+        self.amp_args = {
+            'status': constants.AMPHORA_READY,
+        }
+        amp = self.amphora_repo.create(self.session, **self.amp_args)
+
+        self.delete(self.AMPHORA_PATH.format(
+            amphora_id=amp.id), status=204)
+
+        response = self.get(self.AMPHORA_PATH.format(
+            amphora_id=amp.id)).json.get(self.root_tag)
+
+        self.assertEqual(constants.PENDING_DELETE, response[constants.STATUS])
+
+        payload = {constants.AMPHORA_ID: amp.id}
+        mock_cast.assert_called_with({}, 'delete_amphora', **payload)
+
+    @mock.patch('oslo_messaging.RPCClient.cast')
+    def test_delete_not_found(self, mock_cast):
+        self.delete(self.AMPHORA_PATH.format(amphora_id='bogus-id'),
+                    status=404)
+        mock_cast.assert_not_called()
+
+    @mock.patch('oslo_messaging.RPCClient.cast')
+    def test_delete_immutable(self, mock_cast):
+        self.amp_args = {
+            'status': constants.AMPHORA_ALLOCATED,
+        }
+        amp = self.amphora_repo.create(self.session, **self.amp_args)
+
+        self.delete(self.AMPHORA_PATH.format(
+            amphora_id=amp.id), status=409)
+
+        mock_cast.assert_not_called()
+
+    @mock.patch('oslo_messaging.RPCClient.cast')
+    def test_delete_authorized(self, mock_cast):
+        self.amp_args = {
+            'status': constants.AMPHORA_READY,
+        }
+        amp = self.amphora_repo.create(self.session, **self.amp_args)
+
+        self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
+        auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
+        self.conf.config(group='api_settings', auth_strategy=constants.TESTING)
+        with mock.patch.object(octavia.common.context.Context, 'project_id',
+                               self.project_id):
+            override_credentials = {
+                'service_user_id': None,
+                'user_domain_id': None,
+                'is_admin_project': True,
+                'service_project_domain_id': None,
+                'service_project_id': None,
+                'roles': ['load-balancer_member'],
+                'user_id': None,
+                'is_admin': True,
+                'service_user_domain_id': None,
+                'project_domain_id': None,
+                'service_roles': [],
+                'project_id': self.project_id}
+            with mock.patch(
+                    "oslo_context.context.RequestContext.to_policy_values",
+                    return_value=override_credentials):
+                self.delete(self.AMPHORA_PATH.format(amphora_id=amp.id),
+                            status=204)
+        # Reset api auth setting
+        self.conf.config(group='api_settings', auth_strategy=auth_strategy)
+
+        response = self.get(self.AMPHORA_PATH.format(
+            amphora_id=amp.id)).json.get(self.root_tag)
+
+        self.assertEqual(constants.PENDING_DELETE, response[constants.STATUS])
+
+        payload = {constants.AMPHORA_ID: amp.id}
+        mock_cast.assert_called_with({}, 'delete_amphora', **payload)
+
+    @mock.patch('oslo_messaging.RPCClient.cast')
+    def test_delete_not_authorized(self, mock_cast):
+        self.amp_args = {
+            'status': constants.AMPHORA_READY,
+        }
+        amp = self.amphora_repo.create(self.session, **self.amp_args)
+
+        self.conf = self.useFixture(oslo_fixture.Config(cfg.CONF))
+        auth_strategy = self.conf.conf.api_settings.get('auth_strategy')
+        self.conf.config(group='api_settings', auth_strategy=constants.TESTING)
+        with mock.patch.object(octavia.common.context.Context, 'project_id',
+                               self.project_id):
+            self.delete(self.AMPHORA_PATH.format(amphora_id=amp.id),
+                        status=403)
+        # Reset api auth setting
+        self.conf.config(group='api_settings', auth_strategy=auth_strategy)
+
+        mock_cast.assert_not_called()
+
     @mock.patch('oslo_messaging.RPCClient.cast')
     def test_failover(self, mock_cast):
         self.put(self.AMPHORA_FAILOVER_PATH.format(

octavia/tests/functional/db/test_repositories.py

@@ -4022,6 +4022,29 @@ class AmphoraRepositoryTest(BaseRepositoryTest):
                                                         self.FAKE_UUID_1)
         self.assertEqual(lb_ref, lb)
 
+    def test_and_set_status_for_delete(self):
+        # Normal path
+        amphora = self.create_amphora(self.FAKE_UUID_1,
+                                      status=constants.AMPHORA_READY)
+        self.amphora_repo.test_and_set_status_for_delete(self.session,
+                                                         amphora.id)
+        new_amphora = self.amphora_repo.get(self.session, id=amphora.id)
+        self.assertEqual(constants.PENDING_DELETE, new_amphora.status)
+
+        # Test deleted path
+        amphora = self.create_amphora(self.FAKE_UUID_2,
+                                      status=constants.DELETED)
+        self.assertRaises(sa_exception.NoResultFound,
+                          self.amphora_repo.test_and_set_status_for_delete,
+                          self.session, amphora.id)
+
+        # Test in use path
+        amphora = self.create_amphora(self.FAKE_UUID_3,
+                                      status=constants.AMPHORA_ALLOCATED)
+        self.assertRaises(exceptions.ImmutableObject,
+                          self.amphora_repo.test_and_set_status_for_delete,
+                          self.session, amphora.id)
+
 
 class AmphoraHealthRepositoryTest(BaseRepositoryTest):
     def setUp(self):

octavia/tests/unit/controller/queue/v1/test_endpoints.py

@@ -182,3 +182,8 @@ class TestEndpoints(base.TestCase):
         self.ep.update_amphora_agent_config(self.context, self.resource_id)
         self.ep.worker.update_amphora_agent_config.assert_called_once_with(
             self.resource_id)
+
+    def test_delete_amphora(self):
+        self.ep.delete_amphora(self.context, self.resource_id)
+        self.ep.worker.delete_amphora.assert_called_once_with(
+            self.resource_id)

octavia/tests/unit/controller/worker/v1/test_controller_worker.py

@@ -157,6 +157,34 @@ class TestControllerWorker(base.TestCase):
 
         self.assertEqual(AMP_ID, amp)
 
+    @mock.patch('octavia.controller.worker.v1.flows.'
+                'amphora_flows.AmphoraFlows.get_delete_amphora_flow',
+                return_value='TEST')
+    def test_delete_amphora(self,
+                            mock_get_delete_amp_flow,
+                            mock_api_get_session,
+                            mock_dyn_log_listener,
+                            mock_taskflow_load,
+                            mock_pool_repo_get,
+                            mock_member_repo_get,
+                            mock_l7rule_repo_get,
+                            mock_l7policy_repo_get,
+                            mock_listener_repo_get,
+                            mock_lb_repo_get,
+                            mock_health_mon_repo_get,
+                            mock_amp_repo_get):
+
+        _flow_mock.reset_mock()
+
+        cw = controller_worker.ControllerWorker()
+        cw.delete_amphora(_amphora_mock.id)
+
+        (base_taskflow.BaseTaskFlowEngine.taskflow_load.
+            assert_called_once_with('TEST'))
+
+        mock_get_delete_amp_flow.assert_called_once_with(_amphora_mock)
+        _flow_mock.run.assert_called_once_with()
+
     @mock.patch('octavia.db.repositories.AvailabilityZoneRepository.'
                 'get_availability_zone_metadata_dict')
     @mock.patch('octavia.controller.worker.v1.flows.'

releasenotes/notes/add-amphora-delete-69badba140f7b228.yaml

@@ -0,0 +1,4 @@
+---
+features:
+  - |
+    Added the ability to delete amphora that are not in use.