fix(controller): resource status and binding

Author: duanhongyi

rootfs/api/management/commands/measure_resources.py

@@ -19,7 +19,7 @@ def handle(self, *args, **options):
             task_id = uuid.uuid4().hex
             logger.info(f"pushing {task_id} resources to workflow_manager when {timezone.now()}")
             resource_list = []
-            for resource in Resource.objects.all():
+            for resource in Resource.objects.filter(status="Ready"):
                 resource_list.extend(resource.to_measurements(timestamp))
                 if len(resource_list) % 1000 == 0:
                     send_measurements.delay(resource_list)

rootfs/api/models/resource.py

@@ -194,53 +194,9 @@ def message(self):
             return ""
 
     def retrieve(self, *args, **kwargs):
-        update_flag = False
-        if self.status != "Ready":
-            try:
-                resp_i = self._scheduler.svcat.get_instance(
-                    self.app.id, self.name).json()
-                self.status = resp_i.get('status', {}).\
-                    get('lastConditionState')
-                self.options = resp_i.get('spec', {}).get('parameters', {})
-                update_flag = True
-            except KubeException as e:
-                logger.info("retrieve instance info error: {}".format(e))
-        if self.binding != "Ready":
-            try:
-                # We raise an exception when a resource doesn't exist
-                resp_b = self._scheduler.svcat.get_binding(
-                    self.app.id, self.name).json()
-                self.binding = resp_b.get('status', {}).\
-                    get('lastConditionState')
-                update_flag = True
-                secret_name = resp_b.get('spec', {}).get('secretName')
-                if secret_name:
-                    resp_s = self._scheduler.secret.get(
-                        self.app.id, secret_name).json()
-                    self.data = resp_s.get('data', {})
-                    update_flag = True
-            except KubeException as e:
-                logger.info("retrieve binding info error: {}".format(e))
-        if update_flag is True:
+        if self._retrieve_status() or self._retrieve_binding():
             self.save()
-        return self.status == "Ready"
-
-    def detach_resource(self, *args, **kwargs):
-        if self.binding != "Ready":
-            try:
-                resp_b = self._scheduler.svcat.get_binding(
-                    self.app.id, self.name).json()
-                secret_name = resp_b.get('spec', {}).get('secretName')
-                if secret_name:
-                    self._scheduler.secret.delete(self.app.id, secret_name)
-                self._scheduler.svcat.delete_binding(
-                    self.app.id, self.name)
-            except KubeException as e:
-                logger.info("delete binding info error: {}".format(e))
-            self.binding = None
-
-        if (self.status != "Ready") or (not self.binding):
-            self.delete()
+        return self.status == self.binding == "Ready"
 
     def to_measurements(self, timestamp: float):
         return [{
@@ -252,3 +208,40 @@ def to_measurements(self, timestamp: float):
             "usage": 1,
             "timestamp": int(timestamp)
         }]
+
+    def _retrieve_status(self):
+        changed = False
+        try:
+            response = self._scheduler.svcat.get_instance(
+                self.app.id, self.name).json()
+            status = response.get('status', {}).get('lastConditionState')
+            options = response.get('spec', {}).get('parameters', {})
+            if self.status != status:
+                self.status = status
+                changed = True
+            if self.options != options:
+                self.options = options
+                changed = True
+        except KubeException as e:
+            logger.info("retrieve instance info error: {}".format(e))
+        return changed
+
+    def _retrieve_binding(self):
+        changed = False
+        try:
+            # We raise an exception when a resource doesn't exist
+            response = self._scheduler.svcat.get_binding(self.app.id, self.name).json()
+            binding = response.get('status', {}).get('lastConditionState')
+            secret_name = response.get('spec', {}).get('secretName')
+            if self.binding != binding:
+                self.binding = binding
+                changed = True
+            if secret_name:
+                response = self._scheduler.secret.get(self.app.id, secret_name).json()
+                data = response.get('data', {})
+                if self.data != data:
+                    self.data = data
+                    changed = True
+        except KubeException as e:
+            logger.info("retrieve binding info error: {}".format(e))
+        return changed

rootfs/api/tasks.py

@@ -25,8 +25,6 @@ def retrieve_resource(resource):
                 retrieve_resource.apply_async(args=(resource, ), countdown=30)
             elif t < 3600 * 12:
                 retrieve_resource.apply_async(args=(resource, ), countdown=1800)
-            else:
-                resource.detach_resource()
     except (Exception, Resource.DoesNotExist) as e:
         signals.got_request_exception.send(sender=task_id)
         if isinstance(e, Resource.DoesNotExist):
@@ -110,21 +108,22 @@ def mount_app(app, user, volume):
 
 
 @shared_task(
-    retry_kwargs={'max_retries': None}
+    retry_kwargs={'max_retries': 3}
 )
 def downstream_model_owner(app, old_owner, new_owner):
     task_id = uuid.uuid4().hex
     signals.request_started.send(sender=task_id)
     try:
         for downstream_model in [
-            models.appsettings.AppSettings, models.build.Build, models.certificate.Certificate,
-            models.config.Config, models.domain.Domain, models.key.Key, models.release.Release,
-            models.resource.Resource, models.tls.TLS, models.service.Service,
-            models.volume.Volume, models.gateway.Gateway, models.gateway.Route]:
+                models.appsettings.AppSettings, models.build.Build, models.config.Config,
+                models.domain.Domain, models.release.Release, models.resource.Resource,
+                models.tls.TLS, models.service.Service, models.volume.Volume,
+                models.gateway.Gateway, models.gateway.Route]:
             downstream_model.objects.filter(owner=old_owner, app=app).update(owner=new_owner)
         app.owner = new_owner
         app.save()
     except Exception as e:
         signals.got_request_exception.send(sender=task_id)
+        raise e
     else:
         signals.request_finished.send(sender=task_id)