Multiple celery queues (#7226)

* Separate Celery queue for malware tasks

* remove malware queue and routes for now

Author: ewdurbin

Procfile

@@ -1,5 +1,5 @@
 release: bin/release
 web: bin/start-web python -m gunicorn.app.wsgiapp -c gunicorn.conf.py warehouse.wsgi:application
 web-uploads: bin/start-web python -m gunicorn.app.wsgiapp -c gunicorn-uploads.conf.py warehouse.wsgi:application
-worker: bin/start-worker celery -A warehouse worker -l info --max-tasks-per-child 32
+worker: bin/start-worker celery -A warehouse worker -Q default -l info --max-tasks-per-child 32
 worker-beat: bin/start-worker celery -A warehouse beat -S redbeat.RedBeatScheduler -l info

dev/environment

@@ -7,7 +7,7 @@ WAREHOUSE_TOKEN=insecuretoken
 
 AWS_ACCESS_KEY_ID=foo
 AWS_SECRET_ACCESS_KEY=foo
-BROKER_URL=sqs://localstack:4576/warehouse-dev?region=us-east-1
+BROKER_URL=sqs://localstack:4576/?region=us-east-1&queue_name_prefix=warehouse-dev
 
 DATABASE_URL=postgresql://postgres@db/warehouse
 

tests/unit/test_tasks.py

@@ -17,6 +17,7 @@
 import transaction
 
 from celery import Celery, Task
+from kombu import Queue
 from pyramid import scripting
 from pyramid_retry import RetryableException
 
@@ -372,63 +373,69 @@ def test_make_celery_app():
 
 
 @pytest.mark.parametrize(
-    ("env", "ssl", "broker_url", "expected_url", "queue_name", "transport_options"),
+    ("env", "ssl", "broker_url", "expected_url", "transport_options"),
     [
         (
             Environment.development,
             False,
             "amqp://guest@rabbitmq:5672//",
             "amqp://guest@rabbitmq:5672//",
-            "celery",
             {},
         ),
         (
             Environment.production,
             True,
             "amqp://guest@rabbitmq:5672//",
             "amqp://guest@rabbitmq:5672//",
-            "celery",
             {},
         ),
-        (Environment.development, False, "sqs://", "sqs://", "celery", {}),
-        (Environment.production, True, "sqs://", "sqs://", "celery", {}),
-        (Environment.development, False, "sqs:///my-queue", "sqs://", "my-queue", {}),
-        (Environment.production, True, "sqs:///my-queue", "sqs://", "my-queue", {}),
+        (Environment.development, False, "sqs://", "sqs://", {}),
+        (Environment.production, True, "sqs://", "sqs://", {}),
+        (
+            Environment.development,
+            False,
+            "sqs://?queue_name_prefix=warehouse",
+            "sqs://",
+            {"queue_name_prefix": "warehouse-"},
+        ),
+        (
+            Environment.production,
+            True,
+            "sqs://?queue_name_prefix=warehouse",
+            "sqs://",
+            {"queue_name_prefix": "warehouse-"},
+        ),
         (
             Environment.development,
             False,
             "sqs://?region=us-east-2",
             "sqs://",
-            "celery",
             {"region": "us-east-2"},
         ),
         (
             Environment.production,
             True,
             "sqs://?region=us-east-2",
             "sqs://",
-            "celery",
             {"region": "us-east-2"},
         ),
         (
             Environment.development,
             False,
-            "sqs:///my-queue?region=us-east-2",
+            "sqs:///?region=us-east-2&queue_name_prefix=warehouse",
             "sqs://",
-            "my-queue",
-            {"region": "us-east-2"},
+            {"region": "us-east-2", "queue_name_prefix": "warehouse-"},
         ),
         (
             Environment.production,
             True,
-            "sqs:///my-queue?region=us-east-2",
+            "sqs:///?region=us-east-2&queue_name_prefix=warehouse",
             "sqs://",
-            "my-queue",
-            {"region": "us-east-2"},
+            {"region": "us-east-2", "queue_name_prefix": "warehouse-"},
         ),
     ],
 )
-def test_includeme(env, ssl, broker_url, expected_url, queue_name, transport_options):
+def test_includeme(env, ssl, broker_url, expected_url, transport_options):
     registry_dict = {}
     config = pretend.stub(
         action=pretend.call_recorder(lambda *a, **kw: None),
@@ -456,10 +463,13 @@ def test_includeme(env, ssl, broker_url, expected_url, queue_name, transport_opt
         "broker_url": expected_url,
         "broker_use_ssl": ssl,
         "worker_disable_rate_limits": True,
-        "task_default_queue": queue_name,
+        "task_default_queue": "default",
+        "task_default_routing_key": "task.default",
         "task_serializer": "json",
         "accept_content": ["json", "msgpack"],
         "task_queue_ha_policy": "all",
+        "task_queues": (Queue("default", routing_key="task.#"),),
+        "task_routes": ([]),
         "REDBEAT_REDIS_URL": (config.registry.settings["celery.scheduler_url"]),
     }.items():
         assert app.conf[key] == value

warehouse/tasks.py

@@ -21,6 +21,7 @@
 import transaction
 import venusian
 
+from kombu import Queue
 from pyramid.threadlocal import get_current_request
 
 from warehouse.config import Environment
@@ -158,7 +159,6 @@ def add_task():
 def includeme(config):
     s = config.registry.settings
 
-    queue_name = "celery"
     broker_transport_options = {}
 
     broker_url = s["celery.broker_url"]
@@ -169,8 +169,10 @@ def includeme(config):
         # so we'll just remove them from here.
         broker_url = urllib.parse.urlunparse(parsed_url[:2] + ("", "", "", ""))
 
-        if parsed_url.path:
-            queue_name = parsed_url.path[1:]
+        if "queue_name_prefix" in parsed_query:
+            broker_transport_options["queue_name_prefix"] = (
+                parsed_query["queue_name_prefix"][0] + "-"
+            )
 
         if "region" in parsed_query:
             broker_transport_options["region"] = parsed_query["region"][0]
@@ -183,8 +185,11 @@ def includeme(config):
         broker_url=broker_url,
         broker_use_ssl=s["warehouse.env"] == Environment.production,
         broker_transport_options=broker_transport_options,
-        task_default_queue=queue_name,
+        task_default_queue="default",
+        task_default_routing_key="task.default",
         task_queue_ha_policy="all",
+        task_queues=(Queue("default", routing_key="task.#"),),
+        task_routes=([]),
         task_serializer="json",
         worker_disable_rate_limits=True,
         REDBEAT_REDIS_URL=s["celery.scheduler_url"],