feat: allow to setup multiple celery queues

overhangio · Oct 4, 2024 · f257d3f · f257d3f
1 parent cf8925c
commit f257d3f
Show file tree

Hide file tree

Showing 5 changed files with 102 additions and 90 deletions.
diff --git a/tutor/hooks/catalog.py b/tutor/hooks/catalog.py
@@ -514,6 +514,9 @@ def your_filter_callback(some_data):
     #: :param str file_path: The path to the file being checked.
     IS_FILE_RENDERED: Filter[bool, [str]] = Filter()
 
+    #: List of workers
+    CELERY_WORKERS_CONFIG: Filter[dict[str, dict[str, Any]], []] = Filter()
+
 
 class Contexts:
     """

diff --git a/tutor/plugins/openedx.py b/tutor/plugins/openedx.py
@@ -142,3 +142,62 @@ def is_directory_mounted(image_name: str, dirname: str) -> bool:
 hooks.Filters.ENV_TEMPLATE_VARIABLES.add_item(
     ("iter_mounted_directories", iter_mounted_directories)
 )
+
+
+@hooks.Filters.CELERY_WORKERS_CONFIG.add(priority=hooks.priorities.HIGH)
+def _add_default_celery_queues(worker_configs):
+    worker_configs = {
+        "lms": {
+            "default": {
+                "concurrency": 100,
+                "pool": "gevent",
+            },
+            "high": {
+                "concurrency": 1,
+                "pool": "prefork",
+            },
+            "high_mem": { # Not sure what's the difference this one and high
+                "concurrency": 1,
+                "pool": "prefork",
+            },
+        },
+        "cms": {
+            "default": {
+                "concurrency": 100,
+                "pool": "gevent",
+            },
+            "high": {
+                "concurrency": 1,
+                "pool": "prefork",
+            },
+            "low": { # Not sure how to is this used
+                "concurrency": 100,
+                "pool": "gevent",
+            },
+        },
+    }
+    return worker_configs
+
+
+@hooks.lru_cache
+def get_celery_workers_config() -> dict[str, dict[str, t.Any]]:
+    """
+    This function is cached for performance.
+    """
+    return hooks.Filters.CELERY_WORKERS_CONFIG.apply({})
+
+
+def iter_celery_workers_config() -> dict[str, dict[str, t.Any]]:
+    """
+    Yield:
+
+        (name, dict)
+    """
+    return {name: config for name, config in get_celery_workers_config().items()}
+
+
+hooks.Filters.ENV_TEMPLATE_VARIABLES.add_items(
+    [
+        ("iter_celery_workers_config", iter_celery_workers_config),
+    ]
+)
diff --git a/tutor/templates/apps/openedx/settings/partials/common_all.py b/tutor/templates/apps/openedx/settings/partials/common_all.py
@@ -246,5 +246,8 @@
     "user": None,
 }
 
+# Prevents losing tasks when workers are shutdown
+CELERY_ACKS_LATE = True
+
 {{ patch("openedx-common-settings") }}
 ######## End of settings common to LMS and CMS
diff --git a/tutor/templates/k8s/deployments.yml b/tutor/templates/k8s/deployments.yml
@@ -122,53 +122,6 @@ spec:
 ---
 apiVersion: apps/v1
 kind: Deployment
-metadata:
-  name: cms-worker
-  labels:
-    app.kubernetes.io/name: cms-worker
-spec:
-  selector:
-    matchLabels:
-      app.kubernetes.io/name: cms-worker
-  template:
-    metadata:
-      labels:
-        app.kubernetes.io/name: cms-worker
-    spec:
-      securityContext:
-        runAsUser: 1000
-        runAsGroup: 1000
-      containers:
-        - name: cms-worker
-          image: {{ DOCKER_IMAGE_OPENEDX }}
-          args: ["celery", "--app=cms.celery", "worker", "--loglevel=info", "--hostname=edx.cms.core.default.%%h", "--max-tasks-per-child", "100", "--exclude-queues=edx.lms.core.default"]
-          env:
-          - name: SERVICE_VARIANT
-            value: cms
-          - name: DJANGO_SETTINGS_MODULE
-            value: cms.envs.tutor.production
-          volumeMounts:
-            - mountPath: /openedx/edx-platform/lms/envs/tutor/
-              name: settings-lms
-            - mountPath: /openedx/edx-platform/cms/envs/tutor/
-              name: settings-cms
-            - mountPath: /openedx/config
-              name: config
-          securityContext:
-            allowPrivilegeEscalation: false
-      volumes:
-        - name: settings-lms
-          configMap:
-            name: openedx-settings-lms
-        - name: settings-cms
-          configMap:
-            name: openedx-settings-cms
-        - name: config
-          configMap:
-            name: openedx-config
----
-apiVersion: apps/v1
-kind: Deployment
 metadata:
   name: lms
   labels:
@@ -229,52 +182,60 @@ spec:
             - key: uwsgi.ini
               path: uwsgi.ini
 ---
+{% for service, variants in iter_celery_workers_config().items() %}
+{% for variant, config in variants.items() %}
+{% set deployment = service + "-" + "worker" + "-" + variant.replace("_", "-") %}
+---
 apiVersion: apps/v1
 kind: Deployment
 metadata:
-  name: lms-worker
+  name: {{deployment}}
   labels:
-    app.kubernetes.io/name: lms-worker
+    app.kubernetes.io/name: {{deployment}}
 spec:
   selector:
     matchLabels:
-      app.kubernetes.io/name: lms-worker
+      app.kubernetes.io/name: {{deployment}}
   template:
     metadata:
       labels:
-        app.kubernetes.io/name: lms-worker
+        app.kubernetes.io/name: {{deployment}}
     spec:
       securityContext:
         runAsUser: 1000
         runAsGroup: 1000
       containers:
-        - name: lms-worker
+        - name: {{deployment}}
           image: {{ DOCKER_IMAGE_OPENEDX }}
-          args: ["celery", "--app=lms.celery", "worker", "--loglevel=info", "--hostname=edx.lms.core.default.%%h", "--max-tasks-per-child=100", "--exclude-queues=edx.cms.core.default"]
+          args:
+            - "celery"
+            - "--app={{service}}.celery"
+            - "worker"
+            - "--loglevel=info"
+            - "--hostname=edx.{{service}}.core.{{variant}}.%%h"
+            - "--queues=edx.{{service}}.core.{{variant}}"{% for setting, value in config.items() %}
+            - "--{{setting}}={{value}}"{% endfor %}
           env:
           - name: SERVICE_VARIANT
-            value: lms
+            value: {{service}}
           - name: DJANGO_SETTINGS_MODULE
-            value: lms.envs.tutor.production
+            value: {{service}}.envs.tutor.production
           volumeMounts:
-            - mountPath: /openedx/edx-platform/lms/envs/tutor/
-              name: settings-lms
-            - mountPath: /openedx/edx-platform/cms/envs/tutor/
-              name: settings-cms
+            - mountPath: /openedx/edx-platform/{{service}}/envs/tutor/
+              name: settings-{{service}}
             - mountPath: /openedx/config
               name: config
           securityContext:
             allowPrivilegeEscalation: false
       volumes:
-        - name: settings-lms
-          configMap:
-            name: openedx-settings-lms
-        - name: settings-cms
+        - name: settings-{{service}}
           configMap:
-            name: openedx-settings-cms
+            name: openedx-settings-{{service}}
         - name: config
           configMap:
             name: openedx-config
+{% endfor %}
+{% endfor %}
 {% if RUN_ELASTICSEARCH %}
 ---
 apiVersion: apps/v1

diff --git a/tutor/templates/local/docker-compose.prod.yml b/tutor/templates/local/docker-compose.prod.yml
@@ -26,42 +26,28 @@ services:
     {% endif %}
 
   ############# LMS and CMS workers
-  lms-worker:
+  {% for service, variants in iter_celery_workers_config().items() %}
+  {% for variant, config in variants.items() %}
+  {% set deployment = service + "-" + "worker" + "-" + variant.replace("_", "-") %}
+  {{deployment}}:
     image: {{ DOCKER_IMAGE_OPENEDX }}
     environment:
-      SERVICE_VARIANT: lms
-      DJANGO_SETTINGS_MODULE: lms.envs.tutor.production
-    command: celery --app=lms.celery worker --loglevel=info --hostname=edx.lms.core.default.%%h --max-tasks-per-child=100 --exclude-queues=edx.cms.core.default
+      SERVICE_VARIANT: {{service}}
+      DJANGO_SETTINGS_MODULE: {{service}}.envs.tutor.production
+    command: celery --app={{service}}.celery worker --loglevel=info --hostname=edx.{{service}}.core.default.%%h --queues=edx.{{service}}.core.{{variant}}{% for param, value in config.items() %} --{{param}}={{value}}{%endfor%}
     restart: unless-stopped
     volumes:
       - ../apps/openedx/settings/lms:/openedx/edx-platform/lms/envs/tutor:ro
       - ../apps/openedx/settings/cms:/openedx/edx-platform/cms/envs/tutor:ro
       - ../apps/openedx/config:/openedx/config:ro
-      - ../../data/lms:/openedx/data
+      - ../../data/{{service}}:/openedx/data
       - ../../data/openedx-media:/openedx/media
-      {%- for mount in iter_mounts(MOUNTS, "openedx", "lms-worker") %}
+      {%- for mount in iter_mounts(MOUNTS, "openedx", service + "-worker") %}
       - {{ mount }}
       {%- endfor %}
     depends_on:
-      - lms
-
-  cms-worker:
-    image: {{ DOCKER_IMAGE_OPENEDX }}
-    environment:
-      SERVICE_VARIANT: cms
-      DJANGO_SETTINGS_MODULE: cms.envs.tutor.production
-    command: celery --app=cms.celery worker --loglevel=info --hostname=edx.cms.core.default.%%h --max-tasks-per-child 100 --exclude-queues=edx.lms.core.default
-    restart: unless-stopped
-    volumes:
-      - ../apps/openedx/settings/lms:/openedx/edx-platform/lms/envs/tutor:ro
-      - ../apps/openedx/settings/cms:/openedx/edx-platform/cms/envs/tutor:ro
-      - ../apps/openedx/config:/openedx/config:ro
-      - ../../data/cms:/openedx/data
-      - ../../data/openedx-media:/openedx/media
-      {%- for mount in iter_mounts(MOUNTS, "openedx", "cms-worker") %}
-      - {{ mount }}
-      {%- endfor %}
-    depends_on:
-      - cms
+      - {{service}}
+    {% endfor %}
+    {% endfor %}
 
   {{ patch("local-docker-compose-prod-services")|indent(2) }}