chore(controller): use the kubectl api to override pods restart

Author: duanhongyi

charts/controller/templates/controller-clusterrole.yaml

@@ -48,7 +48,7 @@ rules:
   verbs: ["get", "list", "delete", "update"]
 - apiGroups: ["extensions", "apps"]
   resources: ["deployments"]
-  verbs: ["get", "list", "create", "update", "delete"]
+  verbs: ["get", "list", "create", "update", "patch", "delete"]
 - apiGroups: ["extensions", "apps"]
   resources: ["deployments/scale", "replicasets/scale"]
   verbs: ["get", "update"]

rootfs/api/models/app.py

@@ -356,84 +356,27 @@ def delete(self, *args, **kwargs):
 
     def restart(self, **kwargs):  # noqa
         """
-        Restart found pods by deleting them (RC / Deployment will recreate).
-        Wait until they are all drained away and RC / Deployment has gotten to a good state
+         Restart deployments with the kubectl rollout api
         """
-        try:
-            # Resolve single pod name if short form (cmd-1269180282-1nyfz) is passed
-            if 'name' in kwargs and kwargs['name'].count('-') == 2:
-                kwargs['name'] = '{}-{}'.format(kwargs['id'], kwargs['name'])
-
-            # Iterate over RCs / RSs to get total desired count if not a single item
-            desired = 1
-            if 'name' not in kwargs:
-                desired = 0
-                labels = self._scheduler_filter(**kwargs)
-                # fetch RS (which represent Deployments)
-                controllers = self._scheduler.rs.get(kwargs['id'], labels=labels).json()['items']
-                if not controllers:
-                    controllers = []
-                for controller in controllers:
-                    desired += controller['spec']['replicas']
-        except KubeException:
-            # Nothing was found
-            return []
-
+        deployments = []
+        if 'type' in kwargs and kwargs['type'] in self.structure:
+            deployments.append(self._get_job_id(kwargs['type']))
+        else:
+            for scale_type, _ in self.structure.items():
+                deployments.append(self._get_job_id(scale_type))
         try:
             tasks = [
                 functools.partial(
-                    self._scheduler.pod.delete,
+                    self._scheduler.deployment.restart,
                     self.id,
-                    pod['name']
-                ) for pod in self.list_pods(**kwargs)
+                    deployment
+                ) for deployment in deployments
             ]
-
             apply_tasks(tasks)
         except Exception as e:
-            err = "warning, some pods failed to stop:\n{}".format(str(e))
+            err = "warning, some pods failed to restart:\n{}".format(str(e))
             self.log(err, logging.WARNING)
 
-        # Wait for pods to start
-        try:
-            timeout = 300  # 5 minutes
-            elapsed = 0
-            while True:
-                # timed out
-                if elapsed >= timeout:
-                    raise DryccException('timeout - 5 minutes have passed and pods are not up')
-
-                # restarting a single pod behaves differently, fetch the *newest* pod
-                # and hope it is the right one. Comes back sorted
-                if 'name' in kwargs:
-                    del kwargs['name']
-                    pods = self.list_pods(**kwargs)
-                    # Add in the latest name
-                    if len(pods) == 0:
-                        # if pod is not even scheduled wait for it and pass dummy kwargs
-                        # to indicate restart of a single pod
-                        kwargs['name'] = "dummy"
-                        continue
-                    kwargs['name'] = pods[0]['name']
-                    pods = pods[0]
-
-                actual = 0
-                for pod in self.list_pods(**kwargs):
-                    if pod['state'] == 'up':
-                        actual += 1
-
-                if desired == actual:
-                    break
-
-                elapsed += 5
-                time.sleep(5)
-        except Exception as e:
-            err = "warning, some pods failed to start:\n{}".format(str(e))
-            self.log(err, logging.WARNING)
-
-        # Return the new pods
-        pods = self.list_pods(**kwargs)
-        return pods
-
     def _clean_app_logs(self):
         """Delete application logs stored by the logger component"""
         try:

rootfs/api/tests/test_pods.py

@@ -779,39 +779,20 @@ def test_restart_pods(self, mock_requests):
 
         # restart all pods
         response = self.client.post('/v2/apps/{}/pods/restart'.format(app_id))
-        self.assertEqual(response.status_code, 200, response.data)
-        # Compare restarted pods to all pods
-        self.assertEqual(len(response.data), 12)
+        self.assertEqual(response.status_code, 204, response.data)
 
         # restart only the workers
         response = self.client.post('/v2/apps/{}/pods/worker/restart'.format(app_id))
-        self.assertEqual(response.status_code, 200, response.data)
-        # Compare restarted pods to only worker pods
-        self.assertEqual(len(response.data), 8)
+        self.assertEqual(response.status_code, 204, response.data)
 
         # restart only the web
         response = self.client.post('/v2/apps/{}/pods/web/restart'.format(app_id))
-        self.assertEqual(response.status_code, 200, response.data)
-        # Compare restarted pods to only worker pods
-        self.assertEqual(len(response.data), 4)
+        self.assertEqual(response.status_code, 204, response.data)
 
         # restart only one of the web pods
         pods = application.list_pods(type='web')
         self.assertEqual(len(pods), 4)
 
-        pod = pods.pop()
-        response = self.client.post('/v2/apps/{}/pods/web/{}/restart'.format(app_id, pod['name']))
-        self.assertEqual(response.status_code, 200, response.data)
-        self.assertEqual(len(response.data), 1)
-        self.assertEqual(response.data[0]['type'], 'web')
-
-        # restart only one web port but using the short name of web-asdfg
-        name = 'web-' + pod['name'].split('-').pop()
-        response = self.client.post('/v2/apps/{}/pods/web/{}/restart'.format(app_id, name))
-        self.assertEqual(response.status_code, 200, response.data)
-        self.assertEqual(len(response.data), 1)
-        self.assertEqual(response.data[0]['type'], 'web')
-
     def test_list_pods_failure(self, mock_requests):
         """
         Listing all available pods exceptions

rootfs/api/urls.py

@@ -42,10 +42,6 @@
     re_path(
         r"^apps/(?P<id>{})/pods/(?P<type>[-_\w.]+)/restart/?$".format(settings.APP_URL_REGEX),
         views.PodViewSet.as_view({'post': 'restart'})),
-    re_path(
-        r"^apps/(?P<id>{})/pods/(?P<type>[-_\w]+)/(?P<name>[-_\w]+)/restart/?$".format(
-            settings.APP_URL_REGEX),
-        views.PodViewSet.as_view({'post': 'restart'})),
     # list pods
     re_path(
         r"^apps/(?P<id>{})/pods/(?P<type>[-_\w]+)/(?P<name>[-_\w]+)/?$".format(

rootfs/api/views.py

@@ -343,16 +343,8 @@ def list(self, *args, **kwargs):
         return Response(pagination, status=status.HTTP_200_OK)
 
     def restart(self, *args, **kwargs):
-        if "name" in kwargs:  # a single pod uses sync
-            pods = self.get_app().restart(**kwargs)
-        else:  # multi pod uses async
-            restart_app.delay(self.get_app(), **kwargs)
-            pods = self.get_app().list_pods(**kwargs)
-        data = self.get_serializer(pods, many=True).data
-        # fake out pagination for now
-        # pagination = {'results': data, 'count': len(data)}
-        pagination = data
-        return Response(pagination, status=status.HTTP_200_OK)
+        restart_app.delay(self.get_app(), **kwargs)
+        return Response(status=status.HTTP_204_NO_CONTENT)
 
 
 class AppSettingsViewSet(AppResourceViewSet):

rootfs/scheduler/resources/deployment.py

@@ -174,6 +174,34 @@ def scale(self, namespace, name, image, entrypoint, command, **kwargs):
             kwargs['previous_replicas'] = current
             self.wait_until_ready(namespace, name, **kwargs)
 
+    def restart(self, namespace, name):
+        url = self.api(
+            "/namespaces/{}/deployments/{}?fieldManager=kubectl-rollout&pretty=true",
+            namespace, name
+        )
+        restartedAt = "%sZ" % (timedelta(seconds=3) + datetime.utcnow()).isoformat("T")
+        response = self.http_patch(
+            url,
+            data=json.dumps({
+                "spec": {
+                    "template": {
+                        "metadata": {
+                            "annotations": {
+                                "kubectl.kubernetes.io/restartedAt": restartedAt,
+                            }
+                        }
+                    }
+                }
+            }),
+            headers={"Content-Type": "application/merge-patch+json"},
+        )
+        if self.unhealthy(response.status_code):
+            raise KubeHTTPException(
+                response,
+                'restart Deployment "{}" in Namespace "{}"', name, namespace
+            )
+        return response
+
     def in_progress(self, namespace, name, timeout, batches, replicas, tags):
         """
         Determine if a Deployment has a deploy in progress