feat(tasks): add distributed async task

Author: duanhongyi

charts/controller/templates/controller-deployment.yaml

@@ -1,3 +1,4 @@
+{{ $nsqdNodeCount := .Values.nsqd.replicas | int }}
 apiVersion: apps/v1
 kind: Deployment
 metadata:
@@ -53,86 +54,88 @@ spec:
 {{- end }}
 {{- end }}
           env:
-            - name: REGISTRATION_MODE
-              value: {{ .Values.registration_mode }}
-            # NOTE(bacongobbler): use drycc/registry_proxy to work around Docker --insecure-registry requirements
-            - name: "DRYCC_REGISTRY_PROXY_HOST"
-              value: "127.0.0.1"
-            # Environmental variable value for $INGRESS_CLASS
-            - name: "DRYCC_INGRESS_CLASS"
-              value: "{{ .Values.global.ingress_class }}"
-            - name: "DRYCC_PLATFORM_DOMAIN"
-              value: "{{ .Values.global.platform_domain }}"
-            - name: "K8S_API_VERIFY_TLS"
-              value: "{{ .Values.k8s_api_verify_tls }}"
-            - name: "DRYCC_REGISTRY_PROXY_PORT"
-              value: "{{ .Values.global.registry_proxy_port }}"
-            - name: "APP_STORAGE"
-              value: "{{ .Values.global.storage}}"
-            - name: "DRYCC_REGISTRY_LOCATION"
-              value: "{{ .Values.global.registry_location }}"
-            - name: "DRYCC_REGISTRY_SECRET_PREFIX"
-              value: "{{ .Values.global.registry_secret_prefix }}"
-            - name: "IMAGE_PULL_POLICY"
-              value: "{{ .Values.app_pull_policy }}"
+          - name: REGISTRATION_MODE
+            value: {{ .Values.registration_mode }}
+          # NOTE(bacongobbler): use drycc/registry_proxy to work around Docker --insecure-registry requirements
+          - name: "DRYCC_REGISTRY_PROXY_HOST"
+            value: "127.0.0.1"
+          # Environmental variable value for $INGRESS_CLASS
+          - name: "DRYCC_INGRESS_CLASS"
+            value: "{{ .Values.global.ingress_class }}"
+          - name: "DRYCC_PLATFORM_DOMAIN"
+            value: "{{ .Values.global.platform_domain }}"
+          - name: "K8S_API_VERIFY_TLS"
+            value: "{{ .Values.k8s_api_verify_tls }}"
+          - name: "DRYCC_REGISTRY_PROXY_PORT"
+            value: "{{ .Values.global.registry_proxy_port }}"
+          - name: "APP_STORAGE"
+            value: "{{ .Values.global.storage}}"
+          - name: "DRYCC_REGISTRY_LOCATION"
+            value: "{{ .Values.global.registry_location }}"
+          - name: "DRYCC_REGISTRY_SECRET_PREFIX"
+            value: "{{ .Values.global.registry_secret_prefix }}"
+          - name: "IMAGE_PULL_POLICY"
+            value: "{{ .Values.app_pull_policy }}"
 {{- if (.Values.app_storage_class) }}
-            - name: "DRYCC_APP_KUBERNETES_STORAGE_CLASS"
-              value: "{{ .Values.app_storage_class }}"
+          - name: "DRYCC_APP_KUBERNETES_STORAGE_CLASS"
+            value: "{{ .Values.app_storage_class }}"
 {{- end }}
-            - name: "TZ"
-              value: {{ .Values.time_zone | default "UTC" | quote }}
+          - name: "TZ"
+            value: {{ .Values.time_zone | default "UTC" | quote }}
 {{- if (.Values.deploy_hook_urls) }}
-            - name: DRYCC_DEPLOY_HOOK_URLS
-              value: "{{ .Values.deploy_hook_urls }}"
-            - name: DRYCC_DEPLOY_HOOK_SECRET_KEY
-              valueFrom:
-                secretKeyRef:
-                  name: deploy-hook-key
-                  key: secret-key
+          - name: DRYCC_DEPLOY_HOOK_URLS
+            value: "{{ .Values.deploy_hook_urls }}"
+          - name: DRYCC_DEPLOY_HOOK_SECRET_KEY
+            valueFrom:
+              secretKeyRef:
+                name: deploy-hook-key
+                key: secret-key
 {{- end }}
-            - name: DRYCC_SECRET_KEY
-              valueFrom:
-                secretKeyRef:
-                  name: django-secret-key
-                  key: secret-key
-            - name: DRYCC_BUILDER_KEY
-              valueFrom:
-                secretKeyRef:
-                  name: builder-key-auth
-                  key: builder-key
+          - name: DRYCC_SECRET_KEY
+            valueFrom:
+              secretKeyRef:
+                name: django-secret-key
+                key: secret-key
+          - name: DRYCC_BUILDER_KEY
+            valueFrom:
+              secretKeyRef:
+                name: builder-key-auth
+                key: builder-key
 {{- if eq .Values.global.database_location "off-cluster" }}
-            - name: DRYCC_DATABASE_NAME
-              valueFrom:
-                secretKeyRef:
-                  name: database-creds
-                  key: name
-            - name: DRYCC_DATABASE_SERVICE_HOST
-              valueFrom:
-                secretKeyRef:
-                  name: database-creds
-                  key: host
-            - name: DRYCC_DATABASE_SERVICE_PORT
-              valueFrom:
-                secretKeyRef:
-                  name: database-creds
-                  key: port
+          - name: DRYCC_DATABASE_NAME
+            valueFrom:
+              secretKeyRef:
+                name: database-creds
+                key: name
+          - name: DRYCC_DATABASE_SERVICE_HOST
+            valueFrom:
+              secretKeyRef:
+                name: database-creds
+                key: host
+          - name: DRYCC_DATABASE_SERVICE_PORT
+            valueFrom:
+              secretKeyRef:
+                name: database-creds
+                key: port
 {{- end }}
-            - name: DRYCC_DATABASE_USER
-              valueFrom:
-                secretKeyRef:
-                  name: database-creds
-                  key: user
-            - name: DRYCC_DATABASE_PASSWORD
-              valueFrom:
-                secretKeyRef:
-                  name: database-creds
-                  key: password
-            - name: RESERVED_NAMES
-              value: "drycc, drycc-builder, grafana"
-            - name: WORKFLOW_NAMESPACE
-              valueFrom:
-                fieldRef:
-                  fieldPath: metadata.namespace
+          - name: DRYCC_DATABASE_USER
+            valueFrom:
+              secretKeyRef:
+                name: database-creds
+                key: user
+          - name: DRYCC_DATABASE_PASSWORD
+            valueFrom:
+              secretKeyRef:
+                name: database-creds
+                key: password
+          - name: RESERVED_NAMES
+            value: "drycc, drycc-builder, drycc-monitor-grafana"
+          - name: WORKFLOW_NAMESPACE
+            valueFrom:
+              fieldRef:
+                fieldPath: metadata.namespace
+          - name: DRYCC_NSQD_ADDRS
+            value: "{{range $i := until $nsqdNodeCount}}drycc-nsqd-{{$i}}.drycc-nsqd.{{ $.Release.Namespace }}.svc.cluster.local:{{$.Values.nsqd.tcp_port}}{{if lt (add 1 $i) $nsqdNodeCount}},{{end}}{{end}}"
           volumeMounts:
             - mountPath: /etc/slugrunner
               name: slugrunner-config

charts/controller/values.yaml

@@ -19,6 +19,11 @@ k8s_api_verify_tls: "true"
 # Set storageClassName, It is used for application mount.
 app_storage_class: ""
 
+nsqd:
+  replicas: 1
+  tcp_port: 4150
+  http_port: 4151
+
 global:
   # Admin email, used for each component to send email to administrator
   email: "drycc@drycc.cc"

rootfs/Dockerfile.test

@@ -4,7 +4,8 @@ COPY requirements.txt /app/requirements.txt
 COPY dev_requirements.txt /app/dev_requirements.txt
 
 ENV PGDATA /var/lib/postgresql/12
-RUN apk add --update --virtual .build-deps \
+RUN echo https://dl-cdn.alpinelinux.org/alpine/edge/testing >>/etc/apk/repositories \
+  && apk add --update --virtual .build-deps \
     postgresql-dev \
     gcc \
     libffi-dev \
@@ -26,7 +27,8 @@ RUN apk add --update --virtual .build-deps \
     su-exec \
     bash \
     shadow \
-    postgresql==12.3-r2 \
+    nsq \
+    postgresql \
   && mkdir -p /run/postgresql $PGDATA \
   && chown -R postgres:postgres /run/postgresql $PGDATA \
   && apk del .build-deps \
@@ -37,4 +39,3 @@ COPY . /app
 WORKDIR /app
 CMD ["/app/bin/boot"]
 EXPOSE 8000
-

rootfs/api/tests/test_build.py

@@ -349,7 +349,7 @@ def test_build_no_remove_process(self, mock_requests):
         self.assertEqual(container['type'], 'worker')
         self.assertEqual(container['release'], 'v2')
         # pod name is auto generated so use regex
-        self.assertRegex(container['name'], app_id + '-worker-[0-9]{8,10}-[a-z0-9]{5}')
+        self.assertRegex(container['name'], app_id + '-worker-[0-9]{7,10}-[a-z0-9]{5}')
 
         # do another deploy for this time forget Procfile
         url = "/v2/apps/{app_id}/builds".format(**locals())

rootfs/bin/test-unit

@@ -6,7 +6,14 @@
 # fail hard and fast even on pipelines
 set -eou pipefail
 
+function start_nsq() {
+    cd /tmp
+    nohup nsqd > /var/log/nsqd.log 2>&1 &
+    cd -
+}
+
 su-exec postgres pg_ctl -D "$PGDATA" start
+start_nsq
 python3 manage.py check
-coverage run manage.py test --settings=api.settings.testing --noinput api scheduler.tests
+coverage run manage.py test --settings=api.settings.testing --noinput tasks.tests api scheduler.tests
 coverage report -m

rootfs/drycc/gunicorn/config.py

@@ -8,6 +8,9 @@
 bind = '0.0.0.0'
 workers = int(os.environ.get('GUNICORN_WORKERS', cpu_count() * 4 + 1))
 
+# Use tornado, otherwise NSQ needs to be started separately
+worker_class = 'tornado'
+
 pythonpath = dirname(dirname(dirname(realpath(__file__))))
 timeout = 1200
 pidfile = '/tmp/gunicorn.pid'

rootfs/requirements.txt

@@ -15,6 +15,7 @@ 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
 pyldap==3.0.0.post1
 pyOpenSSL==19.1.0

rootfs/tasks/__init__.py

@@ -0,0 +1 @@
+from .task import task, apply_async  # noqa

rootfs/tasks/task.py

@@ -0,0 +1,52 @@
+import os
+import json
+import threading
+import nsq
+from functools import wraps
+
+
+def _message_handler(message):
+    data = json.loads(message.body)
+    method = TASKS[data["target_id"]]
+    threading.Thread(
+        target=method, args=data["args"], kwargs=data["kwargs"]).start()
+    return True
+
+
+TASKS = {}
+NSQD_ADDRS = os.environ.get('DRYCC_NSQD_ADDRS', '127.0.0.1:4150').split(",")
+NSQ_TOPIC = os.environ.get('DRYCC_NSQ_TASKS_TOPIC', 'tasks:topic')
+NSQ_CHANNEL = os.environ.get('DRYCC_NSQ_TASKS_CHANNEL', '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):
+    target_id = "%s.%s" % (target.__module__, target.__name__)
+    if target_id not in TASKS:
+        raise NotImplemented("This task is not registered.")
+    message = json.dumps({
+        "target_id": target_id,
+        "args": args,
+        "kwargs": kwargs
+    }).encode("utf-8")
+    if delay <= 0:
+        NSQD_WRITER.pub(NSQ_TOPIC, message, callback=callback)
+    else:
+        NSQD_WRITER.dpub(NSQ_TOPIC, delay, message, callback=callback)