feat(k8s): differential update state to k8s

Author: duanhongyi

rootfs/api/management/commands/load_db_state_to_k8s.py

@@ -20,14 +20,9 @@ class Command(BaseCommand):
     to k8s.
     """
 
-    def _deploy(self, app, release):
-        if release.build is None:
-            print('WARNING: {} has no build associated with '
-                  'its latest release. Skipping deployment...'.format(app))
-            return
-
+    def _deploy(self, app):
         try:
-            app.deploy(release)
+            app.state_to_k8s()
         except AlreadyExists as error:
             logger.debug(error)
             print('WARNING: {} has a deployment in progress. '
@@ -50,13 +45,9 @@ def handle(self, *args, **options):
                 cert.attach_in_kubernetes(domain)
 
         # deploy apps
-        print("Deploying available applications")
+        print("Deploying available applications.")
         for app in App.objects.all():
-            release = app.release_set.filter(failed=False).latest()
-            if release.canary:
-                self._deploy(app, app.release_set.filter(failed=False, canary=False).latest())
-            self._deploy(app, release)
-
+            self._deploy(app)
         print("Done Publishing DB state to kubernetes.")
 
     def save_apps(self):

rootfs/api/models/app.py

@@ -512,6 +512,33 @@ def image_pull_secret(self, namespace, registry, image):
 
         return name
 
+    def state_to_k8s(self):
+        def _load_procfile_types(canary):
+            procfile_types = set()
+            for procfile_type, scale in self.structure.items():
+                response = self.scheduler().deployment.get(
+                    self.id, self._get_job_id(procfile_type, canary),
+                    ignore_exception=True)
+                if response.status_code == 404 and scale > 0:
+                    procfile_types.add(procfile_type)
+                elif response.status_code != 200:
+                    data = response.json()
+                    self.log('get deployment status_code {}, message: {}'.format(
+                        response.status_code, data.get("message", "")), logging.ERROR)
+            return procfile_types
+
+        release = self.release_set.filter(failed=False).latest()
+        if release.build is None:
+            self.log('the last release does not have a build, skipping deployment...')
+            return
+        procfile_types = _load_procfile_types(False)
+        if release.canary:
+            procfile_types = procfile_types.union(_load_procfile_types(canary=True))
+        if len(procfile_types) == 0:
+            self.log('the cluster status is the latest, skipping deployment...')
+            return
+        self.deploy(release, procfile_types)
+
     def set_application_config(self, release, procfile_type):
         """
         Creates the application config as a secret in Kubernetes and

rootfs/scheduler/resources/deployment.py

@@ -9,7 +9,7 @@ class Deployment(Resource):
     api_prefix = 'apis'
     api_version = 'apps/v1'
 
-    def get(self, namespace, name=None, **kwargs):
+    def get(self, namespace, name=None, ignore_exception=False, **kwargs):
         """
         Fetch a single Deployment or a list
         """
@@ -24,7 +24,7 @@ def get(self, namespace, name=None, **kwargs):
 
         url = self.api(url, *args)
         response = self.http_get(url, params=self.query_params(**kwargs))
-        if self.unhealthy(response.status_code):
+        if self.unhealthy(response.status_code) and not ignore_exception:
             args.reverse()  # error msg is in reverse order
             raise KubeHTTPException(response, message, *args)
 
@@ -115,40 +115,43 @@ def manifest(self, namespace, name, image, command, args, spec_annotations, **kw
 
         return manifest
 
-    def create(self, namespace, name, image, command, args, spec_annotations, **kwargs):
+    def create(self, namespace, name, image, command, args,
+               spec_annotations, ignore_exception=False, **kwargs):
         manifest = self.manifest(namespace, name, image,
                                  command, args, spec_annotations, **kwargs)
 
         url = self.api("/namespaces/{}/deployments", namespace)
         response = self.http_post(url, json=manifest)
         if self.unhealthy(response.status_code):
             self.log(namespace, 'template: {}'.format(json.dumps(manifest, indent=4)), 'DEBUG')
-            raise KubeHTTPException(
-                response,
-                'create Deployment "{}" in Namespace "{}"', name, namespace
-            )
-
-        self.wait_until_updated(namespace, name)
-        self.wait_until_ready(namespace, name, **kwargs)
-
+            if not ignore_exception:
+                raise KubeHTTPException(
+                    response,
+                    'create Deployment "{}" in Namespace "{}"', name, namespace
+                )
+        else:
+            self.wait_until_updated(namespace, name)
+            self.wait_until_ready(namespace, name, **kwargs)
         return response
 
-    def update(self, namespace, name, image, command, args, spec_annotations, **kwargs):
+    def update(self, namespace, name, image, command, args,
+               spec_annotations, ignore_exception=False, **kwargs):
         manifest = self.manifest(namespace, name, image,
                                  command, args, spec_annotations, **kwargs)
 
         url = self.api("/namespaces/{}/deployments/{}", namespace, name)
         response = self.http_put(url, json=manifest)
         if self.unhealthy(response.status_code):
             self.log(namespace, 'template: {}'.format(json.dumps(manifest, indent=4)), 'DEBUG')
-            raise KubeHTTPException(response, 'update Deployment "{}"', name)
-
-        self.wait_until_updated(namespace, name)
-        self.wait_until_ready(namespace, name, **kwargs)
-
+            if not ignore_exception:
+                raise KubeHTTPException(response, 'update Deployment "{}"', name)
+        else:
+            self.wait_until_updated(namespace, name)
+            self.wait_until_ready(namespace, name, **kwargs)
         return response
 
-    def patch(self, namespace, name, image, command, args, spec_annotations, **kwargs):
+    def patch(self, namespace, name, image, command, args,
+              spec_annotations, ignore_exception=False, **kwargs):
         manifest = self.manifest(namespace, name, image,
                                  command, args, spec_annotations, **kwargs)
 
@@ -161,11 +164,11 @@ def patch(self, namespace, name, image, command, args, spec_annotations, **kwarg
 
         if self.unhealthy(response.status_code):
             self.log(namespace, 'template: {}'.format(json.dumps(manifest, indent=4)), 'DEBUG')
-            raise KubeHTTPException(response, 'patch Deployment "{}"', name)
-
-        self.wait_until_updated(namespace, name)
-        self.wait_until_ready(namespace, name, **kwargs)
-
+            if not ignore_exception:
+                raise KubeHTTPException(response, 'patch Deployment "{}"', name)
+        else:
+            self.wait_until_updated(namespace, name)
+            self.wait_until_ready(namespace, name, **kwargs)
         return response
 
     def delete(self, namespace, name, ignore_exception=False):
@@ -200,7 +203,7 @@ def scale(self, namespace, name, **kwargs):
             kwargs['previous_replicas'] = current
             self.wait_until_ready(namespace, name, **kwargs)
 
-    def restart(self, namespace, name):
+    def restart(self, namespace, name, ignore_exception=False):
         url = self.api(
             "/namespaces/{}/deployments/{}?fieldManager=kubectl-rollout&pretty=true",
             namespace, name
@@ -221,7 +224,7 @@ def restart(self, namespace, name):
             }),
             headers={"Content-Type": "application/merge-patch+json"},
         )
-        if self.unhealthy(response.status_code):
+        if self.unhealthy(response.status_code) and not ignore_exception:
             raise KubeHTTPException(
                 response,
                 'restart Deployment "{}" in Namespace "{}"', name, namespace