feat(config): set config to celery

Author: duanhongyi

rootfs/api/tasks.py

@@ -62,6 +62,38 @@ def run_pipeline(release, *args, **kwargs):
         signals.request_finished.send(sender=task_id)
 
 
+@shared_task(
+    autoretry_for=(ServiceUnavailable, ),
+    retry_kwargs={'max_retries': None}
+)
+def run_deploy(release, config):
+    task_id = uuid.uuid4().hex
+    signals.request_started.send(sender=task_id)
+    try:
+        if release.build is not None:
+            procfile_types = set()
+            for field, diff in config.diff().items():
+                if field in config.procfile_fields:
+                    for value in diff.values():
+                        procfile_types.update(value.keys())
+            # all_diff_fields changed, deploy all.
+            procfile_types = procfile_types if procfile_types else None
+            config.app.deploy(release, procfile_types)
+        release.state = "succeed"
+        release.save()
+    except Exception as e:
+        release.state = "crashed"
+        release.failed = True
+        if release.summary:
+            release.summary += " "
+        release.summary += "{} deployed a config that failed".format(release.owner)
+        # Get the exception that has occured
+        release.exception = "error: {}".format(str(e))
+        release.save()
+    finally:
+        signals.request_finished.send(sender=task_id)
+
+
 @shared_task(
     autoretry_for=(ServiceUnavailable, ),
     retry_jitter=True,

rootfs/api/tests/test_config.py

@@ -312,7 +312,11 @@ def test_config_deploy_failure(self, mock_requests):
             url = f'/v2/apps/{app_id}/config'
             body = {'values': json.dumps({'test': "testvalue"})}
             response = self.client.post(url, body)
-            self.assertEqual(response.status_code, 400)
+            self.assertEqual(response.status_code, 201)
+            app = App.objects.get(id=app_id)
+            release = app.release_set.latest()
+            self.assertEqual(release.failed, True)
+            self.assertEqual(release.exception, "error: Boom!")
 
     def test_invalid_config_keys(self, mock_requests):
         """Test that invalid config keys are rejected.
@@ -416,8 +420,10 @@ def test_config_failures(self, mock_requests):
             url = f'/v2/apps/{app_id}/config'
             body = {'values': json.dumps({'test': "testvalue"})}
             response = self.client.post(url, body)
-            self.assertEqual(response.status_code, 400)
+            self.assertEqual(response.status_code, 201)
             self.assertEqual(app.release_set.latest().version, 4)
+            self.assertEqual(app.release_set.latest().failed, True)
+            self.assertEqual(app.release_set.latest().exception, "error: Boom!")
             self.assertEqual(app.release_set.filter(failed=False).latest().version, 3)
 
         # create a build to see that the new release is created with the last successful config

rootfs/api/views.py

@@ -27,7 +27,7 @@
 
 from api import monitor, models, permissions, serializers, viewsets, authentication
 from api.tasks import scale_app, restart_app, mount_app, downstream_model_owner, \
-    delete_pod
+    delete_pod, run_deploy
 from api.exceptions import AlreadyExists, ServiceUnavailable, DryccException
 
 from django.views.decorators.cache import never_cache
@@ -394,29 +394,9 @@ def post_save(self, config):
         try:
             release = latest_release.new(
                 self.request.user, config=config, build=latest_release.build)
-            if release.build is not None:
-                procfile_types = set()
-                for field, diff in config.diff().items():
-                    if field in config.procfile_fields:
-                        for value in diff.values():
-                            procfile_types.update(value.keys())
-                # all_diff_fields changed, deploy all.
-                procfile_types = procfile_types if procfile_types else None
-                config.app.deploy(release, procfile_types)
-            release.state = "succeed"
-            release.save()
+            run_deploy.delay(release, config)
         except Exception as e:
-            if 'release' in locals():
-                release.state = "crashed"
-                release.failed = True
-                if release.summary:
-                    release.summary += " "
-                release.summary += "{} deployed a config that failed".format(self.request.user)
-                # Get the exception that has occured
-                release.exception = "error: {}".format(str(e))
-                release.save()
-            else:
-                config.delete()
+            config.delete()
             if isinstance(e, AlreadyExists):
                 raise
             raise DryccException(str(e)) from e

rootfs/scheduler/resources/pod.py

@@ -749,33 +749,41 @@ def _handle_not_ready_pods(self, namespace, labels):
                     message = "\n".join([x.strip() for x in event['message'].split("\n")])
                     raise KubeException(message)
 
+    def _get_probe_timeout(self, probe):
+        period_seconds = probe.get('periodSeconds', 10)
+        timeout_seconds = probe.get('timeoutSeconds', 1)
+        failure_threshold = probe.get('failureThreshold', 3)
+        success_threshold = probe.get('successThreshold', 1)
+        initial_delay_seconds = probe.get('initialDelaySeconds', 0)
+        if period_seconds > initial_delay_seconds:
+            timeout = 0
+        else:
+            timeout = initial_delay_seconds
+        timeout += (period_seconds + timeout_seconds) * failure_threshold
+        timeout += (period_seconds + timeout_seconds) * success_threshold
+        return int(timeout)
+
     def deploy_probe_timeout(self, timeout, namespace, labels, containers):
         """
         Added in additional timeouts based on readiness and liveness probe
 
         Uses the max of the two instead of combining them as the checks are stacked.
         """
-
         container_name = '{}-{}'.format(labels['app'], labels['type'])
         container = self.pod.find_container(container_name, containers)
 
         # get health info from container
         added_timeout = []
         if 'readinessProbe' in container:
-            # If there is initial delay on the readiness check then timeout needs to be higher
-            # this is to account for kubernetes having readiness check report as failure until
-            # the initial delay period is up
-            added_timeout.append(int(container['readinessProbe'].get('initialDelaySeconds', 50)))
-
+            added_timeout.append(self._get_probe_timeout(container['readinessProbe']))
         if 'livenessProbe' in container:
-            # If there is initial delay on the readiness check then timeout needs to be higher
-            # this is to account for kubernetes having liveness check report as failure until
-            # the initial delay period is up
-            added_timeout.append(int(container['livenessProbe'].get('initialDelaySeconds', 50)))
-
+            added_timeout.append(self._get_probe_timeout(container['livenessProbe']))
         if added_timeout:
             delay = max(added_timeout)
-            self.log(namespace, "adding {}s on to the original {}s timeout to account for the initial delay specified in the liveness / readiness probe".format(delay, timeout))  # noqa
+            self.log(
+                namespace,
+                "adding {}s on to the original {}s timeout for the probe".format(delay, timeout))
             timeout += delay
-
+        if 'startupProbe' in container:
+            timeout += self._get_probe_timeout(container['startupProbe'])
         return timeout