update configs

backend/danswer/background/celery/apps/app_base.py

@@ -20,8 +20,6 @@ from danswer.background.celery.celery_redis import RedisConnectorPruning
 from danswer.background.celery.celery_redis import RedisDocumentSet
 from danswer.background.celery.celery_redis import RedisUserGroup
 from danswer.background.celery.celery_utils import celery_is_worker_primary
-from danswer.configs.constants import DanswerRedisLocks
-from danswer.db.engine import get_all_tenant_ids
 from danswer.redis.redis_pool import get_redis_client
 from danswer.utils.logger import ColoredFormatter
 from danswer.utils.logger import PlainFormatter
@@ -173,52 +171,9 @@ def wait_for_redis(sender: Any, **kwargs: Any) -> None:
 
 
 def on_secondary_worker_init(sender: Any, **kwargs: Any) -> None:
-    WAIT_INTERVAL = 5
-    WAIT_LIMIT = 60
-
     logger.info("Running as a secondary celery worker.")
     logger.info("Waiting for all tenant primary workers to be ready...")
-    time_start = time.monotonic()
 
-    while True:
-        tenant_ids = get_all_tenant_ids()
-        # Check if we have a primary worker lock for each tenant
-        all_tenants_ready = all(
-            get_redis_client(tenant_id=tenant_id).exists(
-                DanswerRedisLocks.PRIMARY_WORKER
-            )
-            for tenant_id in tenant_ids
-        )
-
-        if all_tenants_ready:
-            break
-
-        time_elapsed = time.monotonic() - time_start
-        ready_tenants = sum(
-            1
-            for tenant_id in tenant_ids
-            if get_redis_client(tenant_id=tenant_id).exists(
-                DanswerRedisLocks.PRIMARY_WORKER
-            )
-        )
-
-        logger.info(
-            f"Not all tenant primary workers are ready yet. "
-            f"Ready tenants: {ready_tenants}/{len(tenant_ids)} "
-            f"elapsed={time_elapsed:.1f} timeout={WAIT_LIMIT:.1f}"
-        )
-
-        if time_elapsed > WAIT_LIMIT:
-            msg = (
-                f"Not all tenant primary workers were ready within the timeout "
-                f"({WAIT_LIMIT} seconds). Exiting..."
-            )
-            logger.error(msg)
-            raise WorkerShutdown(msg)
-
-        time.sleep(WAIT_INTERVAL)
-
-    logger.info("All tenant primary workers are ready. Continuing...")
     return
 
 

backend/danswer/background/celery/apps/beat.py

@@ -3,26 +3,113 @@ from typing import Any
 
 from celery import Celery
 from celery import signals
+from celery.beat import PersistentScheduler  # type: ignore
 from celery.signals import beat_init
 
 import danswer.background.celery.apps.app_base as app_base
-from danswer.configs.constants import DanswerCeleryPriority
 from danswer.configs.constants import POSTGRES_CELERY_BEAT_APP_NAME
 from danswer.db.engine import get_all_tenant_ids
 from danswer.db.engine import SqlEngine
 from danswer.utils.logger import setup_logger
+from danswer.utils.variable_functionality import fetch_versioned_implementation
 
 logger = setup_logger()
 
+
 celery_app = Celery(__name__)
 celery_app.config_from_object("danswer.background.celery.configs.beat")
 
 
+class DynamicTenantScheduler(PersistentScheduler):
+    def __init__(self, *args: Any, **kwargs: Any) -> None:
+        super(DynamicTenantScheduler, self).__init__(*args, **kwargs)
+        self._last_reload = self.app.now()
+        self._reload_interval = timedelta(minutes=1)
+        self._update_tenant_tasks()
+
+    def setup_schedule(self) -> None:
+        super(DynamicTenantScheduler, self).setup_schedule()
+
+    def tick(self) -> float:
+        retval = super(DynamicTenantScheduler, self).tick()
+        now = self.app.now()
+        if (
+            self._last_reload is None
+            or (now - self._last_reload) > self._reload_interval
+        ):
+            logger.info("Reloading schedule to check for new tenants...")
+            self._update_tenant_tasks()
+            self._last_reload = now
+        return retval
+
+    def _update_tenant_tasks(self) -> None:
+        logger.info("Checking for tenant task updates...")
+        try:
+            tenant_ids = get_all_tenant_ids()
+            tasks_to_schedule = fetch_versioned_implementation(
+                "danswer.background.celery.tasks.tasks", "get_tasks_to_schedule"
+            )
+
+            new_beat_schedule: dict[str, dict[str, Any]] = {}
+
+            current_schedule = getattr(self, "_store", {"entries": {}}).get(
+                "entries", {}
+            )
+
+            existing_tenants = set()
+            for task_name in current_schedule.keys():
+                if "-" in task_name:
+                    existing_tenants.add(task_name.split("-")[-1])
+
+            for tenant_id in tenant_ids:
+                if tenant_id not in existing_tenants:
+                    logger.info(f"Found new tenant: {tenant_id}")
+
+                for task in tasks_to_schedule():
+                    task_name = f"{task['name']}-{tenant_id}"
+                    new_task = {
+                        "task": task["task"],
+                        "schedule": task["schedule"],
+                        "kwargs": {"tenant_id": tenant_id},
+                    }
+                    if options := task.get("options"):
+                        new_task["options"] = options
+                    new_beat_schedule[task_name] = new_task
+
+            if self._should_update_schedule(current_schedule, new_beat_schedule):
+                logger.info(
+                    "Updating schedule",
+                    extra={
+                        "new_tasks": len(new_beat_schedule),
+                        "current_tasks": len(current_schedule),
+                    },
+                )
+                if not hasattr(self, "_store"):
+                    self._store: dict[str, dict] = {"entries": {}}
+                self.update_from_dict(new_beat_schedule)
+                logger.info("Tenant tasks updated successfully")
+            else:
+                logger.debug("No schedule updates needed")
+
+        except (AttributeError, KeyError):
+            logger.exception("Error processing task configuration:")
+        except Exception as e:
+            logger.exception("Unexpected error updating tenant tasks: %s", str(e))
+
+    def _should_update_schedule(
+        self, current_schedule: dict, new_schedule: dict
+    ) -> bool:
+        """Compare schedules to determine if an update is needed."""
+        current_tasks = set(current_schedule.keys())
+        new_tasks = set(new_schedule.keys())
+        return current_tasks != new_tasks
+
+
 @beat_init.connect
 def on_beat_init(sender: Any, **kwargs: Any) -> None:
     logger.info("beat_init signal received.")
 
-    # celery beat shouldn't touch the db at all. But just setting a low minimum here.
+    # Celery beat shouldn't touch the db at all. But just setting a low minimum here.
     SqlEngine.set_app_name(POSTGRES_CELERY_BEAT_APP_NAME)
     SqlEngine.init_engine(pool_size=2, max_overflow=0)
     app_base.wait_for_redis(sender, **kwargs)
@@ -35,68 +122,4 @@ def on_setup_logging(
     app_base.on_setup_logging(loglevel, logfile, format, colorize, **kwargs)
 
 
-#####
-# Celery Beat (Periodic Tasks) Settings
-#####
-
-tenant_ids = get_all_tenant_ids()
-
-tasks_to_schedule = [
-    {
-        "name": "check-for-vespa-sync",
-        "task": "check_for_vespa_sync_task",
-        "schedule": timedelta(seconds=5),
-        "options": {"priority": DanswerCeleryPriority.HIGH},
-    },
-    {
-        "name": "check-for-connector-deletion",
-        "task": "check_for_connector_deletion_task",
-        "schedule": timedelta(seconds=60),
-        "options": {"priority": DanswerCeleryPriority.HIGH},
-    },
-    {
-        "name": "check-for-indexing",
-        "task": "check_for_indexing",
-        "schedule": timedelta(seconds=10),
-        "options": {"priority": DanswerCeleryPriority.HIGH},
-    },
-    {
-        "name": "check-for-prune",
-        "task": "check_for_pruning",
-        "schedule": timedelta(seconds=10),
-        "options": {"priority": DanswerCeleryPriority.HIGH},
-    },
-    {
-        "name": "kombu-message-cleanup",
-        "task": "kombu_message_cleanup_task",
-        "schedule": timedelta(seconds=3600),
-        "options": {"priority": DanswerCeleryPriority.LOWEST},
-    },
-    {
-        "name": "monitor-vespa-sync",
-        "task": "monitor_vespa_sync",
-        "schedule": timedelta(seconds=5),
-        "options": {"priority": DanswerCeleryPriority.HIGH},
-    },
-]
-
-
-# Build the celery beat schedule dynamically
-beat_schedule = {}
-
-for tenant_id in tenant_ids:
-    for task in tasks_to_schedule:
-        task_name = f"{task['name']}-{tenant_id}"  # Unique name for each scheduled task
-        beat_schedule[task_name] = {
-            "task": task["task"],
-            "schedule": task["schedule"],
-            "options": task["options"],
-            "kwargs": {"tenant_id": tenant_id},  # Must pass tenant_id as an argument
-        }
-
-# Include any existing beat schedules
-existing_beat_schedule = celery_app.conf.beat_schedule or {}
-beat_schedule.update(existing_beat_schedule)
-
-# Update the Celery app configuration once
-celery_app.conf.beat_schedule = beat_schedule
+celery_app.conf.beat_scheduler = DynamicTenantScheduler

backend/danswer/background/celery/apps/primary.py

@@ -5,7 +5,6 @@ from celery import bootsteps  # type: ignore
 from celery import Celery
 from celery import signals
 from celery import Task
-from celery.exceptions import WorkerShutdown
 from celery.signals import celeryd_init
 from celery.signals import worker_init
 from celery.signals import worker_ready
@@ -13,13 +12,6 @@ from celery.signals import worker_shutdown
 
 import danswer.background.celery.apps.app_base as app_base
 from danswer.background.celery.apps.app_base import task_logger
-from danswer.background.celery.celery_redis import RedisConnectorCredentialPair
-from danswer.background.celery.celery_redis import RedisConnectorDeletion
-from danswer.background.celery.celery_redis import RedisConnectorIndexing
-from danswer.background.celery.celery_redis import RedisConnectorPruning
-from danswer.background.celery.celery_redis import RedisConnectorStop
-from danswer.background.celery.celery_redis import RedisDocumentSet
-from danswer.background.celery.celery_redis import RedisUserGroup
 from danswer.background.celery.celery_utils import celery_is_worker_primary
 from danswer.configs.constants import CELERY_PRIMARY_WORKER_LOCK_TIMEOUT
 from danswer.configs.constants import DanswerRedisLocks
@@ -81,90 +73,6 @@ def on_worker_init(sender: Any, **kwargs: Any) -> None:
 
     sender.primary_worker_locks = {}
 
-    # This is singleton work that should be done on startup exactly once
-    # by the primary worker
-    tenant_ids = get_all_tenant_ids()
-    for tenant_id in tenant_ids:
-        r = get_redis_client(tenant_id=tenant_id)
-
-        # For the moment, we're assuming that we are the only primary worker
-        # that should be running.
-        # TODO: maybe check for or clean up another zombie primary worker if we detect it
-        r.delete(DanswerRedisLocks.PRIMARY_WORKER)
-
-        # this process wide lock is taken to help other workers start up in order.
-        # it is planned to use this lock to enforce singleton behavior on the primary
-        # worker, since the primary worker does redis cleanup on startup, but this isn't
-        # implemented yet.
-        lock = r.lock(
-            DanswerRedisLocks.PRIMARY_WORKER,
-            timeout=CELERY_PRIMARY_WORKER_LOCK_TIMEOUT,
-        )
-
-        logger.info("Primary worker lock: Acquire starting.")
-        acquired = lock.acquire(blocking_timeout=CELERY_PRIMARY_WORKER_LOCK_TIMEOUT / 2)
-        if acquired:
-            logger.info("Primary worker lock: Acquire succeeded.")
-        else:
-            logger.error("Primary worker lock: Acquire failed!")
-            raise WorkerShutdown("Primary worker lock could not be acquired!")
-
-        # tacking on our own user data to the sender
-        sender.primary_worker_locks[tenant_id] = lock
-
-        # As currently designed, when this worker starts as "primary", we reinitialize redis
-        # to a clean state (for our purposes, anyway)
-        r.delete(DanswerRedisLocks.CHECK_VESPA_SYNC_BEAT_LOCK)
-        r.delete(DanswerRedisLocks.MONITOR_VESPA_SYNC_BEAT_LOCK)
-
-        r.delete(RedisConnectorCredentialPair.get_taskset_key())
-        r.delete(RedisConnectorCredentialPair.get_fence_key())
-
-        for key in r.scan_iter(RedisDocumentSet.TASKSET_PREFIX + "*"):
-            r.delete(key)
-
-        for key in r.scan_iter(RedisDocumentSet.FENCE_PREFIX + "*"):
-            r.delete(key)
-
-        for key in r.scan_iter(RedisUserGroup.TASKSET_PREFIX + "*"):
-            r.delete(key)
-
-        for key in r.scan_iter(RedisUserGroup.FENCE_PREFIX + "*"):
-            r.delete(key)
-
-        for key in r.scan_iter(RedisConnectorDeletion.TASKSET_PREFIX + "*"):
-            r.delete(key)
-
-        for key in r.scan_iter(RedisConnectorDeletion.FENCE_PREFIX + "*"):
-            r.delete(key)
-
-        for key in r.scan_iter(RedisConnectorPruning.TASKSET_PREFIX + "*"):
-            r.delete(key)
-
-        for key in r.scan_iter(RedisConnectorPruning.GENERATOR_COMPLETE_PREFIX + "*"):
-            r.delete(key)
-
-        for key in r.scan_iter(RedisConnectorPruning.GENERATOR_PROGRESS_PREFIX + "*"):
-            r.delete(key)
-
-        for key in r.scan_iter(RedisConnectorPruning.FENCE_PREFIX + "*"):
-            r.delete(key)
-
-        for key in r.scan_iter(RedisConnectorIndexing.TASKSET_PREFIX + "*"):
-            r.delete(key)
-
-        for key in r.scan_iter(RedisConnectorIndexing.GENERATOR_COMPLETE_PREFIX + "*"):
-            r.delete(key)
-
-        for key in r.scan_iter(RedisConnectorIndexing.GENERATOR_PROGRESS_PREFIX + "*"):
-            r.delete(key)
-
-        for key in r.scan_iter(RedisConnectorIndexing.FENCE_PREFIX + "*"):
-            r.delete(key)
-
-        for key in r.scan_iter(RedisConnectorStop.FENCE_PREFIX + "*"):
-            r.delete(key)
-
 
 @worker_ready.connect
 def on_worker_ready(sender: Any, **kwargs: Any) -> None:

backend/danswer/background/celery/tasks/tasks.py

@@ -0,0 +1,48 @@
+from datetime import timedelta
+from typing import Any
+
+from danswer.configs.constants import DanswerCeleryPriority
+
+
+tasks_to_schedule = [
+    {
+        "name": "check-for-vespa-sync",
+        "task": "check_for_vespa_sync_task",
+        "schedule": timedelta(seconds=5),
+        "options": {"priority": DanswerCeleryPriority.HIGH},
+    },
+    {
+        "name": "check-for-connector-deletion",
+        "task": "check_for_connector_deletion_task",
+        "schedule": timedelta(seconds=60),
+        "options": {"priority": DanswerCeleryPriority.HIGH},
+    },
+    {
+        "name": "check-for-indexing",
+        "task": "check_for_indexing",
+        "schedule": timedelta(seconds=10),
+        "options": {"priority": DanswerCeleryPriority.HIGH},
+    },
+    {
+        "name": "check-for-prune",
+        "task": "check_for_pruning",
+        "schedule": timedelta(seconds=10),
+        "options": {"priority": DanswerCeleryPriority.HIGH},
+    },
+    {
+        "name": "kombu-message-cleanup",
+        "task": "kombu_message_cleanup_task",
+        "schedule": timedelta(seconds=3600),
+        "options": {"priority": DanswerCeleryPriority.LOWEST},
+    },
+    {
+        "name": "monitor-vespa-sync",
+        "task": "monitor_vespa_sync",
+        "schedule": timedelta(seconds=5),
+        "options": {"priority": DanswerCeleryPriority.HIGH},
+    },
+]
+
+
+def get_tasks_to_schedule() -> list[dict[str, Any]]:
+    return tasks_to_schedule

backend/ee/danswer/background/celery/apps/beat.py

@@ -1,52 +0,0 @@
-#####
-# Celery Beat (Periodic Tasks) Settings
-#####
-from datetime import timedelta
-
-from danswer.background.celery.apps.beat import celery_app
-from danswer.db.engine import get_all_tenant_ids
-
-
-tenant_ids = get_all_tenant_ids()
-
-tasks_to_schedule = [
-    {
-        "name": "sync-external-doc-permissions",
-        "task": "check_sync_external_doc_permissions_task",
-        "schedule": timedelta(seconds=5),  # TODO: optimize this
-    },
-    {
-        "name": "sync-external-group-permissions",
-        "task": "check_sync_external_group_permissions_task",
-        "schedule": timedelta(seconds=5),  # TODO: optimize this
-    },
-    {
-        "name": "autogenerate_usage_report",
-        "task": "autogenerate_usage_report_task",
-        "schedule": timedelta(days=30),  # TODO: change this to config flag
-    },
-    {
-        "name": "check-ttl-management",
-        "task": "check_ttl_management_task",
-        "schedule": timedelta(hours=1),
-    },
-]
-
-# Build the celery beat schedule dynamically
-beat_schedule = {}
-
-for tenant_id in tenant_ids:
-    for task in tasks_to_schedule:
-        task_name = f"{task['name']}-{tenant_id}"  # Unique name for each scheduled task
-        beat_schedule[task_name] = {
-            "task": task["task"],
-            "schedule": task["schedule"],
-            "kwargs": {"tenant_id": tenant_id},  # Must pass tenant_id as an argument
-        }
-
-# Include any existing beat schedules
-existing_beat_schedule = celery_app.conf.beat_schedule or {}
-beat_schedule.update(existing_beat_schedule)
-
-# Update the Celery app configuration
-celery_app.conf.beat_schedule = beat_schedule

backend/ee/danswer/background/celery/tasks/tasks.py

@@ -0,0 +1,33 @@
+from datetime import timedelta
+from typing import Any
+
+from danswer.background.celery.tasks.tasks import (
+    tasks_to_schedule as base_tasks_to_schedule,
+)
+
+ee_tasks_to_schedule = [
+    {
+        "name": "sync-external-doc-permissions",
+        "task": "check_sync_external_doc_permissions_task",
+        "schedule": timedelta(seconds=5),  # TODO: optimize this
+    },
+    {
+        "name": "sync-external-group-permissions",
+        "task": "check_sync_external_group_permissions_task",
+        "schedule": timedelta(seconds=5),  # TODO: optimize this
+    },
+    {
+        "name": "autogenerate_usage_report",
+        "task": "autogenerate_usage_report_task",
+        "schedule": timedelta(days=30),  # TODO: change this to config flag
+    },
+    {
+        "name": "check-ttl-management",
+        "task": "check_ttl_management_task",
+        "schedule": timedelta(hours=1),
+    },
+]
+
+
+def get_tasks_to_schedule() -> list[dict[str, Any]]:
+    return ee_tasks_to_schedule + base_tasks_to_schedule

backend/scripts/dev_run_background_jobs.py

@@ -75,7 +75,7 @@ def run_jobs() -> None:
     cmd_beat = [
         "celery",
         "-A",
-        "danswer.background.celery.versioned_apps.beat",
+        "danswer.background.celery.apps.beat",
         "beat",
         "--loglevel=INFO",
     ]

deployment/cloud_kubernetes/hpa/workers_hpa.yaml

@@ -0,0 +1,56 @@
+apiVersion: autoscaling/v2
+kind: HorizontalPodAutoscaler
+metadata:
+  name: celery-worker-heavy-hpa
+spec:
+  scaleTargetRef:
+    apiVersion: apps/v1
+    kind: Deployment
+    name: celery-worker-heavy
+  minReplicas: 1
+  maxReplicas: 5
+  metrics:
+    - type: Resource
+      resource:
+        name: cpu
+        target:
+          type: Utilization
+          averageUtilization: 60
+---
+apiVersion: autoscaling/v2
+kind: HorizontalPodAutoscaler
+metadata:
+  name: celery-worker-light-hpa
+spec:
+  scaleTargetRef:
+    apiVersion: apps/v1
+    kind: Deployment
+    name: celery-worker-light
+  minReplicas: 1
+  maxReplicas: 10
+  metrics:
+    - type: Resource
+      resource:
+        name: cpu
+        target:
+          type: Utilization
+          averageUtilization: 70
+---
+apiVersion: autoscaling/v2
+kind: HorizontalPodAutoscaler
+metadata:
+  name: celery-worker-indexing-hpa
+spec:
+  scaleTargetRef:
+    apiVersion: apps/v1
+    kind: Deployment
+    name: celery-worker-indexing
+  minReplicas: 1
+  maxReplicas: 10
+  metrics:
+    - type: Resource
+      resource:
+        name: cpu
+        target:
+          type: Utilization
+          averageUtilization: 70

deployment/cloud_kubernetes/workers/beat.yaml

@@ -0,0 +1,44 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: celery-beat
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: celery-beat
+  template:
+    metadata:
+      labels:
+        app: celery-beat
+    spec:
+      containers:
+        - name: celery-beat
+          image: danswer/danswer-backend:v0.11.0-cloud.beta.2
+          imagePullPolicy: Always
+          command:
+            [
+              "celery",
+              "-A",
+              "danswer.background.celery.versioned_apps.beat",
+              "beat",
+              "--loglevel=INFO",
+            ]
+          env:
+            - name: REDIS_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: danswer-secrets
+                  key: redis_password
+            - name: DANSWER_VERSION
+              value: "v0.11.0-cloud.beta.1"
+          envFrom:
+            - configMapRef:
+                name: env-configmap
+          resources:
+            requests:
+              cpu: "250m"
+              memory: "512Mi"
+            limits:
+              cpu: "500m"
+              memory: "1Gi"

deployment/cloud_kubernetes/workers/heavy_worker.yaml

@@ -0,0 +1,60 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: celery-worker-heavy
+spec:
+  replicas: 2
+  selector:
+    matchLabels:
+      app: celery-worker-heavy
+  template:
+    metadata:
+      labels:
+        app: celery-worker-heavy
+    spec:
+      containers:
+        - name: celery-worker-heavy
+          image: danswer/danswer-backend:v0.11.0-cloud.beta.2
+          imagePullPolicy: Always
+          command:
+            [
+              "celery",
+              "-A",
+              "danswer.background.celery.versioned_apps.heavy",
+              "worker",
+              "--loglevel=INFO",
+              "--hostname=heavy@%n",
+              "-Q",
+              "connector_pruning",
+            ]
+          env:
+            - name: REDIS_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: danswer-secrets
+                  key: redis_password
+            - name: DANSWER_VERSION
+              value: "v0.11.0-cloud.beta.1"
+          envFrom:
+            - configMapRef:
+                name: env-configmap
+          volumeMounts:
+            - name: vespa-certificates
+              mountPath: "/app/certs"
+              readOnly: true
+          resources:
+            requests:
+              cpu: "1000m"
+              memory: "2Gi"
+            limits:
+              cpu: "2000m"
+              memory: "4Gi"
+      volumes:
+        - name: vespa-certificates
+          secret:
+            secretName: vespa-certificates
+            items:
+              - key: cert.pem
+                path: cert.pem
+              - key: key.pem
+                path: key.pem

deployment/cloud_kubernetes/workers/indexing_worker.yaml

@@ -0,0 +1,60 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: celery-worker-indexing
+spec:
+  replicas: 3
+  selector:
+    matchLabels:
+      app: celery-worker-indexing
+  template:
+    metadata:
+      labels:
+        app: celery-worker-indexing
+    spec:
+      containers:
+        - name: celery-worker-indexing
+          image: danswer/danswer-backend:v0.11.0-cloud.beta.2
+          imagePullPolicy: Always
+          command:
+            [
+              "celery",
+              "-A",
+              "danswer.background.celery.versioned_apps.indexing",
+              "worker",
+              "--loglevel=INFO",
+              "--hostname=indexing@%n",
+              "-Q",
+              "connector_indexing",
+            ]
+          env:
+            - name: REDIS_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: danswer-secrets
+                  key: redis_password
+            - name: DANSWER_VERSION
+              value: "v0.11.0-cloud.beta.1"
+          envFrom:
+            - configMapRef:
+                name: env-configmap
+          volumeMounts:
+            - name: vespa-certificates
+              mountPath: "/app/certs"
+              readOnly: true
+          resources:
+            requests:
+              cpu: "500m"
+              memory: "1Gi"
+            limits:
+              cpu: "1000m"
+              memory: "2Gi"
+      volumes:
+        - name: vespa-certificates
+          secret:
+            secretName: vespa-certificates
+            items:
+              - key: cert.pem
+                path: cert.pem
+              - key: key.pem
+                path: key.pem

deployment/cloud_kubernetes/workers/light_worker.yaml

@@ -0,0 +1,60 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: celery-worker-light
+spec:
+  replicas: 2
+  selector:
+    matchLabels:
+      app: celery-worker-light
+  template:
+    metadata:
+      labels:
+        app: celery-worker-light
+    spec:
+      containers:
+        - name: celery-worker-light
+          image: danswer/danswer-backend:v0.11.0-cloud.beta.2
+          imagePullPolicy: Always
+          command:
+            [
+              "celery",
+              "-A",
+              "danswer.background.celery.versioned_apps.light",
+              "worker",
+              "--loglevel=INFO",
+              "--hostname=light@%n",
+              "-Q",
+              "vespa_metadata_sync,connector_deletion",
+            ]
+          env:
+            - name: REDIS_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: danswer-secrets
+                  key: redis_password
+            - name: DANSWER_VERSION
+              value: "v0.11.0-cloud.beta.1"
+          envFrom:
+            - configMapRef:
+                name: env-configmap
+          volumeMounts:
+            - name: vespa-certificates
+              mountPath: "/app/certs"
+              readOnly: true
+          resources:
+            requests:
+              cpu: "500m"
+              memory: "1Gi"
+            limits:
+              cpu: "1000m"
+              memory: "2Gi"
+      volumes:
+        - name: vespa-certificates
+          secret:
+            secretName: vespa-certificates
+            items:
+              - key: cert.pem
+                path: cert.pem
+              - key: key.pem
+                path: key.pem

deployment/cloud_kubernetes/workers/primary.yaml

@@ -0,0 +1,60 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: celery-worker-primary
+spec:
+  replicas: 1
+  selector:
+    matchLabels:
+      app: celery-worker-primary
+  template:
+    metadata:
+      labels:
+        app: celery-worker-primary
+    spec:
+      containers:
+        - name: celery-worker-primary
+          image: danswer/danswer-backend:v0.11.0-cloud.beta.2
+          imagePullPolicy: Always
+          command:
+            [
+              "celery",
+              "-A",
+              "danswer.background.celery.versioned_apps.primary",
+              "worker",
+              "--loglevel=INFO",
+              "--hostname=primary@%n",
+              "-Q",
+              "celery",
+            ]
+          env:
+            - name: REDIS_PASSWORD
+              valueFrom:
+                secretKeyRef:
+                  name: danswer-secrets
+                  key: redis_password
+            - name: DANSWER_VERSION
+              value: "v0.11.0-cloud.beta.1"
+          envFrom:
+            - configMapRef:
+                name: env-configmap
+          volumeMounts:
+            - name: vespa-certificates
+              mountPath: "/app/certs"
+              readOnly: true
+          resources:
+            requests:
+              cpu: "500m"
+              memory: "1Gi"
+            limits:
+              cpu: "1000m"
+              memory: "2Gi"
+      volumes:
+        - name: vespa-certificates
+          secret:
+            secretName: vespa-certificates
+            items:
+              - key: cert.pem
+                path: cert.pem
+              - key: key.pem
+                path: key.pem