Add multiple queue: feeds and default

.pre-commit-config.yaml

@@ -34,3 +34,8 @@ repos:
     rev: v1.1.390
     hooks:
       - id: pyright
+
+  - repo: https://github.com/gruntwork-io/pre-commit
+    rev: v0.1.15
+    hooks:
+      - id: helmlint

deploy/run_worker.sh

@@ -6,6 +6,4 @@ ROOT_DIR=$(dirname "$BASE_DIR")
 
 cd $ROOT_DIR
 
-# concurrency: Number of workers
-# max-tasks-per-child: Max number of tasks a worker can run before it is terminated
-celery -A main worker -l info --concurrency 4 --max-tasks-per-child 10
+celery -A main worker -l info $CELERY_ARGS

helm/linter_values.yaml

@@ -0,0 +1 @@
+./values-test.yaml

helm/templates/worker-beat/deployment.yaml

@@ -34,12 +34,7 @@ spec:
           command: ["/code/deploy/run_worker_beat.sh"]
           # TODO: livenessProbe
           resources:
-            requests:
-              cpu: {{ .Values.worker.beat.resources.requests.cpu }}
-              memory: {{ .Values.worker.beat.resources.requests.memory }}
-            limits:
-              cpu: {{ .Values.worker.beat.resources.limits.cpu }}
-              memory: {{ .Values.worker.beat.resources.limits.memory }}
+            {{- toYaml .Values.worker.beat.resources | nindent 12 }}
           env:
             - name: DJANGO_APP_TYPE
               value: worker

helm/templates/worker/deployment.yaml

@@ -1,52 +1,61 @@
 {{- if .Values.worker.enabled }}
 
+{{- range $queue_name, $config := .Values.worker.queues }}
+
 apiVersion: apps/v1
 kind: Deployment
 metadata:
-  name: {{ template "ifrcgo-alert-hub.fullname" . }}-worker
+  name: {{ template "ifrcgo-alert-hub.fullname" $ }}-worker-{{ $queue_name }}
   labels:
-    app: {{ include "ifrcgo-alert-hub.fullname" . }}
+    app: {{ include "ifrcgo-alert-hub.fullname" $ }}
     component: worker
-    environment: {{ .Values.environment }}
-    release: {{ .Release.Name }}
+    queue: {{ $queue_name }}
+    environment: {{ $.Values.environment }}
+    release: {{ $.Release.Name }}
   annotations:
     reloader.stakater.com/auto: "true"
 spec:
-  replicas: {{ .Values.worker.replicaCount }}
+  replicas: {{ $config.replicaCount }}
   selector:
     matchLabels:
-      app: {{ include "ifrcgo-alert-hub.fullname" . }}
+      app: {{ include "ifrcgo-alert-hub.fullname" $ }}
       component: worker
+      queue: {{ $queue_name }}
   template:
     metadata:
-      {{- if not .Values.azure.aksSecretsProviderAvailable }}
+      {{- if not $.Values.azure.aksSecretsProviderAvailable }}
       annotations:
-        checksum/secret: {{ include (print .Template.BasePath "/config/secret.yaml") . | sha256sum }}
-        checksum/configmap: {{ include (print .Template.BasePath "/config/configmap.yaml") . | sha256sum }}
+        checksum/secret: {{ include (print $.Template.BasePath "/config/secret.yaml") $ | sha256sum }}
+        checksum/configmap: {{ include (print $.Template.BasePath "/config/configmap.yaml") $ | sha256sum }}
       {{- end }}
       labels:
-        app: {{ include "ifrcgo-alert-hub.fullname" . }}
+        app: {{ include "ifrcgo-alert-hub.fullname" $ }}
         component: worker
+        queue: {{ $queue_name }}
     spec:
       containers:
         - name: worker
-          image: "{{ .Values.image.name }}:{{ .Values.image.tag }}"
+          image: "{{ $.Values.image.name }}:{{ $.Values.image.tag }}"
           command: ["/code/deploy/run_worker.sh"]
           # TODO: livenessProbe
           resources:
-            {{- toYaml .Values.worker.resources | nindent 12 }}
+            {{- toYaml $config.resources | nindent 12 }}
           env:
             - name: DJANGO_APP_TYPE
               value: worker
+            - name: CELERY_ARGS
+              value: {{ required "worker.queues.*.celeryArgs" $config.celeryArgs | quote }}
           envFrom:
             - secretRef:
-                name: {{ template "ifrcgo-alert-hub.secretname" . }}
+                name: {{ template "ifrcgo-alert-hub.secretname" $ }}
             - configMapRef:
-                name: {{ template "ifrcgo-alert-hub.fullname" . }}-api-configmap
+                name: {{ template "ifrcgo-alert-hub.fullname" $ }}-api-configmap
 
-      {{- with .Values.imagePullSecrets }}
+      {{- with $.Values.imagePullSecrets }}
       imagePullSecrets:
         {{- toYaml . | nindent 8 }}
       {{- end }}
 
 {{- end }}
+
+{{- end }}

helm/values-test.yaml

@@ -53,6 +53,8 @@ secretsAdditional:
   MAGIC_KEY: to-much-fun
 
 argoHooks:
+  db-migrate:
+    preserveHistory: false
   custom-python-script-01:
     enabled: true
     hook: PostSync

helm/values.yaml

@@ -53,14 +53,6 @@ api:
 
 worker:
   enabled: true
-  replicaCount: 1
-  resources:
-    requests:
-      cpu: "1"
-      memory: 1Gi
-    limits:
-      cpu: "1"
-      memory: 2Gi
   beat:
     resources:
       requests:
@@ -69,6 +61,30 @@ worker:
       limits:
         cpu: "1"
         memory: 1Gi
+  queues:
+    # NOTE: Make sure keys are lowercase
+    default:
+      enabled: true
+      replicaCount: 1
+      celeryArgs: "-Q default --concurrency 4 --max-tasks-per-child 10"
+      resources:
+        requests:
+          cpu: "1"
+          memory: 1Gi
+        limits:
+          cpu: "1"
+          memory: 2Gi
+    feeds:
+      enabled: true
+      replicaCount: 2
+      celeryArgs: "-Q feeds --concurrency 4 --max-tasks-per-child 10 --soft-time-limit 300 --time-limit 500"
+      resources:
+        requests:
+          cpu: "2"
+          memory: 1Gi
+        limits:
+          cpu: "1"
+          memory: 2Gi
 
 argoHooks:
   # NOTE: Make sure keys are lowercase

main/celery.py

@@ -40,7 +40,7 @@ def on_configure(self):  # type: ignore[reportIncompatibleVariableOverride]
     },
     f'{INTERNAL_CELERY_TASK_NAME_PREFIX}process_pending_subscription_alerts': {
         'task': 'apps.subscription.tasks.process_pending_subscription_alerts',
-        'schedule': timedelta(minutes=30),  # TODO: Lower this?
+        'schedule': timedelta(minutes=10),
         'options': {'queue': 'default'},
     },
     f'{INTERNAL_CELERY_TASK_NAME_PREFIX}send_daily_user_alert_subscriptions_email': {
@@ -65,6 +65,18 @@ def on_configure(self):  # type: ignore[reportIncompatibleVariableOverride]
     },
 }
 
+
+class CeleryQueue:
+    # NOTE: Make sure all queue names are lowercase (They are in k8s)
+    default = Queue('default')
+    feeds = Queue('feeds')
+
+    ALL_QUEUE = (
+        default,
+        feeds,
+    )
+
+
 # Using a string here means the worker doesn't have to serialize
 # the configuration object to child processes.
 # - namespace='CELERY' means all celery-related configuration keys
@@ -74,24 +86,13 @@ def on_configure(self):  # type: ignore[reportIncompatibleVariableOverride]
 # Load task modules from all registered Django apps.
 app.autodiscover_tasks()
 
-app.conf.task_default_queue = 'default'
-app.conf.task_queues = (Queue('default', routing_key='poll.#', exchange='poll'),)
-app.conf.task_default_exchange = 'poll'
-app.conf.task_default_exchange_type = 'topic'
-app.conf.task_default_routing_key = 'poll.default'
 app.conf.result_expires = settings.CELERY_TASK_EXPIRE
+app.conf.task_default_queue = CeleryQueue.default.name
 
-task_routes = {
-    'apps.cap_feed.tasks.poll_feed': {
-        'queue': 'default',
-        'routing_key': 'poll.#',
-        'exchange': 'poll',
-    },
-    'apps.cap_feed.tasks.tag_expired_alerts': {
-        'queue': 'default',
-        'routing_key': 'poll.#',
-        'exchange': 'poll',
-    },
+app.conf.task_queues = CeleryQueue.ALL_QUEUE
+
+app.conf.task_routes = {
+    'apps.cap_feed.tasks.poll_feed': {'queue': CeleryQueue.feeds},
 }