Merge "Bugfix: Clean up trusts for all deleted clusters"

Author: Zuul

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):