fix(controller): miss err msg

duanhongyi · duanhongyi · commit 2f84679f6129 · 2024-02-25T22:45:02.000+08:00
diff --git a/charts/controller/templates/controller-celery-deloyment.yaml b/charts/controller/templates/controller-celery-deloyment.yaml
@@ -39,12 +39,14 @@ spec:
           - $(DRYCC_REDIS_ADDRS),$(DRYCC_CONTROLLER_API_SERVICE_HOST):$(DRYCC_CONTROLLER_API_SERVICE_PORT)
         {{- include "controller.envs" . | indent 8 }}
       containers:
-      - name: drycc-controller-celery
+      {{- range $key := (list "low" "middle" "high") }}
+      - name: drycc-controller-celery-{{$key}}
         image: {{$.Values.imageRegistry}}/{{$.Values.imageOrg}}/controller:{{$.Values.imageTag}}
         imagePullPolicy: {{$.Values.imagePullPolicy}}
         args:
           - /bin/bash
           - -c
-          - celery -A api worker --autoscale=32,1 --loglevel=WARNING
+          - celery -A api worker -Q {{$key}} --autoscale=32,1 --loglevel=WARNING
         {{- include "controller.limits" $ | indent 8 }}
         {{- include "controller.envs" $ | indent 8 }}
+      {{- end }}
diff --git a/rootfs/api/consumers.py b/rootfs/api/consumers.py
@@ -1,5 +1,7 @@
 import json
 import time
+import six
+import ssl
 import aiohttp
 import asyncio
 import collections
@@ -10,6 +12,7 @@
 from kubernetes.client import Configuration
 from kubernetes.client.api import core_v1_api
 from kubernetes.stream import stream
+from kubernetes.stream.ws_client import STDOUT_CHANNEL, STDERR_CHANNEL, ERROR_CHANNEL
 
 from channels.db import database_sync_to_async
 from channels.exceptions import DenyConnection
@@ -85,25 +88,27 @@ async def connect(self):
         await super().connect()
         self.pod_id = self.scope["url_route"]["kwargs"]["pod_id"]
 
-    async def send(self, data):
+    async def send(self, data, channel=STDOUT_CHANNEL):
+        channel_prefix = chr(channel)
         if data is None:
             return
         elif isinstance(data, bytes):
-            await super().send(bytes_data=data)
+            channel_prefix = six.binary_type(channel_prefix, "ascii")
+            await super().send(bytes_data=channel_prefix+data)
         elif isinstance(data, str):
-            await super().send(text_data=data)
+            await super().send(text_data=channel_prefix+data)
 
     async def task(self):
         deadline = time.time() + settings.DRYCC_APP_POD_EXEC_TIMEOUT
         while self.stream.is_open() and self.conneted and time.time() < deadline:
-            await sync_to_async(self.stream.update)(0.1)
-            if self.stream.peek_stdout():
-                data = self.stream.read_stdout()
-            elif self.stream.peek_stderr():
-                data = self.stream.read_stderr()
-            else:
-                data = None
-            await self.send(data)
+            try:
+                await sync_to_async(self.stream.update)(0.1)
+                for channel in (ERROR_CHANNEL, STDOUT_CHANNEL, STDERR_CHANNEL):
+                    if channel in self.stream._channels:
+                        data = self.stream.read_channel(channel)
+                        await self.send(data, channel)
+            except ssl.SSLEOFError:
+                break
         await self.close(code=1000)
 
     async def disconnect(self, close_code):
@@ -115,14 +120,9 @@ async def receive(self, text_data=None, bytes_data=None):
         if self.stream is None and text_data is not None:
             args = (self.kubernetes.connect_get_namespaced_pod_exec, self.pod_id, self.id)
             kwargs = json.loads(text_data)
-            kwargs.update({"stderr": True, "stdout": True})
-            if kwargs["stdin"]:
-                kwargs.update({"_preload_content": False})
-                self.stream = stream(*args, **kwargs)
-                asyncio.create_task(self.task())
-            else:
-                await self.send(stream(*args, **kwargs))
-                await self.close(code=1000)
+            kwargs.update({"stderr": True, "stdout": True, "_preload_content": False})
+            self.stream = stream(*args, **kwargs)
+            asyncio.create_task(self.task())
         elif self.stream is not None:
             data = text_data if text_data else bytes_data
             channel, data = ord(data[0]), data[1:]
diff --git a/rootfs/api/models/app.py b/rootfs/api/models/app.py
@@ -381,7 +381,7 @@ def deploy(self, release, force_deploy=False, rollback_on_failure=True):  # noqa
         self.cleanup_old()
         release.cleanup_old()
 
-    def mount(self, user, volume):
+    def mount(self, user, volume, structure=None):
         if self.release_set.filter(failed=False).latest().build is None:
             raise DryccException('No build associated with this release')
         release = self.release_set.filter(failed=False).latest()
@@ -391,7 +391,8 @@ def mount(self, user, volume):
                 user,
                 volume,
                 self.release_set.filter(failed=False, canary=False).latest(),
-                app_settings
+                app_settings,
+                structure=structure,
             )
         self._mount(user, volume, release, app_settings)
 
@@ -699,14 +700,14 @@ def _clean_app_logs(self):
             err = 'Error deleting existing application logs: {}'.format(e)
             self.log(err, logging.WARNING)
 
-    def _mount(self, user, volume, release, app_settings):
-        volumes = list(Volume.objects.filter(app=self).exclude(name=volume.name))
-        volumes.append(volume)
+    def _mount(self, user, volume, release, app_settings, structure=None):
+        volumes = Volume.objects.filter(app=self)
         tasks = []
-        for scale_type, replicas in self.structure.items():
+        for scale_type, replicas in structure.items() if structure else self.structure.items():
             if not release.canary or scale_type in app_settings.canaries:
                 replicas = self.structure.get(scale_type, 0)
-                scale_type_volumes = [_ for _ in volumes if scale_type in _.path.keys()]
+                scale_type_volumes = [
+                    volume for volume in volumes if scale_type in volume.path.keys()]
                 data = self._gather_app_settings(
                     release, app_settings, scale_type, replicas, volumes=scale_type_volumes)
                 deployment = self._scheduler.deployment.get(
@@ -735,9 +736,7 @@ def _mount(self, user, volume, release, app_settings):
             err = f'(changed volume mount for {volume}: {e}'
             self.log(err, logging.ERROR)
             raise ServiceUnavailable(err) from e
-
-        msg = f'{user.username} changed volume mount for {volume}'
-        self.log(msg)
+        self.log(f'{user.username} changed volume mount for {volume}')
 
     def _scale(self, user, structure, release, app_settings):  # noqa
         """Scale containers up or down to match requested structure."""
diff --git a/rootfs/api/models/volume.py b/rootfs/api/models/volume.py
@@ -29,6 +29,8 @@ def save(self, *args, **kwargs):
         # Attach volume, updates k8s
         if self.type == "csi" and self.created == self.updated:
             self._create_pvc()
+        # check path
+        self._check_path()
         # Save to DB
         return super(Volume, self).save(*args, **kwargs)
 
@@ -97,10 +99,24 @@ def _format_size(size):
             size = size.upper() + "i"
         return size
 
+    def _check_path(self):
+        other_volumes = self.app.volume_set.exclude(name=self.name)
+        type_paths = {}  # {'type1':[path1,path2], tyep2:[path3,path4]}
+        for _ in other_volumes:
+            for k, v in _.path.items():
+                if k not in type_paths:
+                    type_paths[k] = [v]
+                else:
+                    type_paths[k].append(v)
+        repeat_path = [v for k, v in self.path.items() if v in type_paths.get(k, [])]
+        if repeat_path:
+            raise DryccException("path {} is used by another volume".
+                                 format(','.join(repeat_path)))
+
     def _create_pvc(self):
         try:
             self._scheduler.pvc.get(self.app.id, self.name)
-            err = "Volume {} already exists in this namespace".format(self.name)  # noqa
+            err = "Volume {} already exists in this namespace".format(self.name)
             self.log(err, logging.INFO)
             raise AlreadyExists(err)
         except KubeException as e:
diff --git a/rootfs/api/settings/celery.py b/rootfs/api/settings/celery.py
@@ -36,6 +36,11 @@ class Config(object):
             'exchange': 'controller.priority',
             'routing_key': 'controller.priority.high',
         },
+        'api.tasks.mount_app': {
+            'queue': 'low',
+            'exchange': 'controller.priority',
+            'routing_key': 'controller.priority.high',
+        },
         'api.tasks.restart_app': {
             'queue': 'high',
             'exchange': 'controller.priority',
@@ -46,6 +51,11 @@ class Config(object):
             'exchange': 'controller.priority',
             'routing_key': 'controller.priority.high',
         },
+        'api.tasks.downstream_model_owner': {
+            'queue': 'high',
+            'exchange': 'controller.priority',
+            'routing_key': 'controller.priority.high',
+        },
         'api.tasks.send_measurements': {
             'queue': 'middle',
             'exchange': 'controller.priority',
diff --git a/rootfs/api/tasks.py b/rootfs/api/tasks.py
@@ -94,16 +94,27 @@ def restart_app(app, **kwargs):
     retry_jitter=True,
     retry_kwargs={'max_retries': 3}
 )
-def mount_app(app, user, volume):
+def mount_app(app, user, volume, path):
     task_id = uuid.uuid4().hex
     signals.request_started.send(sender=task_id)
     try:
-        app.mount(user, volume)
+        # merge mount volume path and remove keys if a null value is provided
+        for key, value in path.items():
+            if value is None:
+                if key not in volume.path:
+                    continue
+                volume.path.pop(key)
+            else:
+                volume.path[key] = value
         volume.save()
+        structure = {}
+        for scale_type, replicas in app.structure.items():
+            if scale_type in path:
+                structure[scale_type] = replicas
+        app.mount(user, volume, structure)
     except Exception as e:
-        print(e)
         signals.got_request_exception.send(sender=task_id)
-        raise e
+        logger.exception(e)
     else:
         signals.request_finished.send(sender=task_id)
 
diff --git a/rootfs/api/tests/test_volume.py b/rootfs/api/tests/test_volume.py
@@ -186,8 +186,11 @@ def test_volume_mount(self, mock_requests):
         url = '/v2/apps/{app_id}/volumes/myvolume1/path'.format(app_id=app_id)
         mount_path = {"path": {"web": "/data/web1"}}
         response = self.client.patch(url, data=mount_path)
-        expected = response.data['path']
-        self.assertEqual(mount_path['path'], expected)
+        expected = response.data['path']  # old data
+        self.assertEqual({}, expected)
+        url = '/v2/apps/{app_id}/volumes/myvolume1'.format(app_id=app_id)
+        response = self.client.get(url)
+        self.assertEqual(response.data["path"], mount_path["path"])
 
     def test_volume_unmount(self, mock_requests):
         # create
@@ -210,7 +213,7 @@ def test_volume_unmount(self, mock_requests):
         mount_path = {"path": {"web": "/data/web1"}}
         response = self.client.patch(url, data=mount_path)
         expected = response.data['path']
-        self.assertEqual(mount_path['path'], expected)
+        self.assertEqual({}, expected)
         # check mount
         url = '/v2/apps/{app_id}/volumes/myvolume1'.format(app_id=app_id)
         response = self.client.get(url)
@@ -221,7 +224,11 @@ def test_volume_unmount(self, mock_requests):
         mount_path = {"path": {"web": None}}
         response = self.client.patch(url, data=mount_path)
         expected = response.data['path']
-        self.assertEqual({}, expected)
+        self.assertEqual(mount_path["path"], {'web': None})
+        # check mount
+        url = '/v2/apps/{app_id}/volumes/myvolume1'.format(app_id=app_id)
+        response = self.client.get(url)
+        self.assertEqual(response.data["path"], {})
 
     def build_deploy(self, app_id):
         # post a new build with procfile
diff --git a/rootfs/api/views.py b/rootfs/api/views.py
@@ -20,8 +20,7 @@
 
 from api import monitor, models, permissions, serializers, viewsets, authentication
 from api.tasks import scale_app, restart_app, mount_app, downstream_model_owner
-from api.exceptions import AlreadyExists, ServiceUnavailable, DryccException, \
-    UnprocessableEntity
+from api.exceptions import AlreadyExists, ServiceUnavailable, DryccException
 
 from django.views.decorators.cache import never_cache
 from django.contrib.auth import REDIRECT_FIELD_NAME
@@ -714,48 +713,22 @@ def destroy(self, request, **kwargs):
         return Response(status=status.HTTP_204_NO_CONTENT)
 
     def path(self, request, *args, **kwargs):
-        new_path = request.data.get('path')
-        if new_path is None:
+        path = request.data.get('path')
+        if path is None:
             raise DryccException("path is a required field")
         else:
-            new_path = serializers.VolumeSerializer().validate_path(new_path)
+            path = serializers.VolumeSerializer().validate_path(path)
         volume = self.get_object()
-        container_types = [_ for _ in new_path.keys()
+        container_types = [_ for _ in path.keys()
                            if _ not in volume.app.types]
         if container_types:
             raise DryccException("process type {} is not included in profile".
                                  format(','.join(container_types)))
-
-        if set(new_path.items()).issubset(set(volume.path.items())):
+        if set(path.items()).issubset(set(volume.path.items())):
             raise DryccException("mount path not changed")
 
-        other_volumes = self.get_app().volume_set.exclude(name=volume.name)
-        type_paths = {}  # {'type1':[path1,path2], tyep2:[path3,path4]}
-        for _ in other_volumes:
-            for k, v in _.path.items():
-                if k not in type_paths:
-                    type_paths[k] = [v]
-                else:
-                    type_paths[k].append(v)
-        repeat_path = [v for k, v in new_path.items() if v in type_paths.get(k, [])]  # noqa
-        if repeat_path:
-            raise DryccException("path {} is used by another volume".
-                                 format(','.join(repeat_path)))
-        path = volume.path
-        # merge mount path
-        # remove path keys if a null value is provided
-        for key, value in new_path.items():
-            if value is None:
-                # error if unsetting non-existing key
-                if key not in path:
-                    raise UnprocessableEntity(
-                        '{} does not exist under {}'.format(key, "volume"))  # noqa
-                path.pop(key)
-            else:
-                path[key] = value
         app = self.get_app()
-        volume.path = path  # volume save by task success
-        mount_app.delay(app, self.request.user, volume)
+        mount_app.delay(app, self.request.user, volume, path)
         serializer = self.get_serializer(volume, many=False)
         return Response(serializer.data)
 
