Bugfix: Clean up trusts for all deleted clusters

Cluster conductor creates trusts for all drivers, but does not clean
them up. The Heat driver has previously performed this action.

This change moves the lifecycle of trust and certificate creation
to the Conductor, so drivers do not need to clean up resources they
didn't create.

Change-Id: I2b3e99589d2d3069191d0727406601f0647a9722

Author: dalees

magnum/common/keystone.py

@@ -263,14 +263,16 @@ def create_trustee(self, username, password):
                                                 domain_id=domain_id)
         return user
 
-    def delete_trustee(self, trustee_id):
+    def delete_trustee(self, trustee_user_id):
+        if trustee_user_id is None:
+            return
         try:
-            self.domain_admin_client.users.delete(trustee_id)
+            self.domain_admin_client.users.delete(trustee_user_id)
         except kc_exception.NotFound:
             pass
         except Exception:
             LOG.exception('Failed to delete trustee')
-            raise exception.TrusteeDeleteFailed(trustee_id=trustee_id)
+            raise exception.TrusteeDeleteFailed(trustee_id=trustee_user_id)
 
     def get_validate_region_name(self, region_name):
         if region_name is None:

magnum/conductor/handlers/common/trust_manager.py

@@ -44,20 +44,20 @@ def create_trustee_and_trust(osc, cluster):
 
 
 def delete_trustee_and_trust(osc, context, cluster):
+    kst = osc.keystone()
     try:
-        kst = osc.keystone()
-
-        # The cluster which is upgraded from Liberty doesn't have trust_id
         if cluster.trust_id:
             kst.delete_trust(context, cluster)
+            cluster.trust_id = None
     except Exception:
         # Exceptions are already logged by keystone().delete_trust
         pass
     try:
-        # The cluster which is upgraded from Liberty doesn't have
-        # trustee_user_id
         if cluster.trustee_user_id:
-            osc.keystone().delete_trustee(cluster.trustee_user_id)
+            kst.delete_trustee(cluster.trustee_user_id)
+            cluster.trustee_user_id = None
+            cluster.trustee_username = None
+            cluster.trustee_password = None
     except Exception:
         # Exceptions are already logged by keystone().delete_trustee
         pass

magnum/service/periodic.py

@@ -21,10 +21,13 @@
 
 from pycadf import cadftaxonomy as taxonomy
 
+from magnum.common import clients
 from magnum.common import context
 from magnum.common import exception
 from magnum.common import profiler
 from magnum.common import rpc
+from magnum.conductor.handlers.common import cert_manager
+from magnum.conductor.handlers.common import trust_manager
 from magnum.conductor import monitors
 from magnum.conductor import utils as conductor_utils
 import magnum.conf
@@ -95,6 +98,14 @@ def update_status(self):
                 taxonomy.OUTCOME_FAILURE, self.cluster)
         # if we're done with it, delete it
         if self.cluster.status == objects.fields.ClusterStatus.DELETE_COMPLETE:
+            # Clean up trusts and certificates, if they still exist.
+            os_client = clients.OpenStackClients(self.ctx)
+            LOG.debug("Calling delete_trustee_and_trusts from periodic "
+                      "DELETE_COMPLETE")
+            trust_manager.delete_trustee_and_trust(os_client, self.ctx,
+                                                   self.cluster)
+            cert_manager.delete_certificates_from_cluster(self.cluster,
+                                                          context=self.ctx)
             # delete all the nodegroups that belong to this cluster
             for ng in objects.NodeGroup.list(self.ctx, self.cluster.uuid):
                 ng.destroy()

magnum/tests/unit/conductor/handlers/common/test_trust_manager.py

@@ -89,7 +89,7 @@ def test_delete_trustee_and_trust(self):
             context, mock_cluster
         )
         mock_keystone.delete_trustee.assert_called_once_with(
-            mock_cluster.trustee_user_id,
+            'trustee_user_id',
         )
 
     def test_delete_trustee_and_trust_without_trust_id(self):
@@ -105,7 +105,7 @@ def test_delete_trustee_and_trust_without_trust_id(self):
 
         self.assertEqual(0, mock_keystone.delete_trust.call_count)
         mock_keystone.delete_trustee.assert_called_once_with(
-            mock_cluster.trustee_user_id,
+            'trustee_user_id',
         )
 
     def test_delete_trustee_and_trust_without_trustee_user_id(self):