From 25a9550f1f5dae9b0ec0890583460d1e93334fb9 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E6=AE=B5=E6=B4=AA=E4=B9=89?=
 <duanhongyi@duanhongyideMac-mini.local>
Date: Mon, 23 Nov 2020 14:46:03 +0800
Subject: [PATCH 1/2] feat(tasks): use celery replace nsqd

---
 rootfs/Dockerfile.test            |  1 +
 rootfs/api/__init__.py            |  2 +
 rootfs/api/models/__init__.py     | 31 +++++++++++----
 rootfs/api/models/resource.py     | 47 ++--------------------
 rootfs/api/settings/celery.py     | 12 ++++++
 rootfs/api/settings/production.py | 21 ++++++++++
 rootfs/api/tasks.py               | 35 ++++++++++++++++
 rootfs/bin/boot                   |  5 ++-
 rootfs/bin/test-unit              | 16 +++++++-
 rootfs/requirements.txt           |  2 +
 rootfs/tasks/__init__.py          |  1 -
 rootfs/tasks/task.py              | 66 -------------------------------
 rootfs/tasks/tests/__init__.py    |  0
 rootfs/tasks/tests/test_task.py   | 42 --------------------
 14 files changed, 119 insertions(+), 162 deletions(-)
 create mode 100644 rootfs/api/settings/celery.py
 create mode 100644 rootfs/api/tasks.py
 delete mode 100644 rootfs/tasks/__init__.py
 delete mode 100644 rootfs/tasks/task.py
 delete mode 100644 rootfs/tasks/tests/__init__.py
 delete mode 100644 rootfs/tasks/tests/test_task.py

diff --git a/rootfs/Dockerfile.test b/rootfs/Dockerfile.test
index f62274866..ba17ce720 100644
--- a/rootfs/Dockerfile.test
+++ b/rootfs/Dockerfile.test
@@ -29,6 +29,7 @@ RUN echo https://dl-cdn.alpinelinux.org/alpine/edge/testing >>/etc/apk/repositor
     shadow \
     nsq \
     postgresql \
+    redis \
   && mkdir -p /run/postgresql $PGDATA \
   && chown -R postgres:postgres /run/postgresql $PGDATA \
   && apk del .build-deps \
diff --git a/rootfs/api/__init__.py b/rootfs/api/__init__.py
index b5f40a531..ce6e09f73 100644
--- a/rootfs/api/__init__.py
+++ b/rootfs/api/__init__.py
@@ -1,5 +1,7 @@
 """
 The **api** Django app presents a RESTful web API for interacting with the **drycc** system.
 """
+from .settings.celery import app as celery_app
 
 __version__ = '2.3.0'
+__all__ = ('celery_app',)
diff --git a/rootfs/api/models/__init__.py b/rootfs/api/models/__init__.py
index 180d3cb07..af1e984b5 100644
--- a/rootfs/api/models/__init__.py
+++ b/rootfs/api/models/__init__.py
@@ -11,25 +11,25 @@
 import re
 import urllib.parse
 import uuid
+import requests
 
+from datetime import timedelta
 from django.conf import settings
 from django.db import models
 from django.db.models.signals import post_delete, post_save
-from django.dispatch import receiver
+from django.utils.timezone import now
+from django.dispatch import receiver, Signal
 from rest_framework.exceptions import ValidationError
 from rest_framework.authtoken.models import Token
-import requests
 from requests_toolbelt import user_agent
-
-from api import __version__ as drycc_version
-from api.exceptions import DryccException, AlreadyExists, ServiceUnavailable, UnprocessableEntity  # noqa
-from api.utils import dict_merge
-from scheduler import KubeException
+from scheduler.exceptions import KubeException
+from .. import __version__ as drycc_version
+from ..exceptions import DryccException, AlreadyExists, ServiceUnavailable, UnprocessableEntity  # noqa
 
 
 logger = logging.getLogger(__name__)
-
 session = None
+resource_changed = Signal(providing_args=["resource_id"])
 
 
 def get_session():
@@ -153,6 +153,9 @@ class Meta:
 from .volume import Volume  # noqa
 from .resource import Resource  # noqa
 
+from ..tasks import retrieve_resource  # noqa
+from ..utils import dict_merge  # noqa
+
 # define update/delete callbacks for synchronizing
 # models with the configuration management backend
 
@@ -257,3 +260,15 @@ def _hook_release_created(**kwargs):
 def create_auth_token(sender, instance=None, created=False, **kwargs):
     if created:
         Token.objects.create(user=instance)
+
+
+@receiver(resource_changed)
+def resource_changed_handle(sender, **kwargs):
+    data = {
+        "task_id": uuid.uuid4().hex,
+        "resource_id": kwargs.get("resource_id"),
+    }
+    retrieve_resource.apply_async(
+        args=(data, ),
+        eta=now() + timedelta(seconds=30)
+    )
diff --git a/rootfs/api/models/resource.py b/rootfs/api/models/resource.py
index f87d18d05..7c5de6c85 100644
--- a/rootfs/api/models/resource.py
+++ b/rootfs/api/models/resource.py
@@ -1,14 +1,11 @@
 import logging
-import uuid
-import time
-from django.core import signals
 from django.conf import settings
 from django.db import models, transaction
 from jsonfield import JSONField
 from api.exceptions import DryccException, AlreadyExists, ServiceUnavailable
 from api.models import UuidAuditedModel, validate_label
 from scheduler import KubeException
-from tasks import task, apply_async
+from . import resource_changed
 
 logger = logging.getLogger(__name__)
 
@@ -58,11 +55,7 @@ def attach(self, *args, **kwargs):
                 self._scheduler.svcat.create_instance(
                     self.app.id, self.name, **kwargs
                 )
-                data = {
-                    "task_id": uuid.uuid4().hex,
-                    "resource_id": str(self.uuid),
-                }
-                apply_async(retrieve_task, delay=30000, args=(data, ))
+                resource_changed.send(sender=Resource, resource_id=str(self.uuid))
             except KubeException as e:
                 msg = 'There was a problem creating the resource ' \
                       '{} for {}'.format(self.name, self.app_id)
@@ -111,11 +104,7 @@ def bind(self, *args, **kwargs):
             try:
                 self._scheduler.svcat.create_binding(
                     self.app.id, self.name, **kwargs)
-                data = {
-                    "task_id": uuid.uuid4().hex,
-                    "resource_id": str(self.uuid),
-                }
-                apply_async(retrieve_task, delay=30000, args=(data, ))
+                resource_changed.send(sender=Resource, resource_id=str(self.uuid))
             except KubeException as e:
                 msg = 'There was a problem binding the resource ' \
                       '{} for {}'.format(self.name, self.app_id)
@@ -153,11 +142,7 @@ def attach_update(self, *args, **kwargs):
             self._scheduler.svcat.put_instance(
                 self.app.id, self.name, version, **kwargs
             )
-            data = {
-                "task_id": uuid.uuid4().hex,
-                "resource_id": str(self.uuid),
-            }
-            apply_async(retrieve_task, delay=30000, args=(data, ))
+            resource_changed.send(sender=Resource, resource_id=str(self.uuid))
         except KubeException as e:
             msg = 'There was a problem update the resource ' \
                   '{} for {}'.format(self.name, self.app_id)
@@ -214,27 +199,3 @@ def detach_resource(self, *args, **kwargs):
 
         if (self.status != "Ready") or (not self.binding):
             self.delete()
-
-
-@task
-def retrieve_task(data):
-    try:
-        signals.request_started.send(sender=data['task_id'])
-        try:
-            resource = Resource.objects.get(uuid=data['resource_id'])
-        except Resource.DoesNotExist:
-            logger.info("retrieve task not found resource: {}".format(data['resource_id']))  # noqa
-            return True
-        _ = resource.retrieve()
-        if _:
-            return True
-        else:
-            t = time.time() - resource.created.timestamp()
-            if t < 3600:
-                apply_async(retrieve_task, delay=30000, args=(data, ))
-            elif t < 3600 * 12:
-                apply_async(retrieve_task, delay=1800000, args=(data, ))
-            else:
-                resource.detach_resource()
-    finally:
-        signals.request_finished.send(sender=data['task_id'])
diff --git a/rootfs/api/settings/celery.py b/rootfs/api/settings/celery.py
new file mode 100644
index 000000000..41044df52
--- /dev/null
+++ b/rootfs/api/settings/celery.py
@@ -0,0 +1,12 @@
+import os
+from celery import Celery
+
+os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'api.settings.production')
+app = Celery('drycc')
+app.config_from_object('django.conf:settings', namespace='CELERY')
+app.conf.update(
+    task_routes={
+        'api.tasks.retrieve_resource': {'queue': 'priority.high'},
+    },
+)
+app.autodiscover_tasks()
diff --git a/rootfs/api/settings/production.py b/rootfs/api/settings/production.py
index b66c09b2c..e1f27f2ac 100644
--- a/rootfs/api/settings/production.py
+++ b/rootfs/api/settings/production.py
@@ -443,6 +443,18 @@
     }
 }
 
+REDIS_ADDRS = os.environ.get('REDIS_ADDRS', 'redis://127.0.0.1:6379/0').split(",")
+
+CACHES = {
+    "default": {
+        "BACKEND": "django_redis.cache.RedisCache",
+        "LOCATION": REDIS_ADDRS,
+        "OPTIONS": {
+            "CLIENT_CLASS": "django_redis.client.ShardClient",
+        }
+    }
+}
+
 APP_URL_REGEX = '[a-z0-9-]+'
 
 # LDAP settings taken from environment variables.
@@ -499,3 +511,12 @@
     AUTH_LDAP_MIRROR_GROUPS = True
     AUTH_LDAP_FIND_GROUP_PERMS = True
     AUTH_LDAP_CACHE_GROUPS = False
+
+# Celery Configuration Options
+CELERY_TIMEZONE = "Asia/Shanghai"
+CELERY_TASK_TRACK_STARTED = True
+CELERY_TASK_TIME_LIMIT = 30 * 60
+CELERY_BROKER_URL = REDIS_ADDRS[0]
+CELERY_RESULT_BACKEND = REDIS_ADDRS[0]
+CELERY_CACHE_BACKEND = 'django-cache'
+CELERY_DEFAULT_QUEUE = 'priority.middle'
diff --git a/rootfs/api/tasks.py b/rootfs/api/tasks.py
new file mode 100644
index 000000000..991d6fd4f
--- /dev/null
+++ b/rootfs/api/tasks.py
@@ -0,0 +1,35 @@
+# Create your tasks here
+import time
+import logging
+from django.core import signals
+from datetime import timedelta
+from django.utils.timezone import now
+from celery import shared_task
+
+from .models.resource import Resource
+
+
+logger = logging.getLogger(__name__)
+
+
+@shared_task
+def retrieve_resource(data):
+    signals.request_started.send(sender=data['task_id'])
+    try:
+        resource = Resource.objects.get(uuid=data['resource_id'])
+        if not resource.retrieve():
+            t = time.time() - resource.created.timestamp()
+            if t < 3600:
+                retrieve_resource.apply_async(
+                    args=(data, ),
+                    eta=now() + timedelta(seconds=30))
+            elif t < 3600 * 12:
+                retrieve_resource.apply_async(
+                    args=(data, ),
+                    eta=now() + timedelta(seconds=1800))
+            else:
+                resource.detach_resource()
+    except Resource.DoesNotExist:
+        logger.info("retrieve task not found resource: {}".format(data['resource_id']))  # noqa
+    finally:
+        signals.request_finished.send(sender=data['task_id'])
diff --git a/rootfs/bin/boot b/rootfs/bin/boot
index 9012b07d5..331843bd2 100755
--- a/rootfs/bin/boot
+++ b/rootfs/bin/boot
@@ -37,7 +37,10 @@ echo "Loading database information to Kubernetes in the background"
 echo "Log of the run can be found in /app/data/logs/load_db_state_to_k8s.log"
 # python -u avoids output buffering
 nohup python -u /app/manage.py load_db_state_to_k8s > /app/data/logs/load_db_state_to_k8s.log &
-
+# celery
+nohup su-exec nobody celery -A api worker -Q priority.low --loglevel=WARNING > /app/data/logs/celery.log 2>&1 &
+nohup su-exec nobody celery -A api worker -Q priority.middle --loglevel=WARNING > /app/data/logs/celery.log 2>&1 &
+nohup su-exec nobody celery -A api worker -Q priority.high --loglevel=WARNING > /app/data/logs/celery.log 2>&1 &
 # smart shutdown on SIGTERM (SIGINT is handled by gunicorn)
 function on_exit() {
 	GUNICORN_PID=$(cat /tmp/gunicorn.pid)
diff --git a/rootfs/bin/test-unit b/rootfs/bin/test-unit
index f06a4b242..512652bd3 100755
--- a/rootfs/bin/test-unit
+++ b/rootfs/bin/test-unit
@@ -12,8 +12,22 @@ function start_nsq() {
     cd -
 }
 
+function start_redis() {
+    cd /tmp
+    nohup redis-server > /var/log/redis.log 2>&1 &
+    cd -
+}
+
+function start_celery() {
+    cd /tmp
+    nohup su-exec nobody celery -A api worker -Q priority.low,priority.middle,priority.high --loglevel=info > /var/log/celery.log 2>&1 &
+    cd -
+}
+
 su-exec postgres pg_ctl -D "$PGDATA" start
 start_nsq
+start_redis
+start_celery
 python3 manage.py check
-coverage run manage.py test --settings=api.settings.testing --noinput tasks.tests api scheduler.tests
+coverage run manage.py test --settings=api.settings.testing --noinput api scheduler.tests
 coverage report -m
diff --git a/rootfs/requirements.txt b/rootfs/requirements.txt
index 00eeab0f3..d73f7d932 100644
--- a/rootfs/requirements.txt
+++ b/rootfs/requirements.txt
@@ -21,3 +21,5 @@ pyOpenSSL==19.1.0
 pytz==2020.1
 requests==2.24.0
 requests-toolbelt==0.9.1
+celery==5.0.2
+django_redis==4.12.1
diff --git a/rootfs/tasks/__init__.py b/rootfs/tasks/__init__.py
deleted file mode 100644
index 98745cf19..000000000
--- a/rootfs/tasks/__init__.py
+++ /dev/null
@@ -1 +0,0 @@
-from .task import task, apply_async  # noqa
diff --git a/rootfs/tasks/task.py b/rootfs/tasks/task.py
deleted file mode 100644
index 3100ee52d..000000000
--- a/rootfs/tasks/task.py
+++ /dev/null
@@ -1,66 +0,0 @@
-import os
-import json
-import logging
-import asyncio
-import concurrent
-import nsq
-from functools import wraps, partial
-
-logger = logging.getLogger(__name__)
-
-
-def _message_handler(message):
-
-    async def _async_task(loop):
-        data = json.loads(message.body)
-        function = TASKS[data["target_id"]]
-        _sync_task = partial(function, *data["args"], **data["kwargs"])
-        with concurrent.futures.ThreadPoolExecutor() as executor:
-            await loop.run_in_executor(executor, _sync_task)
-            message.finish()
-
-    message.enable_async()
-    asyncio.set_event_loop(asyncio.new_event_loop())
-    asyncio.ensure_future(_async_task(asyncio.get_event_loop()))
-
-
-TASKS = {}
-NSQD_ADDRS = os.environ.get('DRYCC_NSQD_ADDRS', '127.0.0.1:4150').split(",")
-NSQ_TOPIC = os.environ.get('DRYCC_NSQ_TASKS_TOPIC', 'drycc-tasks-topic')
-NSQ_CHANNEL = os.environ.get('DRYCC_NSQ_TASKS_CHANNEL', 'drycc-tasks-channel')
-NSQD_WRITER = nsq.Writer(NSQD_ADDRS)
-NSQD_READER = nsq.Reader(
-    message_handler=_message_handler,
-    nsqd_tcp_addresses=NSQD_ADDRS,
-    topic=NSQ_TOPIC,
-    channel=NSQ_CHANNEL,
-    lookupd_poll_interval=15,
-)
-
-
-def task(func):
-    target_id = "%s.%s" % (func.__module__, func.__name__)
-    TASKS[target_id] = func
-
-    @wraps(func)
-    def register_task(*args, **kwargs):
-        return func(*args, **kwargs)
-
-    return register_task
-
-
-def apply_async(target, delay=0, callback=None, args=(), kwargs=None):
-    target_id = "%s.%s" % (target.__module__, target.__name__)
-    if target_id not in TASKS:
-        raise NotImplementedError("This task is not registered.")
-    message = json.dumps({
-        "target_id": target_id,
-        "args": args,
-        "kwargs": {} if kwargs is None else kwargs
-    }).encode("utf-8")
-
-    asyncio.set_event_loop(asyncio.new_event_loop())
-    if delay <= 0:
-        NSQD_WRITER.pub(NSQ_TOPIC, message, callback=callback)
-    else:
-        NSQD_WRITER.dpub(NSQ_TOPIC, delay, message, callback=callback)
diff --git a/rootfs/tasks/tests/__init__.py b/rootfs/tasks/tests/__init__.py
deleted file mode 100644
index e69de29bb..000000000
diff --git a/rootfs/tasks/tests/test_task.py b/rootfs/tasks/tests/test_task.py
deleted file mode 100644
index 9da2f29d9..000000000
--- a/rootfs/tasks/tests/test_task.py
+++ /dev/null
@@ -1,42 +0,0 @@
-import unittest
-import time
-import datetime
-import tornado.ioloop
-from tasks.task import TASKS
-from tasks import task, apply_async
-
-
-class TestUtils(unittest.TestCase):
-
-    def test_task(self):
-        @task
-        def t():
-            pass
-        target_id = "%s.%s" % (t.__module__, t.__name__)
-        self.assertEqual(target_id in TASKS, True)
-
-    def test_apply_async(self):
-
-        @task
-        def t1(name, value):
-            self.assertEqual(name == "hi", True)
-            self.assertEqual(value == "word", True)
-
-        @task
-        def t2(t):
-            self.assertEqual(time.time() - t > 3, True)
-
-        def callback(_, msg):
-            self.assertEqual(msg == b'OK', True)
-
-        def run_test_task():
-            loop = tornado.ioloop.IOLoop.current()
-            try:
-                apply_async(t1, callback=callback, args=("hi", "word"))
-                apply_async(t2, callback=callback, delay=3000, args=(time.time(),))
-            finally:
-                _loop.add_timeout(datetime.timedelta(seconds=9), loop.stop)
-
-        _loop = tornado.ioloop.IOLoop.current()
-        _loop.add_timeout(datetime.timedelta(seconds=3), run_test_task)
-        _loop.start()

From c90448f038920e7b98034dbf8ad3d880a7b61925 Mon Sep 17 00:00:00 2001
From: lijianguo <lijianguo@zhumengyuan.com>
Date: Tue, 24 Nov 2020 15:46:08 +0800
Subject: [PATCH 2/2] chore(controller):modify redis config

---
 .../templates/controller-deployment.yaml          | 15 +++++++++++++++
 rootfs/api/settings/production.py                 | 13 +++++++++----
 rootfs/bin/boot                                   |  8 ++++----
 rootfs/drycc/gunicorn/wsgi.py                     |  4 ----
 rootfs/requirements.txt                           |  1 -
 5 files changed, 28 insertions(+), 13 deletions(-)
 delete mode 100644 rootfs/drycc/gunicorn/wsgi.py

diff --git a/charts/controller/templates/controller-deployment.yaml b/charts/controller/templates/controller-deployment.yaml
index 7f280064a..0698b5dfd 100644
--- a/charts/controller/templates/controller-deployment.yaml
+++ b/charts/controller/templates/controller-deployment.yaml
@@ -136,6 +136,21 @@ spec:
                 fieldPath: metadata.namespace
           - name: DRYCC_NSQD_ADDRS
             value: "{{range $i := until $nsqdNodeCount}}drycc-nsqd-{{$i}}.drycc-nsqd.{{$.Release.Namespace}}.svc.{{$.Values.global.cluster_domain}}:{{$.Values.nsqd.tcp_port}}{{if lt (add 1 $i) $nsqdNodeCount}},{{end}}{{end}}"
+{{ if eq .Values.global.redis_location "on-cluster"}}
+        - name: DRYCC_REDIS_ADDRS
+          value: "{{range $i := until $redisNodeCount}}drycc-redis-{{$i}}.drycc-redis.{{$.Release.Namespace}}.svc.{{$.Values.global.cluster_domain}}:{{$.Values.redis.port}}{{if lt (add 1 $i) $redisNodeCount}},{{end}}{{end}}"
+{{- else if eq .Values.global.redis_location "off-cluster" }}
+        - name: DRYCC_REDIS_ADDRS
+          valueFrom:
+            secretKeyRef:
+              name: redis-creds
+              key: addrs
+{{- end }}
+        - name: DRYCC_REDIS_PASSWORD
+          valueFrom:
+            secretKeyRef:
+              name: redis-creds
+              key: password
           {{- range $key, $value := .Values.environment }}
           - name: {{ $key }}
             value: {{ $value | quote }}
diff --git a/rootfs/api/settings/production.py b/rootfs/api/settings/production.py
index e1f27f2ac..63e2ea020 100644
--- a/rootfs/api/settings/production.py
+++ b/rootfs/api/settings/production.py
@@ -443,12 +443,14 @@
     }
 }
 
-REDIS_ADDRS = os.environ.get('REDIS_ADDRS', 'redis://127.0.0.1:6379/0').split(",")
+DRYCC_REDIS_ADDRS = os.environ.get('DRYCC_REDIS_ADDRS', '127.0.0.1:6379').split(",")
+DRYCC_REDIS_PASSWORD = os.environ.get('DRYCC_REDIS_PASSWORD', '')
 
 CACHES = {
     "default": {
         "BACKEND": "django_redis.cache.RedisCache",
-        "LOCATION": REDIS_ADDRS,
+        "LOCATION": ['redis://:{}@{}'.format(DRYCC_REDIS_PASSWORD, DRYCC_REDIS_ADDR) \
+                     for DRYCC_REDIS_ADDR in DRYCC_REDIS_ADDRS],  # noqa
         "OPTIONS": {
             "CLIENT_CLASS": "django_redis.client.ShardClient",
         }
@@ -514,9 +516,12 @@
 
 # Celery Configuration Options
 CELERY_TIMEZONE = "Asia/Shanghai"
+CELERY_ENABLE_UTC = True
 CELERY_TASK_TRACK_STARTED = True
 CELERY_TASK_TIME_LIMIT = 30 * 60
-CELERY_BROKER_URL = REDIS_ADDRS[0]
-CELERY_RESULT_BACKEND = REDIS_ADDRS[0]
+CELERYD_MAX_TASKS_PER_CHILD = 200
+CELERY_TASK_RESULT_EXPIRES = 24 * 60 * 60
+CELERY_BROKER_URL ='redis://:{}@{}'.format(DRYCC_REDIS_PASSWORD, DRYCC_REDIS_ADDRS[0])  # noqa
+CELERY_RESULT_BACKEND = 'redis://:{}@{}'.format(DRYCC_REDIS_PASSWORD, DRYCC_REDIS_ADDRS[0])  # noqa
 CELERY_CACHE_BACKEND = 'django-cache'
 CELERY_DEFAULT_QUEUE = 'priority.middle'
diff --git a/rootfs/bin/boot b/rootfs/bin/boot
index 331843bd2..6bb6abdab 100755
--- a/rootfs/bin/boot
+++ b/rootfs/bin/boot
@@ -30,7 +30,7 @@ python /app/manage.py migrate --noinput
 # spawn a gunicorn server in the background
 echo ""
 echo "Starting up Gunicorn"
-gunicorn -c /app/drycc/gunicorn/config.py drycc.gunicorn.wsgi &
+gunicorn -c /app/drycc/gunicorn/config.py api.wsgi &
 
 echo ""
 echo "Loading database information to Kubernetes in the background"
@@ -38,9 +38,9 @@ echo "Log of the run can be found in /app/data/logs/load_db_state_to_k8s.log"
 # python -u avoids output buffering
 nohup python -u /app/manage.py load_db_state_to_k8s > /app/data/logs/load_db_state_to_k8s.log &
 # celery
-nohup su-exec nobody celery -A api worker -Q priority.low --loglevel=WARNING > /app/data/logs/celery.log 2>&1 &
-nohup su-exec nobody celery -A api worker -Q priority.middle --loglevel=WARNING > /app/data/logs/celery.log 2>&1 &
-nohup su-exec nobody celery -A api worker -Q priority.high --loglevel=WARNING > /app/data/logs/celery.log 2>&1 &
+nohup su-exec nobody celery -A api worker -Q priority.low --autoscale=8,1 --loglevel=WARNING > /app/data/logs/celery.log 2>&1 &
+nohup su-exec nobody celery -A api worker -Q priority.middle --autoscale=16,1 --loglevel=WARNING > /app/data/logs/celery.log 2>&1 &
+nohup su-exec nobody celery -A api worker -Q priority.high --autoscale=32,1 --loglevel=WARNING > /app/data/logs/celery.log 2>&1 &
 # smart shutdown on SIGTERM (SIGINT is handled by gunicorn)
 function on_exit() {
 	GUNICORN_PID=$(cat /tmp/gunicorn.pid)
diff --git a/rootfs/drycc/gunicorn/wsgi.py b/rootfs/drycc/gunicorn/wsgi.py
deleted file mode 100644
index 469049d14..000000000
--- a/rootfs/drycc/gunicorn/wsgi.py
+++ /dev/null
@@ -1,4 +0,0 @@
-from tornado.wsgi import WSGIContainer
-from api.wsgi import application as handler
-
-application = WSGIContainer(handler)
diff --git a/rootfs/requirements.txt b/rootfs/requirements.txt
index d73f7d932..1843d94fd 100644
--- a/rootfs/requirements.txt
+++ b/rootfs/requirements.txt
@@ -15,7 +15,6 @@ morph==0.1.4
 ndg-httpsclient==0.5.1
 packaging==20.4
 pyasn1==0.4.8
-pynsq==0.9.0
 psycopg2-binary==2.8.5
 pyOpenSSL==19.1.0
 pytz==2020.1
