Fixes

bin/celery-queues.env

@@ -2,4 +2,4 @@
 # Important: Add new queues to make Celery consume tasks from them.
 
 # NOTE: Keep in sync with posthog/tasks/utils.py
-CELERY_WORKER_QUEUES=celery,email,insight_export,insight_refresh,analytics_queries
+CELERY_WORKER_QUEUES=celery,email,insight_export,insight_refresh,analytics_queries,exports,subscription_delivery

ee/tasks/subscriptions/__init__.py

@@ -11,6 +11,7 @@
 from ee.tasks.subscriptions.subscription_utils import generate_assets
 from posthog import settings
 from posthog.models.subscription import Subscription
+from posthog.tasks.utils import CeleryQueue
 
 logger = structlog.get_logger(__name__)
 
@@ -120,7 +121,7 @@ def _deliver_subscription_report(
         subscription.save()
 
 
-@shared_task()
+@shared_task(queue=CeleryQueue.SUBSCRIPTION_DELIVERY)
 def schedule_all_subscriptions() -> None:
     """
     Schedule all past notifications (with a buffer) to be delivered
@@ -148,12 +149,20 @@ def schedule_all_subscriptions() -> None:
 report_timeout_seconds = settings.PARALLEL_ASSET_GENERATION_MAX_TIMEOUT_MINUTES * 60 * 1.5
 
 
-@shared_task(soft_time_limit=report_timeout_seconds, time_limit=report_timeout_seconds + 10)
+@shared_task(
+    soft_time_limit=report_timeout_seconds,
+    time_limit=report_timeout_seconds + 10,
+    queue=CeleryQueue.SUBSCRIPTION_DELIVERY,
+)
 def deliver_subscription_report(subscription_id: int) -> None:
     return _deliver_subscription_report(subscription_id)
 
 
-@shared_task(soft_time_limit=report_timeout_seconds, time_limit=report_timeout_seconds + 10)
+@shared_task(
+    soft_time_limit=report_timeout_seconds,
+    time_limit=report_timeout_seconds + 10,
+    queue=CeleryQueue.SUBSCRIPTION_DELIVERY,
+)
 def handle_subscription_value_change(
     subscription_id: int, previous_value: str, invite_message: Optional[str] = None
 ) -> None:

posthog/tasks/exporter.py

@@ -5,6 +5,7 @@
 
 from posthog import settings
 from posthog.models import ExportedAsset
+from posthog.tasks.utils import CeleryQueue
 
 EXPORT_QUEUED_COUNTER = Counter(
     "exporter_task_queued",
@@ -42,6 +43,7 @@
     acks_late=True,
     ignore_result=False,
     time_limit=settings.ASSET_GENERATION_MAX_TIMEOUT_SECONDS,
+    queue=CeleryQueue.EXPORTS,
 )
 def export_asset(exported_asset_id: int, limit: Optional[int] = None) -> None:
     from posthog.tasks.exports import csv_exporter, image_exporter

posthog/tasks/utils.py

@@ -1,5 +1,27 @@
 from enum import Enum
 
+# NOTE: These are the queues used for logically separating workloads.
+# Many queues are consumed by one "consumer" - a worker configured to consume from that queue.
+# The goal should be to split up queues based on the type of work being done, so that we can scale effectively
+# and change the consumer configs without the need for code changes
+#
+# Worker consumers config here https://github.com/PostHog/posthog-cloud-infra/blob/main/helm/values/prod.yml#L368
+# e.g.
+#   consumers:
+#     - name: priority
+#       queues:
+#         - email
+#         - stats
+#     - name: default
+#       concurrency: 4
+#       queues:
+#         - celery # default queue for Celery
+#     - name: async
+#       concurrency: 4
+#       queues:
+#         - analytics_queries
+#         - subscriptions
+
 
 # NOTE: Keep in sync with bin/celery-queues.env
 class CeleryQueue(Enum):
@@ -8,4 +30,5 @@ class CeleryQueue(Enum):
     INSIGHT_EXPORT = "insight_export"
     INSIGHT_REFRESH = "insight_refresh"
     ANALYTICS_QUERIES = "analytics_queries"
-    # EXPORTS = "exports"
+    EXPORTS = "exports"
+    SUBSCRIPTION_DELIVERY = "subscription_delivery"