fixing Pyramid integration caller_package level (backport #2950) (#2954)

* Fix Pyramid integration caller_package level (#2950)

This was causing Pyramid to fail to start.

Co-authored-by: Zachary Groves <[email protected]>
Co-authored-by: Kyle Verhoog <[email protected]>

Author: 3 people

.circleci/config.yml

@@ -722,10 +722,11 @@ jobs:
           pattern: '^pyodbc_contrib-'
 
   pyramid:
-    <<: *contrib_job
+    <<: *machine_executor
     steps:
       - run_test:
           pattern: 'pyramid'
+          snapshot: true
 
   requests:
     <<: *contrib_job

ddtrace/contrib/pyramid/patch.py

@@ -1,7 +1,6 @@
 import os
 
 import pyramid.config
-from pyramid.path import caller_package
 
 from ddtrace import config
 from ddtrace.vendor import wrapt
@@ -64,18 +63,6 @@ def traced_init(wrapped, instance, args, kwargs):
     # explicitly set our tween too since `add_tween` will be ignored.
     insert_tween_if_needed(trace_settings)
     kwargs["settings"] = trace_settings
-
-    # `caller_package` works by walking a fixed amount of frames up the stack
-    # to find the calling package. So if we let the original `__init__`
-    # function call it, our wrapper will mess things up.
-    if not kwargs.get("package", None):
-        # Get the package for the third frame up from this one.
-        #   - ddtrace.contrib.pyramid.path
-        #   - ddtrace.vendor.wrapt
-        #   - (this is the frame we want)
-        # DEV: Default is `level=2` which will give us the package from `wrapt`
-        kwargs["package"] = caller_package(level=3)
-
     wrapped(*args, **kwargs)
     trace_pyramid(instance)
 

releasenotes/notes/fix_pyramid_caller_package_level-220f67c8063c2d25.yaml

@@ -0,0 +1,5 @@
+---
+fixes:
+  - |
+    Fix Pyramid caller_package level issue which resulted in crashes when starting Pyramid applications. Level now left at default (2).
+    

riotfile.py

@@ -942,6 +942,7 @@ def select_pys(min_version=MIN_PYTHON_VERSION, max_version=MAX_PYTHON_VERSION):
                 Venv(
                     pys=select_pys(),
                     pkgs={
+                        "requests": [latest],
                         "webtest": [latest],
                         "pyramid": [
                             "~=1.7",

tests/contrib/django/django_app/settings.py

@@ -1,15 +1,7 @@
 import os
 
 from ddtrace import tracer
-from ddtrace.filters import TraceFilter
-
-
-class PingFilter(TraceFilter):
-    def process_trace(self, trace):
-        # Filter out all traces with trace_id = 1
-        # This is done to prevent certain traces from being included in snapshots and
-        # accomplished by propagating an http trace id of 1 with the request to Django.
-        return None if trace and trace[0].trace_id == 1 else trace
+from tests.webclient import PingFilter
 
 
 tracer.configure(

tests/contrib/django/test_django_snapshots.py

@@ -3,65 +3,14 @@
 
 import django
 import pytest
-import requests
-import six
-import tenacity
 
-from ddtrace.context import Context
-from ddtrace.propagation.http import HTTPPropagator
 from tests.utils import snapshot
+from tests.webclient import Client
 
 
 SERVER_PORT = 8000
 
 
-class Client(object):
-    """HTTP Client for making requests to a local http server."""
-
-    def __init__(self, base_url):
-        # type: (str) -> None
-        self._base_url = base_url
-        self._session = requests.Session()
-        # Propagate traces with trace_id = 1 for the ping trace so we can filter them out.
-        c, d = Context(trace_id=1, span_id=1), {}
-        HTTPPropagator.inject(c, d)
-        self._ignore_headers = d
-
-    def _url(self, path):
-        # type: (str) -> str
-        return six.moves.urllib.parse.urljoin(self._base_url, path)
-
-    def get(self, path, **kwargs):
-        return self._session.get(self._url(path), **kwargs)
-
-    def get_ignored(self, path, **kwargs):
-        """Do a normal get request but signal that the trace should be filtered out.
-
-        The signal is a distributed trace id header with the value 1.
-        """
-        headers = kwargs.get("headers", {}).copy()
-        headers.update(self._ignore_headers)
-        kwargs["headers"] = headers
-        return self._session.get(self._url(path), **kwargs)
-
-    def post(self, path, *args, **kwargs):
-        return self._session.post(self._url(path), *args, **kwargs)
-
-    def request(self, method, path, *args, **kwargs):
-        return self._session.request(method, self._url(path), *args, **kwargs)
-
-    def wait(self, path="/", max_tries=100, delay=0.1):
-        # type: (str, int, float) -> None
-        """Wait for the server to start by repeatedly http `get`ting `path` until a 200 is received."""
-
-        @tenacity.retry(stop=tenacity.stop_after_attempt(max_tries), wait=tenacity.wait_fixed(delay))
-        def ping():
-            r = self.get_ignored(path)
-            assert r.status_code == 200
-
-        ping()
-
-
 @pytest.fixture(scope="function")
 def daphne_client(snapshot):
     """Runs a django app hosted with a daphne webserver in a subprocess and

tests/contrib/pyramid/app/app.py

@@ -0,0 +1,36 @@
+from wsgiref.simple_server import make_server
+
+from pyramid.config import Configurator
+from pyramid.response import Response
+
+from ddtrace import tracer
+from tests.webclient import PingFilter
+
+
+tracer.configure(
+    settings={
+        "FILTERS": [PingFilter()],
+    }
+)
+
+
+def hello_world(request):
+    return Response("Hello World!")
+
+
+def tracer_shutdown(request):
+    tracer.shutdown()
+    return Response("shutdown")
+
+
+if __name__ == "__main__":
+    with Configurator() as config:
+        config.add_route("hello", "/")
+        config.add_view(hello_world, route_name="hello")
+
+        config.add_route("tracer-shutdown", "/shutdown-tracer")
+        config.add_view(tracer_shutdown, route_name="tracer-shutdown")
+
+        app = config.make_wsgi_app()
+    server = make_server("0.0.0.0", 8000, app)
+    server.serve_forever()

tests/contrib/pyramid/test_pyramid.py

@@ -1,11 +1,20 @@
+import os
+import subprocess
+
+import pytest
+
 from ddtrace import config
 from ddtrace.constants import ORIGIN_KEY
 from ddtrace.constants import SAMPLING_PRIORITY_KEY
+from tests.webclient import Client
 
 from .utils import PyramidBase
 from .utils import PyramidTestCase
 
 
+SERVER_PORT = 8000
+
+
 def includeme(config):
     pass
 
@@ -102,3 +111,42 @@ def test_distributed_tracing_disabled(self):
         assert span.parent_id != 42
         assert span.get_metric(SAMPLING_PRIORITY_KEY) != 2
         assert span.get_tag(ORIGIN_KEY) != "synthetics"
+
+
+@pytest.fixture(scope="function")
+def pyramid_client(snapshot):
+    """Runs a Pyramid app in a subprocess and returns a client which can be used to query it.
+
+    Traces are flushed by invoking a tracer.shutdown() using a /shutdown-tracer route
+    at the end of the testcase.
+    """
+
+    env = os.environ.copy()
+    env["SERVER_PORT"] = str(SERVER_PORT)
+
+    cmd = ["ddtrace-run", "python", "tests/contrib/pyramid/app/app.py"]
+    proc = subprocess.Popen(
+        cmd,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        close_fds=True,
+        env=env,
+    )
+
+    client = Client("http://localhost:%d" % SERVER_PORT)
+
+    # Wait for the server to start up
+    client.wait()
+
+    try:
+        yield client
+    finally:
+        resp = client.get_ignored("/shutdown-tracer")
+        assert resp.status_code == 200
+        proc.terminate()
+
+
+@pytest.mark.snapshot()
+def test_simple_pyramid_app_endpoint(pyramid_client):
+    r = pyramid_client.get("/")
+    assert r.status_code == 200

tests/snapshots/tests.contrib.pyramid.test_pyramid.test_simple_pyramid_app_endpoint.json

@@ -0,0 +1,27 @@
+[[
+  {
+    "name": "pyramid.request",
+    "service": "pyramid",
+    "resource": "GET hello",
+    "trace_id": 0,
+    "span_id": 1,
+    "error": 0,
+    "parent_id": 0,
+    "type": "web",
+    "meta": {
+      "http.method": "GET",
+      "http.status_code": "200",
+      "http.url": "http://localhost:8000/",
+      "pyramid.route.name": "hello",
+      "runtime-id": "af71bf0b951244fba40d444b10b2e462"
+    },
+    "metrics": {
+      "_dd.agent_psr": 1.0,
+      "_dd.measured": 1,
+      "_dd.tracer_kr": 1.0,
+      "_sampling_priority_v1": 1,
+      "system.pid": 98244
+    },
+    "duration": 178000,
+    "start": 1635361523989149000
+  }]]

tests/webclient.py

@@ -0,0 +1,62 @@
+import requests
+import six
+import tenacity
+
+from ddtrace.context import Context
+from ddtrace.filters import TraceFilter
+from ddtrace.propagation.http import HTTPPropagator
+
+
+class Client(object):
+    """HTTP Client for making requests to a local http server."""
+
+    def __init__(self, base_url):
+        # type: (str) -> None
+        self._base_url = base_url
+        self._session = requests.Session()
+        # Propagate traces with trace_id = 1 for the ping trace so we can filter them out.
+        c, d = Context(trace_id=1, span_id=1), {}
+        HTTPPropagator.inject(c, d)
+        self._ignore_headers = d
+
+    def _url(self, path):
+        # type: (str) -> str
+        return six.moves.urllib.parse.urljoin(self._base_url, path)
+
+    def get(self, path, **kwargs):
+        return self._session.get(self._url(path), **kwargs)
+
+    def get_ignored(self, path, **kwargs):
+        """Do a normal get request but signal that the trace should be filtered out.
+
+        The signal is a distributed trace id header with the value 1.
+        """
+        headers = kwargs.get("headers", {}).copy()
+        headers.update(self._ignore_headers)
+        kwargs["headers"] = headers
+        return self._session.get(self._url(path), **kwargs)
+
+    def post(self, path, *args, **kwargs):
+        return self._session.post(self._url(path), *args, **kwargs)
+
+    def request(self, method, path, *args, **kwargs):
+        return self._session.request(method, self._url(path), *args, **kwargs)
+
+    def wait(self, path="/", max_tries=100, delay=0.1):
+        # type: (str, int, float) -> None
+        """Wait for the server to start by repeatedly http `get`ting `path` until a 200 is received."""
+
+        @tenacity.retry(stop=tenacity.stop_after_attempt(max_tries), wait=tenacity.wait_fixed(delay))
+        def ping():
+            r = self.get_ignored(path)
+            assert r.status_code == 200
+
+        ping()
+
+
+class PingFilter(TraceFilter):
+    def process_trace(self, trace):
+        # Filter out all traces with trace_id = 1
+        # This is done to prevent certain traces from being included in snapshots and
+        # accomplished by propagating an http trace id of 1 with the request to the webserver.
+        return None if trace and trace[0].trace_id == 1 else trace