fix: support django channels >= 3.0 (#2956) (#3149)

Currently ddtrace is not working with daphne #2945. This is due to how django channels configures consumers.

Instead of passing a view to django.urls.path and django.urls.re_path django channel's URLRouter expects an asgi application. This causes an application to fail on startup after we incorrectly wrap an asgi application as a view.

To fix this issue this change checks if the asgi application callback in URLPatterns is wrapped as a views. If so, we unwrap the asgi application. This approach should fix the issue described above.

Co-authored-by: Brett Langdon <[email protected]>
Co-authored-by: Kyle Verhoog <[email protected]>
(cherry picked from commit e10a574)

Co-authored-by: Munir Abdinur <[email protected]>

Author: mergify[bot]

ddtrace/contrib/django/patch.py

@@ -404,7 +404,7 @@ def _instrument_view(django, view):
 
     # If the view itself is not wrapped, wrap it
     if not isinstance(view, wrapt.ObjectProxy):
-        view = wrapt.FunctionWrapper(
+        view = utils.DjangoViewProxy(
             view, traced_func(django, "django.view", resource=func_name(view), ignored_excs=[django.http.Http404])
         )
     return view
@@ -448,6 +448,24 @@ def django_asgi_modifier(span, scope):
     return TraceMiddleware(func(*args, **kwargs), integration_config=config.django, span_modifier=django_asgi_modifier)
 
 
+def unwrap_views(func, instance, args, kwargs):
+    """
+    Django channels uses path() and re_path() to route asgi applications. This broke our initial assumption that
+    django path/re_path/url functions only accept views. Here we unwrap ddtrace view instrumentation from asgi
+    applications.
+
+    Ex. ``channels.routing.URLRouter([path('', get_asgi_application())])``
+    On startup ddtrace.contrib.django.path.instrument_view() will wrap get_asgi_application in a DjangoViewProxy.
+    Since get_asgi_application is not a django view callback this function will unwrap it.
+    """
+    routes = get_argument_value(args, kwargs, 0, "routes")
+    for route in routes:
+        if isinstance(route.callback, utils.DjangoViewProxy):
+            route.callback = route.callback.__wrapped__
+
+    return func(*args, **kwargs)
+
+
 def _patch(django):
     Pin().onto(django)
     trace_utils.wrap(django, "apps.registry.Apps.populate", traced_populate(django))
@@ -497,6 +515,17 @@ def _patch(django):
         import django.views.generic.base
     trace_utils.wrap(django, "views.generic.base.View.as_view", traced_as_view(django))
 
+    try:
+        import channels
+        import channels.routing
+
+        channels_version = tuple(int(x) for x in channels.__version__.split("."))
+        if channels_version >= (3, 0):
+            # ASGI3 is only supported in channels v3.0+
+            trace_utils.wrap(channels.routing, "URLRouter.__init__", unwrap_views)
+    except ImportError:
+        pass  # channels is not installed
+
 
 def patch():
     # DEV: this import will eventually be replaced with the module given from an import hook

ddtrace/contrib/django/utils.py

@@ -10,6 +10,7 @@
 
 from .. import trace_utils
 from ...internal.logger import get_logger
+from ...vendor.wrapt import FunctionWrapper
 from .compat import get_resolver
 from .compat import user_is_authenticated
 
@@ -294,3 +295,21 @@ def _after_request_tags(pin, span, request, response):
     finally:
         if span.resource == REQUEST_DEFAULT_RESOURCE:
             span.resource = request.method
+
+
+class DjangoViewProxy(FunctionWrapper):
+    """
+    This custom function wrapper is used to wrap the callback passed to django views handlers (path/re_path/url).
+    This allows us to distinguish between wrapped django views and wrapped asgi applications in django channels.
+    """
+
+    @property
+    def __module__(self):
+        """
+        DjangoViewProxy.__module__ defaults to ddtrace.contrib.django when a wrapped function does not have
+        a __module__ attribute. This method ensures that DjangoViewProxy.__module__ always returns the module
+        attribute of the wrapped function or an empty string if this attribute is not available.
+        The function Django.urls.path() does not have a __module__ attribute and would require this override
+        to resolve the correct module name.
+        """
+        return self.__wrapped__.__module__

releasenotes/notes/add-django-channels-support-44aefaa21961adff.yaml

@@ -0,0 +1,4 @@
+---
+fixes:
+  - |
+    Fix application crash on startup when using ``channels >= 3.0``.

riotfile.py

@@ -516,11 +516,19 @@ def select_pys(min_version=MIN_PYTHON_VERSION, max_version=MAX_PYTHON_VERSION):
                         "django": [
                             ">=2.1,<2.2",
                             ">=2.2,<2.3",
+                        ],
+                    },
+                ),
+                Venv(
+                    pys=select_pys(min_version="3.6"),
+                    pkgs={
+                        "django": [
                             "~=3.0",
                             "~=3.0.0",
                             "~=3.1.0",
                             "~=3.2.0",
                         ],
+                        "channels": ["~=3.0", latest],
                     },
                 ),
                 Venv(
@@ -530,6 +538,7 @@ def select_pys(min_version=MIN_PYTHON_VERSION, max_version=MAX_PYTHON_VERSION):
                             "~=4.0.0",
                             latest,
                         ],
+                        "channels": ["~=3.0", latest],
                     },
                 ),
             ],

tests/contrib/django/asgi.py

@@ -1,4 +1,30 @@
+from channels.auth import AuthMiddlewareStack
+from channels.routing import ProtocolTypeRouter
+from channels.routing import URLRouter
 from django.core.asgi import get_asgi_application
+from django.urls import re_path
+
+from ddtrace.contrib.asgi import TraceMiddleware
 
 
 application = get_asgi_application()
+
+
+async def simple_asgi_app(scope, receive, send):
+    await send({"type": "http.response.start", "status": 200, "headers": [(b"Content-Type", b"text/plain")]})
+    await send({"type": "http.response.body", "body": b"Hello World. It's me simple asgi app"})
+
+
+channels_application = ProtocolTypeRouter(
+    {
+        "http": AuthMiddlewareStack(
+            URLRouter(
+                [
+                    re_path(r"traced-simple-asgi-app/", TraceMiddleware(simple_asgi_app)),
+                    re_path(r"simple-asgi-app/", simple_asgi_app),
+                    re_path(r"", application),
+                ]
+            ),
+        )
+    }
+)

tests/contrib/django/test_django_snapshots.py

@@ -1,3 +1,4 @@
+from contextlib import contextmanager
 import os
 import subprocess
 import sys
@@ -12,8 +13,8 @@
 SERVER_PORT = 8000
 
 
-@pytest.fixture(scope="function")
-def daphne_client(snapshot):
+@contextmanager
+def daphne_client(django_asgi):
     """Runs a django app hosted with a daphne webserver in a subprocess and
     returns a client which can be used to query it.
 
@@ -33,7 +34,7 @@ def daphne_client(snapshot):
 
     # ddtrace-run uses execl which replaces the process but the webserver process itself might spawn new processes.
     # Right now it doesn't but it's possible that it might in the future (ex. uwsgi).
-    cmd = ["ddtrace-run", "daphne", "-p", str(SERVER_PORT), "tests.contrib.django.asgi:application"]
+    cmd = ["ddtrace-run", "daphne", "-p", str(SERVER_PORT), "tests.contrib.django.asgi:%s" % django_asgi]
     proc = subprocess.Popen(
         cmd,
         stdout=subprocess.PIPE,
@@ -156,28 +157,52 @@ def test_psycopg_query_default(client, psycopg2_patched):
 
 
 @pytest.mark.skipif(django.VERSION < (3, 0, 0), reason="ASGI not supported in django<3")
-@pytest.mark.snapshot(
+@snapshot(
     variants={
         "30": (3, 0, 0) <= django.VERSION < (3, 1, 0),
         "31": (3, 1, 0) <= django.VERSION < (3, 2, 0),
         "3x": django.VERSION >= (3, 2, 0),
     },
+    token_override="tests.contrib.django.test_django_snapshots.test_asgi_200",
 )
-def test_asgi_200(daphne_client):
-    resp = daphne_client.get("/")
-    assert resp.status_code == 200
-    assert resp.content == b"Hello, test app."
+@pytest.mark.parametrize("django_asgi", ["application", "channels_application"])
+def test_asgi_200(django_asgi):
+    with daphne_client(django_asgi) as client:
+        resp = client.get("/")
+        assert resp.status_code == 200
+        assert resp.content == b"Hello, test app."
+
+
+@pytest.mark.skipif(django.VERSION < (3, 0, 0), reason="ASGI not supported in django<3")
+@snapshot()
+def test_asgi_200_simple_app():
+    # The path simple-asgi-app/ routes to an ASGI Application that is not traced
+    # This test should generate an empty snapshot
+    with daphne_client("channels_application") as client:
+        resp = client.get("/simple-asgi-app/")
+        assert resp.status_code == 200
+        assert resp.content == b"Hello World. It's me simple asgi app"
 
 
 @pytest.mark.skipif(django.VERSION < (3, 0, 0), reason="ASGI not supported in django<3")
-@pytest.mark.snapshot(
+@snapshot()
+def test_asgi_200_traced_simple_app():
+    with daphne_client("channels_application") as client:
+        resp = client.get("/traced-simple-asgi-app/")
+        assert resp.status_code == 200
+        assert resp.content == b"Hello World. It's me simple asgi app"
+
+
+@pytest.mark.skipif(django.VERSION < (3, 0, 0), reason="ASGI not supported in django<3")
+@snapshot(
     ignores=["meta.error.stack"],
     variants={
         "30": (3, 0, 0) <= django.VERSION < (3, 1, 0),
         "31": (3, 1, 0) <= django.VERSION < (3, 2, 0),
         "3x": django.VERSION >= (3, 2, 0),
     },
 )
-def test_asgi_500(daphne_client):
-    resp = daphne_client.get("/error-500/")
-    assert resp.status_code == 500
+def test_asgi_500():
+    with daphne_client("application") as client:
+        resp = client.get("/error-500/")
+        assert resp.status_code == 500

tests/snapshots/tests.contrib.django.test_django_snapshots.test_asgi_200_simple_app.json

@@ -0,0 +1 @@
+[]

tests/snapshots/tests.contrib.django.test_django_snapshots.test_asgi_200_traced_simple_app.json

@@ -0,0 +1,27 @@
+[[
+  {
+    "name": "asgi.request",
+    "service": null,
+    "resource": "GET /traced-simple-asgi-app/",
+    "trace_id": 0,
+    "span_id": 1,
+    "parent_id": 0,
+    "type": "web",
+    "meta": {
+      "asgi.version": "3.0",
+      "http.method": "GET",
+      "http.status_code": "200",
+      "http.url": "http://127.0.0.1:8000/traced-simple-asgi-app/",
+      "http.version": "1.1",
+      "runtime-id": "74a5040e28844593a7cbe579f49ab5e4"
+    },
+    "metrics": {
+      "_dd.agent_psr": 1.0,
+      "_dd.top_level": 1,
+      "_dd.tracer_kr": 1.0,
+      "_sampling_priority_v1": 1,
+      "system.pid": 3299
+    },
+    "duration": 491700,
+    "start": 1641325348884634800
+  }]]