Celery broker_url transport options (#16787)

* handle url params that celery/kombu accepts only as broker transport options

* add note/comment

Author: ewdurbin

bin/redis-tls

@@ -16,4 +16,10 @@ unset REDIS_PORT
 
 # Ensure all clients have a 5-second timeout.
 # Configure the TLS settings for our Redis connection
+# When adding parameters such as socket_timeout,
+# you may need to update the broker url parsing in
+# warehouse/tasks.py to apply them to celery_transport_options
+# See https://docs.celeryq.dev/projects/kombu/en/stable/_modules/kombu/connection.html#Connection
+# and https://docs.celeryq.dev/projects/kombu/en/stable/reference/kombu.transport.redis.html#transport-options
+# for available/allowed arguments/mappings
 REDIS_URL="$REDIS_URL?socket_timeout=5&ssl_cert_reqs=required&ssl_ca_certs=$(python -m certifi)"

tests/unit/test_tasks.py

@@ -593,6 +593,23 @@ def test_make_celery_app():
             "redis://127.0.0.1:6379/10",
             {},
         ),
+        (
+            Environment.production,
+            True,
+            None,
+            (
+                "rediss://user:pass@redis.example.com:6379/10"
+                "?socket_timeout=5&irreleveant=0"
+                "&ssl_cert_reqs=required&ssl_ca_certs=/p/a/t/h/cacert.pem"
+            ),
+            (
+                "rediss://user:pass@redis.example.com:6379/10"
+                "?ssl_cert_reqs=required&ssl_ca_certs=/p/a/t/h/cacert.pem"
+            ),
+            {
+                "socket_timeout": 5,
+            },
+        ),
     ],
 )
 def test_includeme(

warehouse/tasks.py

@@ -306,6 +306,31 @@ def includeme(config):
             "tcp_keepalive": True,
         }
 
+    if broker_url.startswith("redis"):
+        parsed_url = urllib.parse.urlparse(  # noqa: WH001, going to urlunparse this
+            broker_url
+        )
+        parsed_query = urllib.parse.parse_qs(parsed_url.query)
+
+        celery_transport_options = {
+            "socket_timeout": int,
+        }
+
+        for key, value in parsed_query.copy().items():
+            if key.startswith("ssl_"):
+                continue
+            else:
+                if key in celery_transport_options:
+                    broker_transport_options[key] = celery_transport_options[key](
+                        value[0]
+                    )
+                del parsed_query[key]
+
+        parsed_url = parsed_url._replace(
+            query=urllib.parse.urlencode(parsed_query, doseq=True, safe="/")
+        )
+        broker_url = urllib.parse.urlunparse(parsed_url)
+
     config.registry["celery.app"] = celery.Celery(
         "warehouse", autofinalize=False, set_as_current=False
     )