feat(tasks): use celery replace nsqd

Author: 段洪义

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 \

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',)

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)
+    )

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'])

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()

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'

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'])

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)

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

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