From 2fa8088e77fbf77f4d7583c8ca95fc6b03f940d8 Mon Sep 17 00:00:00 2001 From: Tianning Li Date: Thu, 20 Aug 2026 00:06:16 -0400 Subject: [PATCH 1/4] feat(runtime): add explicit identity refresh Some runtimes can restore a process from a snapshot without creating a real fork. Reusing the fork hook there would give the process a fresh runtime id, but it would also record fake parent and ancestor lineage. Add a direct refresh_identity() path for that case. Runtime-id subscribers are weakly held and isolated from each other so long-lived components can rebuild cached identity state without leaking instances or breaking the refresh caller. --- ddtrace/internal/runtime/__init__.py | 84 +++++++++++++--- tests/tracer/runtime/test_runtime_id.py | 125 +++++++++++++++++++++++- 2 files changed, 193 insertions(+), 16 deletions(-) diff --git a/ddtrace/internal/runtime/__init__.py b/ddtrace/internal/runtime/__init__.py index 3bfb2ff28aa..0cd60046b6c 100644 --- a/ddtrace/internal/runtime/__init__.py +++ b/ddtrace/internal/runtime/__init__.py @@ -1,9 +1,13 @@ import typing as t import uuid +import weakref +from ddtrace.internal import forksafe +from ddtrace.internal.logger import get_logger from ddtrace.internal.settings import env -from .. import forksafe + +log = get_logger(__name__) __all__ = [ @@ -12,6 +16,7 @@ "get_runtime_id", "get_parent_runtime_id", "get_runtime_propagation_envs", + "refresh_identity", ] @@ -30,30 +35,87 @@ def _generate_runtime_id() -> str: _PARENT_RUNTIME_ID: t.Optional[str] = env.get(_ENV_PARENT_SESSION_ID) # IMPORTANT: Do not change t.Set to set until minimum Python version is 3.11+ # Module-level set[...] in Python 3.10 affects import timing. See packages.py for details. -_ON_RUNTIME_ID_CHANGE: t.Set[t.Callable[[str], None]] = set() # noqa: UP006 +# Held as weak references: subscribers are typically long-lived singletons or objects +# owned elsewhere (RemoteConfigClient, TelemetryWriter, trace writer instances), and tests +# construct many short-lived instances of some of these. A strong reference here would keep +# every instance ever constructed alive for the life of the process. +_ON_RUNTIME_ID_CHANGE: t.Set["weakref.ReferenceType[t.Callable[[str], None]]"] = set() # noqa: UP006 def on_runtime_id_change(cb: t.Callable[[str], None]) -> None: - """Register a callback to be called when the runtime ID changes. - - This can happen after a fork. + """Register a callback to be called when refresh_identity() runs. + + refresh_identity() is the non-fork trigger for a new logical process + instance. It is deliberately not called after a plain fork: forked children + already get a fresh runtime ID silently (see _set_runtime_id()), and code + that needs to react to a fork specifically should use forksafe.register(). + Only a weak reference to cb is kept, so the caller must keep it alive for + it to keep firing. """ global _ON_RUNTIME_ID_CHANGE - _ON_RUNTIME_ID_CHANGE.add(cb) + try: + ref = weakref.WeakMethod(cb) if hasattr(cb, "__self__") else weakref.ref(cb) + except TypeError: + # Some callables with a __self__ (e.g. certain C-implemented bound methods) aren't + # compatible with WeakMethod, and some objects aren't weakrefable at all. Skip + # registration rather than crash the caller's (often component-init) code path. + log.debug("Could not weakly reference on_runtime_id_change() subscriber %r; skipping", cb) + return + _ON_RUNTIME_ID_CHANGE.add(ref) + + +def _regenerate_runtime_id() -> None: + global _RUNTIME_ID + _RUNTIME_ID = _generate_runtime_id() + + +def _notify_runtime_id_subscribers() -> None: + global _ON_RUNTIME_ID_CHANGE + dead = set() + # Snapshot into a list before iterating: a concurrent on_runtime_id_change() (e.g. a + # RemoteConfigClient/writer being constructed on another thread) mutating the live set + # mid-iteration would otherwise raise "Set changed size during iteration". + for ref in list(_ON_RUNTIME_ID_CHANGE): + cb = ref() + if cb is None: + dead.add(ref) + continue + try: + cb(_RUNTIME_ID) + except Exception: + # One broken subscriber must not prevent other subscribers from seeing the + # refreshed runtime ID. + log.debug("Error notifying on_runtime_id_change() subscriber", exc_info=True) + _ON_RUNTIME_ID_CHANGE -= dead @forksafe.register -def _set_runtime_id(): - global _RUNTIME_ID, _ANCESTOR_RUNTIME_ID, _PARENT_RUNTIME_ID +def _set_runtime_id() -> None: + global _ANCESTOR_RUNTIME_ID, _PARENT_RUNTIME_ID # Save the runtime ID of the common ancestor of all processes. if _ANCESTOR_RUNTIME_ID is None: _ANCESTOR_RUNTIME_ID = _RUNTIME_ID _PARENT_RUNTIME_ID = _RUNTIME_ID - _RUNTIME_ID = _generate_runtime_id() - for cb in _ON_RUNTIME_ID_CHANGE: - cb(_RUNTIME_ID) + # Does not notify on_runtime_id_change() subscribers: a fork has its own dedicated + # forksafe hooks (per subscriber) for resetting fork-inherited state, which differs + # from a plain rebuild-in-place (e.g. RemoteConfigClient's SHM-for-fork native client + # must survive a fork untouched; see RemoteConfigPoller.reset_at_fork()). + _regenerate_runtime_id() + + +def refresh_identity() -> None: + """Regenerate the runtime ID without recording fork lineage. + + Unlike a fork, this does not update _PARENT_RUNTIME_ID / _ANCESTOR_RUNTIME_ID: + the previous runtime ID was not a real parent process, so recording it there + would make get_process_role() and friends misreport a fork lineage that never + existed. Use this when a new logical process instance is created by a mechanism + other than fork(). + """ + _regenerate_runtime_id() + _notify_runtime_id_subscribers() def get_runtime_id() -> str: diff --git a/tests/tracer/runtime/test_runtime_id.py b/tests/tracer/runtime/test_runtime_id.py index 0b44fccb902..40d57733798 100644 --- a/tests/tracer/runtime/test_runtime_id.py +++ b/tests/tracer/runtime/test_runtime_id.py @@ -3,7 +3,7 @@ @pytest.mark.subprocess def test_get_runtime_id(): - from ddtrace.internal import runtime + import ddtrace.internal.runtime as runtime runtime_id = runtime.get_runtime_id() assert isinstance(runtime_id, str) @@ -15,7 +15,7 @@ def test_get_runtime_id(): def test_get_runtime_id_fork(): import os - from ddtrace.internal import runtime + import ddtrace.internal.runtime as runtime runtime_id = runtime.get_runtime_id() assert isinstance(runtime_id, str) @@ -44,7 +44,7 @@ def test_get_runtime_id_fork(): def test_get_runtime_id_double_fork(): import os - from ddtrace.internal import runtime + import ddtrace.internal.runtime as runtime runtime_id = runtime.get_runtime_id() @@ -88,7 +88,7 @@ def test_ancestor_runtime_id(): """ import os - from ddtrace.internal import runtime + import ddtrace.internal.runtime as runtime ancestor_runtime_id = runtime.get_runtime_id() @@ -133,7 +133,7 @@ def test_parent_runtime_id(): """get_parent_runtime_id() tracks the immediate parent process, not the root.""" import os - from ddtrace.internal import runtime + import ddtrace.internal.runtime as runtime root_id = runtime.get_runtime_id() assert runtime.get_parent_runtime_id() is None @@ -210,3 +210,118 @@ def test_get_process_role_spawn_child() -> None: from ddtrace.internal.runtime import get_process_role assert get_process_role() == "worker", get_process_role() + + +@pytest.mark.subprocess +def test_refresh_identity_changes_runtime_id(): + """refresh_identity() is the non-fork trigger for a new logical process instance.""" + import ddtrace.internal.runtime as runtime + + runtime_id = runtime.get_runtime_id() + runtime.refresh_identity() + new_runtime_id = runtime.get_runtime_id() + + assert isinstance(new_runtime_id, str) + assert new_runtime_id != runtime_id + assert new_runtime_id == runtime.get_runtime_id() + + +@pytest.mark.subprocess( + env={ + "_DD_ROOT_PY_SESSION_ID": None, + "_DD_PARENT_PY_SESSION_ID": None, + "DD_TRACE_SUBPROCESS_ENABLED": "false", + } +) +def test_refresh_identity_does_not_record_fork_lineage(): + """Unlike a fork, refresh_identity() must not make get_process_role() report a fake worker. + + The previous runtime ID was not a real parent process, so recording it as one would + corrupt process-lineage telemetry. + """ + import ddtrace.internal.runtime as runtime + + assert runtime.get_process_role() is None + assert runtime.get_parent_runtime_id() is None + assert runtime.get_ancestor_runtime_id() is None + + runtime.refresh_identity() + + assert runtime.get_process_role() is None + assert runtime.get_parent_runtime_id() is None + assert runtime.get_ancestor_runtime_id() is None + + +@pytest.mark.subprocess +def test_refresh_identity_notifies_subscribers(): + import ddtrace.internal.runtime as runtime + + seen = [] + + class _Subscriber: + def on_change(self, new_id): + seen.append(new_id) + + subscriber = _Subscriber() + runtime.on_runtime_id_change(subscriber.on_change) + + runtime.refresh_identity() + + assert seen == [runtime.get_runtime_id()] + + +@pytest.mark.subprocess +def test_refresh_identity_isolates_subscriber_exceptions(): + """One subscriber raising must not stop refresh_identity() or block other subscribers.""" + import ddtrace.internal.runtime as runtime + + seen = [] + + class _BadSubscriber: + def on_change(self, new_id): + raise ValueError("boom") + + class _GoodSubscriber: + def on_change(self, new_id): + seen.append(new_id) + + bad = _BadSubscriber() + good = _GoodSubscriber() + runtime.on_runtime_id_change(bad.on_change) + runtime.on_runtime_id_change(good.on_change) + + runtime.refresh_identity() + + assert seen == [runtime.get_runtime_id()] + + +@pytest.mark.subprocess +def test_on_runtime_id_change_does_not_leak_dead_subscribers(): + """Subscribers are held weakly: once garbage collected they stop firing and are pruned. + + Subscribers are typically objects constructed many times over a process's life (e.g. a + trace writer instance per Tracer()); a strong reference here would keep every one of + them alive for the life of the process. + """ + import gc + + import ddtrace.internal.runtime as runtime + + class _Subscriber: + def on_change(self, new_id): + pass + + # Baseline, not 0: injection/auto-instrumentation may have already constructed a + # RemoteConfigClient/Writer/TelemetryWriter in this process, each of which subscribes. + baseline = len(runtime._ON_RUNTIME_ID_CHANGE) + + subscriber = _Subscriber() + runtime.on_runtime_id_change(subscriber.on_change) + assert len(runtime._ON_RUNTIME_ID_CHANGE) == baseline + 1 + + del subscriber + gc.collect() + + runtime.refresh_identity() + + assert len(runtime._ON_RUNTIME_ID_CHANGE) == baseline From 2df658bdb76f7607f2753e79a76b626258755e8b Mon Sep 17 00:00:00 2001 From: Tianning Li Date: Thu, 20 Aug 2026 00:08:33 -0400 Subject: [PATCH 2/4] feat(web): emit request-starting event The MicroVM /run hook needs to be observed before a web root span reads process identity, but Python has no single HTTP server substrate across WSGI, ASGI, and framework integrations. Introduce a shared core event that web integrations emit once method and path are available, before root span creation. The event is deliberately generic here; it does not know about MicroVMs or refresh runtime ids yet. --- .riot/requirements/1362718.txt | 20 ++++ .riot/requirements/13ed954.txt | 20 ++++ .riot/requirements/1638d0f.txt | 20 ++++ .riot/requirements/180047f.txt | 25 ++++ .riot/requirements/3cb8c3c.txt | 23 ++++ .riot/requirements/43423a3.txt | 20 ++++ ddtrace/_monkey.py | 2 + ddtrace/contrib/internal/asgi/middleware.py | 2 + ddtrace/contrib/internal/bottle/patch.py | 2 + ddtrace/contrib/internal/bottle/trace.py | 10 ++ ddtrace/contrib/internal/cherrypy/patch.py | 2 + ddtrace/contrib/internal/django/response.py | 2 + ddtrace/contrib/internal/falcon/middleware.py | 2 + ddtrace/contrib/internal/flask/patch.py | 1 + .../contrib/internal/http_server/__init__.py | 17 +++ ddtrace/contrib/internal/http_server/patch.py | 71 +++++++++++ ddtrace/contrib/internal/molten/patch.py | 4 + ddtrace/contrib/internal/pyramid/trace.py | 113 +++++++++--------- ddtrace/contrib/internal/sanic/patch.py | 2 + ddtrace/contrib/internal/tornado/handlers.py | 2 + ddtrace/internal/core/__init__.py | 3 + .../settings/_supported_configurations.py | 3 + docs/integrations.rst | 6 + riotfile.py | 8 ++ scripts/integration_registry/registry.yaml | 4 + supported-configurations.json | 17 +++ .../asgi/test_microvm_identity_refresh.py | 75 ++++++++++++ .../bottle/test_microvm_identity_refresh.py | 53 ++++++++ .../cherrypy/test_microvm_identity_refresh.py | 51 ++++++++ .../django/test_microvm_identity_refresh.py | 26 ++++ .../falcon/test_microvm_identity_refresh.py | 34 ++++++ .../flask/test_microvm_identity_refresh.py | 70 +++++++++++ tests/contrib/http_server/__init__.py | 0 .../http_server/test_http_server_patch.py | 26 ++++ .../test_microvm_identity_refresh.py | 65 ++++++++++ .../molten/test_microvm_identity_refresh.py | 46 +++++++ .../pyramid/test_microvm_identity_refresh.py | 41 +++++++ tests/contrib/sanic/test_sanic.py | 27 +++++ tests/contrib/suitespec.yml | 10 ++ .../tornado/test_microvm_identity_refresh.py | 30 +++++ 40 files changed, 899 insertions(+), 56 deletions(-) create mode 100644 .riot/requirements/1362718.txt create mode 100644 .riot/requirements/13ed954.txt create mode 100644 .riot/requirements/1638d0f.txt create mode 100644 .riot/requirements/180047f.txt create mode 100644 .riot/requirements/3cb8c3c.txt create mode 100644 .riot/requirements/43423a3.txt create mode 100644 ddtrace/contrib/internal/http_server/__init__.py create mode 100644 ddtrace/contrib/internal/http_server/patch.py create mode 100644 tests/contrib/asgi/test_microvm_identity_refresh.py create mode 100644 tests/contrib/bottle/test_microvm_identity_refresh.py create mode 100644 tests/contrib/cherrypy/test_microvm_identity_refresh.py create mode 100644 tests/contrib/django/test_microvm_identity_refresh.py create mode 100644 tests/contrib/falcon/test_microvm_identity_refresh.py create mode 100644 tests/contrib/flask/test_microvm_identity_refresh.py create mode 100644 tests/contrib/http_server/__init__.py create mode 100644 tests/contrib/http_server/test_http_server_patch.py create mode 100644 tests/contrib/http_server/test_microvm_identity_refresh.py create mode 100644 tests/contrib/molten/test_microvm_identity_refresh.py create mode 100644 tests/contrib/pyramid/test_microvm_identity_refresh.py create mode 100644 tests/contrib/tornado/test_microvm_identity_refresh.py diff --git a/.riot/requirements/1362718.txt b/.riot/requirements/1362718.txt new file mode 100644 index 00000000000..4c18cd862ab --- /dev/null +++ b/.riot/requirements/1362718.txt @@ -0,0 +1,20 @@ +# +# This file is autogenerated by pip-compile with Python 3.12 +# by the following command: +# +# pip-compile --allow-unsafe --no-annotate .riot/requirements/1362718.in +# +attrs==26.1.0 +coverage[toml]==7.15.4 +hypothesis==6.45.0 +iniconfig==2.3.0 +mock==5.2.0 +opentracing==2.4.0 +packaging==26.3 +pluggy==1.6.0 +pygments==2.21.0 +pytest==9.1.1 +pytest-cov==7.1.0 +pytest-mock==3.15.1 +pytest-randomly==4.1.0 +sortedcontainers==2.4.0 diff --git a/.riot/requirements/13ed954.txt b/.riot/requirements/13ed954.txt new file mode 100644 index 00000000000..16c89155daf --- /dev/null +++ b/.riot/requirements/13ed954.txt @@ -0,0 +1,20 @@ +# +# This file is autogenerated by pip-compile with Python 3.11 +# by the following command: +# +# pip-compile --allow-unsafe --no-annotate .riot/requirements/13ed954.in +# +attrs==26.1.0 +coverage[toml]==7.15.4 +hypothesis==6.45.0 +iniconfig==2.3.0 +mock==5.2.0 +opentracing==2.4.0 +packaging==26.3 +pluggy==1.6.0 +pygments==2.21.0 +pytest==9.1.1 +pytest-cov==7.1.0 +pytest-mock==3.15.1 +pytest-randomly==4.1.0 +sortedcontainers==2.4.0 diff --git a/.riot/requirements/1638d0f.txt b/.riot/requirements/1638d0f.txt new file mode 100644 index 00000000000..4cf4d9cc372 --- /dev/null +++ b/.riot/requirements/1638d0f.txt @@ -0,0 +1,20 @@ +# +# This file is autogenerated by pip-compile with Python 3.14 +# by the following command: +# +# pip-compile --allow-unsafe --no-annotate .riot/requirements/1638d0f.in +# +attrs==26.1.0 +coverage[toml]==7.15.4 +hypothesis==6.45.0 +iniconfig==2.3.0 +mock==5.2.0 +opentracing==2.4.0 +packaging==26.3 +pluggy==1.6.0 +pygments==2.21.0 +pytest==9.1.1 +pytest-cov==7.1.0 +pytest-mock==3.15.1 +pytest-randomly==4.1.0 +sortedcontainers==2.4.0 diff --git a/.riot/requirements/180047f.txt b/.riot/requirements/180047f.txt new file mode 100644 index 00000000000..954fc0398ba --- /dev/null +++ b/.riot/requirements/180047f.txt @@ -0,0 +1,25 @@ +# +# This file is autogenerated by pip-compile with Python 3.9 +# by the following command: +# +# pip-compile --allow-unsafe --no-annotate .riot/requirements/180047f.in +# +attrs==26.1.0 +coverage[toml]==7.10.7 +exceptiongroup==1.3.1 +hypothesis==6.45.0 +importlib-metadata==8.7.1 +iniconfig==2.1.0 +mock==5.2.0 +opentracing==2.4.0 +packaging==26.3 +pluggy==1.6.0 +pygments==2.21.0 +pytest==8.4.2 +pytest-cov==7.1.0 +pytest-mock==3.15.1 +pytest-randomly==4.0.1 +sortedcontainers==2.4.0 +tomli==2.4.1 +typing-extensions==4.16.0 +zipp==3.23.1 diff --git a/.riot/requirements/3cb8c3c.txt b/.riot/requirements/3cb8c3c.txt new file mode 100644 index 00000000000..95297a33249 --- /dev/null +++ b/.riot/requirements/3cb8c3c.txt @@ -0,0 +1,23 @@ +# +# This file is autogenerated by pip-compile with Python 3.10 +# by the following command: +# +# pip-compile --allow-unsafe --no-annotate .riot/requirements/3cb8c3c.in +# +attrs==26.1.0 +coverage[toml]==7.15.4 +exceptiongroup==1.3.1 +hypothesis==6.45.0 +iniconfig==2.3.0 +mock==5.2.0 +opentracing==2.4.0 +packaging==26.3 +pluggy==1.6.0 +pygments==2.21.0 +pytest==9.1.1 +pytest-cov==7.1.0 +pytest-mock==3.15.1 +pytest-randomly==4.1.0 +sortedcontainers==2.4.0 +tomli==2.4.1 +typing-extensions==4.16.0 diff --git a/.riot/requirements/43423a3.txt b/.riot/requirements/43423a3.txt new file mode 100644 index 00000000000..e09b95353d4 --- /dev/null +++ b/.riot/requirements/43423a3.txt @@ -0,0 +1,20 @@ +# +# This file is autogenerated by pip-compile with Python 3.13 +# by the following command: +# +# pip-compile --allow-unsafe --no-annotate .riot/requirements/43423a3.in +# +attrs==26.1.0 +coverage[toml]==7.15.4 +hypothesis==6.45.0 +iniconfig==2.3.0 +mock==5.2.0 +opentracing==2.4.0 +packaging==26.3 +pluggy==1.6.0 +pygments==2.21.0 +pytest==9.1.1 +pytest-cov==7.1.0 +pytest-mock==3.15.1 +pytest-randomly==4.1.0 +sortedcontainers==2.4.0 diff --git a/ddtrace/_monkey.py b/ddtrace/_monkey.py index ab5bc9dcd2d..f9012404428 100644 --- a/ddtrace/_monkey.py +++ b/ddtrace/_monkey.py @@ -75,6 +75,7 @@ "aiopg": True, "aiobotocore": False, "httplib": False, + "http_server": True, "urllib3": False, "vertexai": True, "vertica": True, @@ -165,6 +166,7 @@ "azure_functions": ("azure.functions",), "azure_servicebus": ("azure.servicebus",), "httplib": ("http.client",), + "http_server": ("http.server",), "kafka": ("confluent_kafka",), "google_adk": ("google.adk",), "google_cloud_pubsub": ("google.cloud.pubsub_v1",), diff --git a/ddtrace/contrib/internal/asgi/middleware.py b/ddtrace/contrib/internal/asgi/middleware.py index 0f4b38e6717..5f4f81a30dc 100644 --- a/ddtrace/contrib/internal/asgi/middleware.py +++ b/ddtrace/contrib/internal/asgi/middleware.py @@ -215,6 +215,8 @@ async def __call__(self, scope: Mapping[str, Any], receive: Callable, send: Call method = "websocket" else: return await self.app(scope, receive, send) + if not is_subapp and scope["type"] == "http": + core.dispatch(core.WEB_REQUEST_STARTING, (method, scope["path"])) try: headers = extract_headers(scope) except Exception: diff --git a/ddtrace/contrib/internal/bottle/patch.py b/ddtrace/contrib/internal/bottle/patch.py index 6e1f378f34a..8fd14bef4bf 100644 --- a/ddtrace/contrib/internal/bottle/patch.py +++ b/ddtrace/contrib/internal/bottle/patch.py @@ -6,6 +6,7 @@ from ddtrace.internal.utils.formats import asbool from .trace import TracePlugin +from .trace import traced_wsgi # Configure default configuration @@ -32,6 +33,7 @@ def patch(): bottle._datadog_patch = True wrapt.wrap_function_wrapper("bottle", "Bottle.__init__", traced_init) + wrapt.wrap_function_wrapper("bottle", "Bottle.wsgi", traced_wsgi) def traced_init(wrapped, instance, args, kwargs): diff --git a/ddtrace/contrib/internal/bottle/trace.py b/ddtrace/contrib/internal/bottle/trace.py index c33507a0098..27518929db4 100644 --- a/ddtrace/contrib/internal/bottle/trace.py +++ b/ddtrace/contrib/internal/bottle/trace.py @@ -15,6 +15,16 @@ from ddtrace.vendor.debtcollector import deprecate +def traced_wsgi(wrapped, instance, args, kwargs): + """Wraps Bottle's WSGI entry point so the MicroVM ``/run`` hook is detected even when no + route matches (404) or the matched route is a wildcard -- ``TracePlugin.apply()`` only + wraps callbacks for routes that already matched, so it can't see either case. + """ + environ = args[0] + core.dispatch(core.WEB_REQUEST_STARTING, (environ.get("REQUEST_METHOD"), environ.get("PATH_INFO"))) + return wrapped(*args, **kwargs) + + class TracePlugin(object): name = "trace" api = 2 diff --git a/ddtrace/contrib/internal/cherrypy/patch.py b/ddtrace/contrib/internal/cherrypy/patch.py index c96c983eb28..edd3a6a0955 100644 --- a/ddtrace/contrib/internal/cherrypy/patch.py +++ b/ddtrace/contrib/internal/cherrypy/patch.py @@ -73,6 +73,8 @@ def _setup(self): cherrypy.request.hooks.attach("after_error_response", self._after_error_response, priority=5) def _on_start_resource(self): + core.dispatch(core.WEB_REQUEST_STARTING, (cherrypy.request.method, cherrypy.request.path_info)) + with core.context_with_data( "cherrypy.request", span_name=SPAN_NAME, diff --git a/ddtrace/contrib/internal/django/response.py b/ddtrace/contrib/internal/django/response.py index cbb17d1fc26..dd06601b1ec 100644 --- a/ddtrace/contrib/internal/django/response.py +++ b/ddtrace/contrib/internal/django/response.py @@ -89,6 +89,8 @@ def traced_get_response(func: FunctionType, args: tuple[Any, ...], kwargs: dict[ if request is None: return func(*args, **kwargs) + core.dispatch(core.WEB_REQUEST_STARTING, (request.method, request.path)) + request_headers = utils._get_request_headers(request) pin = Pin.get_from(instance) diff --git a/ddtrace/contrib/internal/falcon/middleware.py b/ddtrace/contrib/internal/falcon/middleware.py index f96966fac10..f6cab1a382c 100644 --- a/ddtrace/contrib/internal/falcon/middleware.py +++ b/ddtrace/contrib/internal/falcon/middleware.py @@ -29,6 +29,8 @@ def __init__(self, tracer=None, service=None, distributed_tracing=None): config.falcon["distributed_tracing"] = distributed_tracing def process_request(self, req, resp): + core.dispatch(core.WEB_REQUEST_STARTING, (req.method, req.path)) + # Falcon uppercases all header names. headers = dict((k.lower(), v) for k, v in req.headers.items()) diff --git a/ddtrace/contrib/internal/flask/patch.py b/ddtrace/contrib/internal/flask/patch.py index 12ceeb8ec9f..d767fb86ad8 100644 --- a/ddtrace/contrib/internal/flask/patch.py +++ b/ddtrace/contrib/internal/flask/patch.py @@ -391,6 +391,7 @@ def unpatch(): def patched_wsgi_app(wrapped, instance, args, kwargs): environ, start_response = args + core.dispatch(core.WEB_REQUEST_STARTING, (environ.get("REQUEST_METHOD"), environ.get("PATH_INFO"))) # Registration is gated on asm_config, not tracing — keep this above the tracing short-circuit. _collect_routes_once(instance, environ.get("SCRIPT_NAME") or "") if not is_tracing_enabled(): diff --git a/ddtrace/contrib/internal/http_server/__init__.py b/ddtrace/contrib/internal/http_server/__init__.py new file mode 100644 index 00000000000..18d5bae76c0 --- /dev/null +++ b/ddtrace/contrib/internal/http_server/__init__.py @@ -0,0 +1,17 @@ +""" +Patch the standard library ``http.server`` module (``BaseHTTPRequestHandler``). + +This integration does **not** create spans. Its only purpose is detecting the AWS Lambda +MicroVM ``/run`` lifecycle hook for applications that implement that hook with a raw +``http.server`` handler instead of a supported web framework. + + +Enabling +~~~~~~~~ + +The http_server integration is enabled by default. Use +:ref:`ddtrace-run` or :ref:`import ddtrace.auto` to enable it, and +disable it with ``DD_TRACE_HTTP_SERVER_ENABLED=false`` if needed:: + + DD_TRACE_HTTP_SERVER_ENABLED=false ddtrace-run .... +""" diff --git a/ddtrace/contrib/internal/http_server/patch.py b/ddtrace/contrib/internal/http_server/patch.py new file mode 100644 index 00000000000..003fe14bcfc --- /dev/null +++ b/ddtrace/contrib/internal/http_server/patch.py @@ -0,0 +1,71 @@ +from __future__ import annotations + +import sys +from types import ModuleType +from typing import TYPE_CHECKING +from typing import Any +from typing import Callable + +from wrapt import wrap_function_wrapper as _w + +from ddtrace.contrib.internal.trace_utils import unwrap as _u +from ddtrace.internal import core + + +if TYPE_CHECKING: + import http.server + + +def _get_http_server() -> ModuleType: + # DEV: When patch() is called from the on-import hook, we're running from inside + # http.server's own exec_module(), before CPython's import machinery binds it as the + # "server" attribute of the "http" package (that setattr() happens only after + # exec_module() returns to _find_and_load()). Accessing it via `http.server` (attribute + # chain) at that point raises "cannot access submodule 'server' of module 'http' (most + # likely due to a circular import)". sys.modules is populated before exec_module() even + # starts, so prefer it -- falling back to a plain import for direct patch() calls that + # happen before anything has imported http.server yet (not in sys.modules at all). + loaded = sys.modules.get("http.server") + if loaded is not None: + return loaded + import http.server + + return http.server + + +def get_version() -> str: + return "" + + +def _supported_versions() -> dict[str, str]: + return {"http.server": "*"} + + +def _wrap_parse_request( + wrapped: Callable[..., bool], + instance: "http.server.BaseHTTPRequestHandler", + args: tuple[Any, ...], + kwargs: dict[str, Any], +) -> bool: + parsed = wrapped(*args, **kwargs) + if parsed: + core.dispatch(core.WEB_REQUEST_STARTING, (instance.command, instance.path)) + return parsed + + +def patch() -> None: + http_server = _get_http_server() + if getattr(http_server, "__datadog_patch", False): + return + http_server.__datadog_patch = True # type: ignore[attr-defined] # patch marker, not a real module attr + + _w(http_server.BaseHTTPRequestHandler, "parse_request", _wrap_parse_request) + + +def unpatch() -> None: + http_server = _get_http_server() + if not getattr(http_server, "__datadog_patch", False): + return + http_server.__datadog_patch = False # type: ignore[attr-defined] # patch marker, not a real module attr + + _u(http_server.BaseHTTPRequestHandler, "parse_request") diff --git a/ddtrace/contrib/internal/molten/patch.py b/ddtrace/contrib/internal/molten/patch.py index a30900743cf..a71050d60b3 100644 --- a/ddtrace/contrib/internal/molten/patch.py +++ b/ddtrace/contrib/internal/molten/patch.py @@ -67,6 +67,10 @@ def unpatch(): def patch_app_call(wrapped, instance, args, kwargs): + # DEV: This is safe because this is the args for a WSGI handler + # https://www.python.org/dev/peps/pep-3333/ + core.dispatch(core.WEB_REQUEST_STARTING, (args[0].get("REQUEST_METHOD"), args[0].get("PATH_INFO"))) + pin = Pin.get_from(molten) if not pin or not pin.enabled(): diff --git a/ddtrace/contrib/internal/pyramid/trace.py b/ddtrace/contrib/internal/pyramid/trace.py index 3022bd9936a..e9b8ea28abf 100644 --- a/ddtrace/contrib/internal/pyramid/trace.py +++ b/ddtrace/contrib/internal/pyramid/trace.py @@ -60,59 +60,60 @@ def trace_tween_factory(handler, registry): # ensure distributed tracing within pyramid settings matches config config.pyramid.distributed_tracing_enabled = asbool(settings.get(SETTINGS_DISTRIBUTED_TRACING, True)) - if enabled: - # make a request tracing function - def trace_tween(request): - with core.context_with_event( - WebFrameworkRequestEvent( - http_operation="pyramid.request", - service=service, - resource="404", - integration_config=config.pyramid, - component=config.pyramid.integration_name, - request_headers=request.headers, - request_url=request.url, - request_method=request.method, - request_route=None, - query=request.query_string, - activate_distributed_headers=True, - headers_case_sensitive=True, - ) - ) as ctx: - response = None - status = None - try: - response = handler(request) - except HTTPException as e: - # If the exception is a pyramid HTTPException, - # that's still valuable information that isn't necessarily - # a 500. For instance, HTTPFound is a 302. - # As described in docs, Pyramid exceptions are all valid - # response types - response = e - raise - except BaseException: - status = 500 - raise - finally: - event: WebFrameworkRequestEvent = ctx.event - # set request tags - if request.matched_route: - event.resource = "{} {}".format(request.method, request.matched_route.name) - event.request_route = request.matched_route.pattern - span_from_context(ctx)._set_attribute("pyramid.route.name", request.matched_route.name) - # set response tags - if response: - status = response.status_code - response_headers = response.headers - else: - response_headers = {} - event.response_headers = response_headers - event.response_status_code = status - - return response - - return trace_tween - - # if timing support is not enabled, return the original handler - return handler + # make a request tracing function + def trace_tween(request): + core.dispatch(core.WEB_REQUEST_STARTING, (request.method, request.path)) + + if not enabled: + return handler(request) + + with core.context_with_event( + WebFrameworkRequestEvent( + http_operation="pyramid.request", + service=service, + resource="404", + integration_config=config.pyramid, + component=config.pyramid.integration_name, + request_headers=request.headers, + request_url=request.url, + request_method=request.method, + request_route=None, + query=request.query_string, + activate_distributed_headers=True, + headers_case_sensitive=True, + ) + ) as ctx: + response = None + status = None + try: + response = handler(request) + except HTTPException as e: + # If the exception is a pyramid HTTPException, + # that's still valuable information that isn't necessarily + # a 500. For instance, HTTPFound is a 302. + # As described in docs, Pyramid exceptions are all valid + # response types + response = e + raise + except BaseException: + status = 500 + raise + finally: + event: WebFrameworkRequestEvent = ctx.event + # set request tags + if request.matched_route: + event.resource = "{} {}".format(request.method, request.matched_route.name) + event.request_route = request.matched_route.pattern + span_from_context(ctx)._set_attribute("pyramid.route.name", request.matched_route.name) + # set response tags + if response: + status = response.status_code + response_headers = response.headers + else: + response_headers = {} + event.response_headers = response_headers + event.response_status_code = status + + return response + + return trace_tween diff --git a/ddtrace/contrib/internal/sanic/patch.py b/ddtrace/contrib/internal/sanic/patch.py index 72a32dfb6c0..d74424cbe7e 100644 --- a/ddtrace/contrib/internal/sanic/patch.py +++ b/ddtrace/contrib/internal/sanic/patch.py @@ -191,6 +191,8 @@ def unwrap(request, write_callback=None, stream_callback=None, **kwargs): def _create_sanic_request_span(request): """Helper to create sanic.request span and attach a pin to request.ctx""" + core.dispatch(core.WEB_REQUEST_STARTING, (request.method, request.path)) + pin = Pin() pin.onto(request.ctx) diff --git a/ddtrace/contrib/internal/tornado/handlers.py b/ddtrace/contrib/internal/tornado/handlers.py index b0bb8aa2437..8a29fdd5dde 100644 --- a/ddtrace/contrib/internal/tornado/handlers.py +++ b/ddtrace/contrib/internal/tornado/handlers.py @@ -28,6 +28,8 @@ async def execute(func, handler, args, kwargs): ``TracerStackContext``. This simplifies users code when the automatic ``Context`` retrieval is used via ``Tracer.trace()`` method. """ + core.dispatch(core.WEB_REQUEST_STARTING, (handler.request.method, handler.request.path)) + # retrieve tracing settings settings = handler.settings[CONFIG_KEY] service = settings["default_service"] diff --git a/ddtrace/internal/core/__init__.py b/ddtrace/internal/core/__init__.py index 512524bef96..3e140012f29 100644 --- a/ddtrace/internal/core/__init__.py +++ b/ddtrace/internal/core/__init__.py @@ -143,6 +143,9 @@ def done_callback(f): ROOT_CONTEXT_ID = "__root" +# Emitted by web integrations after method/path are available but before request +# context/root span creation, so listeners can update process-wide state first. +WEB_REQUEST_STARTING = "web.request.starting" class ExecutionContext(Generic[EventType]): diff --git a/ddtrace/internal/settings/_supported_configurations.py b/ddtrace/internal/settings/_supported_configurations.py index 237d7f2cea8..4248a721d46 100644 --- a/ddtrace/internal/settings/_supported_configurations.py +++ b/ddtrace/internal/settings/_supported_configurations.py @@ -273,6 +273,7 @@ "DD_HTTPX_DISTRIBUTED_TRACING", "DD_HTTPX_SERVICE", "DD_HTTPX_SPLIT_BY_DOMAIN", + "DD_HTTP_SERVER_SERVICE", "DD_HTTP_SERVER_TAG_QUERY_STRING", "DD_IAST_DEDUPLICATE_ENABLED", "DD_IAST_DEDUPLICATION_ENABLED", @@ -577,6 +578,7 @@ "DD_TRACE_HTTPLIB_ENABLED", "DD_TRACE_HTTPX_ENABLED", "DD_TRACE_HTTP_CLIENT_TAG_QUERY_STRING", + "DD_TRACE_HTTP_SERVER_ENABLED", "DD_TRACE_HTTP_SERVER_ERROR_STATUSES", "DD_TRACE_INFERRED_PROXY_SERVICES_ENABLED", "DD_TRACE_JINJA2_ENABLED", @@ -898,6 +900,7 @@ "DD_GRPC_SERVICE": ["DD_GRPC_SERVICE_NAME"], "DD_HTTPLIB_SERVICE": ["DD_HTTPLIB_SERVICE_NAME"], "DD_HTTPX_SERVICE": ["DD_HTTPX_SERVICE_NAME"], + "DD_HTTP_SERVER_SERVICE": ["DD_HTTP_SERVER_SERVICE_NAME"], "DD_JINJA2_SERVICE": ["DD_JINJA2_SERVICE_NAME"], "DD_KAFKA_SERVICE": ["DD_KAFKA_SERVICE_NAME"], "DD_KOMBU_SERVICE": ["DD_KOMBU_SERVICE_NAME"], diff --git a/docs/integrations.rst b/docs/integrations.rst index dfe93bbe59e..625db53ae81 100644 --- a/docs/integrations.rst +++ b/docs/integrations.rst @@ -299,6 +299,12 @@ gunicorn .. automodule:: ddtrace.contrib.internal.gunicorn +.. _http_server: + +http_server +^^^^^^^^^^^ +.. automodule:: ddtrace.contrib.internal.http_server + .. _httplib: httplib diff --git a/riotfile.py b/riotfile.py index 59bf98e8912..62ea1ccd2c0 100644 --- a/riotfile.py +++ b/riotfile.py @@ -830,6 +830,14 @@ def select_pys(min_version: str = MIN_PYTHON_VERSION, max_version: str = MAX_PYT }, pys=select_pys(), ), + Venv( + name="http_server", + command="pytest {cmdargs} tests/contrib/http_server", + pkgs={ + "pytest-randomly": latest, + }, + pys=select_pys(), + ), Venv( name="logging", command="pytest -n auto --dist=worksteal {cmdargs} tests/contrib/logging", diff --git a/scripts/integration_registry/registry.yaml b/scripts/integration_registry/registry.yaml index 989a77edc13..da6b82cb91e 100644 --- a/scripts/integration_registry/registry.yaml +++ b/scripts/integration_registry/registry.yaml @@ -495,6 +495,10 @@ integrations: min: 20.0.4 max: 23.0.0 +- integration_name: http_server + is_external_package: false + is_tested: true + - integration_name: httplib is_external_package: false is_tested: true diff --git a/supported-configurations.json b/supported-configurations.json index 875df6d62ad..40cda882ccf 100644 --- a/supported-configurations.json +++ b/supported-configurations.json @@ -2067,6 +2067,16 @@ "default": "false" } ], + "DD_HTTP_SERVER_SERVICE": [ + { + "implementation": "A", + "type": "string", + "default": null, + "aliases": [ + "DD_HTTP_SERVER_SERVICE_NAME" + ] + } + ], "DD_HTTP_SERVER_TAG_QUERY_STRING": [ { "implementation": "A", @@ -4360,6 +4370,13 @@ "default": "true" } ], + "DD_TRACE_HTTP_SERVER_ENABLED": [ + { + "implementation": "A", + "type": "boolean", + "default": "true" + } + ], "DD_TRACE_HTTP_SERVER_ERROR_STATUSES": [ { "implementation": "A", diff --git a/tests/contrib/asgi/test_microvm_identity_refresh.py b/tests/contrib/asgi/test_microvm_identity_refresh.py new file mode 100644 index 00000000000..0ca6c3cf99f --- /dev/null +++ b/tests/contrib/asgi/test_microvm_identity_refresh.py @@ -0,0 +1,75 @@ +from asgiref.testing import ApplicationCommunicator +import mock +import pytest + +from ddtrace.contrib.internal.asgi.middleware import TraceMiddleware +from ddtrace.internal import core + +from .test_asgi import basic_app + + +REQUEST_STARTING_PATH = "/web-request-starting" + + +def _scope(method, path): + return { + "client": ("127.0.0.1", 32767), + "headers": [], + "method": method, + "path": path, + "query_string": b"", + "scheme": "http", + "server": ("127.0.0.1", 80), + "type": "http", + } + + +@pytest.mark.asyncio +async def test_microvm_run_hook_request(test_spans): + """TraceMiddleware.__call__() must dispatch method/path before request tracing starts. + + Django (ASGI), FastAPI, and Starlette all share this middleware, so one patch covers all + three. + """ + app = TraceMiddleware(basic_app) + instance = ApplicationCommunicator(app, _scope("POST", REQUEST_STARTING_PATH)) + + with mock.patch("ddtrace.contrib.internal.asgi.middleware.core.dispatch", wraps=core.dispatch) as m: + await instance.send_input({"type": "http.request", "body": b""}) + await instance.receive_output(1) + await instance.receive_output(1) + + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", REQUEST_STARTING_PATH)) + + +@pytest.mark.asyncio +async def test_other_request(test_spans): + app = TraceMiddleware(basic_app) + instance = ApplicationCommunicator(app, _scope("GET", "/")) + + with mock.patch("ddtrace.contrib.internal.asgi.middleware.core.dispatch", wraps=core.dispatch) as m: + await instance.send_input({"type": "http.request", "body": b""}) + await instance.receive_output(1) + await instance.receive_output(1) + + m.assert_any_call(core.WEB_REQUEST_STARTING, ("GET", "/")) + + +@pytest.mark.asyncio +async def test_sub_app_does_not_double_refresh(test_spans): + """A sub-mounted app's TraceMiddleware must not re-check a request its parent already saw + (matches the existing not-is_subapp guard around route collection/distributed headers). + """ + app = TraceMiddleware(basic_app) + scope = _scope("POST", REQUEST_STARTING_PATH) + # marks this as a sub-app request, per TraceMiddleware.__call__; request_spans matches + # the shape _on_asgi_request always creates the dict with (ddtrace/_trace/trace_handlers.py) + scope["datadog"] = {"request_spans": []} + instance = ApplicationCommunicator(app, scope) + + with mock.patch("ddtrace.contrib.internal.asgi.middleware.core.dispatch", wraps=core.dispatch) as m: + await instance.send_input({"type": "http.request", "body": b""}) + await instance.receive_output(1) + await instance.receive_output(1) + + assert not any(call.args[0] == core.WEB_REQUEST_STARTING for call in m.call_args_list) diff --git a/tests/contrib/bottle/test_microvm_identity_refresh.py b/tests/contrib/bottle/test_microvm_identity_refresh.py new file mode 100644 index 00000000000..aa1ad09864d --- /dev/null +++ b/tests/contrib/bottle/test_microvm_identity_refresh.py @@ -0,0 +1,53 @@ +import bottle +import mock +import webtest + +from ddtrace.contrib.internal.bottle.patch import TracePlugin +from ddtrace.contrib.internal.bottle.patch import patch +from ddtrace.internal import core +from tests.utils import TracerTestCase + + +REQUEST_STARTING_PATH = "/web-request-starting" + + +class BottleMicrovmIdentityRefreshTestCase(TracerTestCase): + """traced_wsgi() wraps Bottle.wsgi() -- the WSGI entry point, run before routing -- so it + emits every request's real method/path before request tracing starts. Bottle has no + unpatch(); patch() is idempotent. + """ + + def setUp(self): + super().setUp() + patch() + self.app = bottle.Bottle() + + def _trace_app(self): + self.app.install(TracePlugin(service="bottle-app", tracer=self.tracer)) + self.app = webtest.TestApp(self.app) + + def test_microvm_run_hook_request(self): + """No route is registered at the hook path: Bottle.wsgi() runs before routing, so it + must fire even on a 404 -- the stronger, more general form of this check (whether the + route matches doesn't change what gets dispatched). + """ + self._trace_app() + + with mock.patch("ddtrace.contrib.internal.bottle.trace.core.dispatch", wraps=core.dispatch) as m: + resp = self.app.post(REQUEST_STARTING_PATH, expect_errors=True) + + assert resp.status_int == 404 + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", REQUEST_STARTING_PATH)) + + def test_other_request(self): + @self.app.route("/hi/") + def hi(name): + return "hi %s" % name + + self._trace_app() + + with mock.patch("ddtrace.contrib.internal.bottle.trace.core.dispatch", wraps=core.dispatch) as m: + resp = self.app.get("/hi/dougie") + + assert resp.status_int == 200 + m.assert_any_call(core.WEB_REQUEST_STARTING, ("GET", "/hi/dougie")) diff --git a/tests/contrib/cherrypy/test_microvm_identity_refresh.py b/tests/contrib/cherrypy/test_microvm_identity_refresh.py new file mode 100644 index 00000000000..c6f566c1d64 --- /dev/null +++ b/tests/contrib/cherrypy/test_microvm_identity_refresh.py @@ -0,0 +1,51 @@ +import cherrypy +from cherrypy.test import helper +import mock + +from ddtrace.contrib.internal.cherrypy.patch import TraceMiddleware +from ddtrace.internal import core +from tests.utils import TracerTestCase + +from .web import StubApp + + +REQUEST_STARTING_PATH = "/web-request-starting" + + +class CherrypyMicrovmIdentityRefreshTestCase(TracerTestCase, helper.CPWebCase): + """TraceTool._on_start_resource() must dispatch method/path before request tracing starts. + + CherryPy has no automatic patch() -- this only fires once the app has wrapped itself in + TraceMiddleware, unlike the auto-instrumented frameworks. + """ + + @staticmethod + def setup_server(): + cherrypy.tree.mount( + StubApp(), + "/", + { + "/": {"tools.tracer.on": True}, + }, + ) + + def setUp(self): + super(CherrypyMicrovmIdentityRefreshTestCase, self).setUp() + self.traced_app = TraceMiddleware(cherrypy, service="test.cherrypy.service") + + def test_microvm_run_hook_request(self): + """No handler is registered at the hook path: _on_start_resource() still fires on the + 404 (see test_404 in test_middleware.py). + """ + with mock.patch("ddtrace.contrib.internal.cherrypy.patch.core.dispatch", wraps=core.dispatch) as m: + self.getPage(REQUEST_STARTING_PATH, method="POST") + + self.assertStatus("404 Not Found") + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", REQUEST_STARTING_PATH)) + + def test_other_request(self): + with mock.patch("ddtrace.contrib.internal.cherrypy.patch.core.dispatch", wraps=core.dispatch) as m: + self.getPage("/") + + self.assertStatus("200 OK") + m.assert_any_call(core.WEB_REQUEST_STARTING, ("GET", "/")) diff --git a/tests/contrib/django/test_microvm_identity_refresh.py b/tests/contrib/django/test_microvm_identity_refresh.py new file mode 100644 index 00000000000..59fbb2dbb89 --- /dev/null +++ b/tests/contrib/django/test_microvm_identity_refresh.py @@ -0,0 +1,26 @@ +import mock + +from ddtrace.internal import core + +REQUEST_STARTING_PATH = "/web-request-starting" + + +def test_microvm_run_hook_request(client): + """traced_get_response() must dispatch method/path before request tracing starts. + + get_response() runs before URL resolution, so this covers unmatched routes too (see + test_django_request_not_found). + """ + with mock.patch("ddtrace.contrib.internal.django.response.core.dispatch", wraps=core.dispatch) as m: + resp = client.post(REQUEST_STARTING_PATH) + + assert resp.status_code == 404 + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", REQUEST_STARTING_PATH)) + + +def test_other_request(client): + with mock.patch("ddtrace.contrib.internal.django.response.core.dispatch", wraps=core.dispatch) as m: + resp = client.get("/") + + assert resp.status_code == 200 + m.assert_any_call(core.WEB_REQUEST_STARTING, ("GET", "/")) diff --git a/tests/contrib/falcon/test_microvm_identity_refresh.py b/tests/contrib/falcon/test_microvm_identity_refresh.py new file mode 100644 index 00000000000..baa6fc5706c --- /dev/null +++ b/tests/contrib/falcon/test_microvm_identity_refresh.py @@ -0,0 +1,34 @@ +from falcon import testing +import mock + +from ddtrace.internal import core + +from .app import get_app + + +REQUEST_STARTING_PATH = "/web-request-starting" + + +def _client(): + return testing.TestClient(get_app()) + + +def test_microvm_run_hook_request(): + """process_request() must dispatch method/path before request tracing starts. + + No route is registered here: process_request() runs before resource routing, so it still + fires on the 404 (see test_404 in test_suite.py). + """ + with mock.patch("ddtrace.contrib.internal.falcon.middleware.core.dispatch", wraps=core.dispatch) as m: + response = _client().simulate_post(REQUEST_STARTING_PATH) + + assert response.status[:3] == "404" + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", REQUEST_STARTING_PATH)) + + +def test_other_request(): + with mock.patch("ddtrace.contrib.internal.falcon.middleware.core.dispatch", wraps=core.dispatch) as m: + response = _client().simulate_get("/200") + + assert response.status[:3] == "200" + m.assert_any_call(core.WEB_REQUEST_STARTING, ("GET", "/200")) diff --git a/tests/contrib/flask/test_microvm_identity_refresh.py b/tests/contrib/flask/test_microvm_identity_refresh.py new file mode 100644 index 00000000000..a248288fb5a --- /dev/null +++ b/tests/contrib/flask/test_microvm_identity_refresh.py @@ -0,0 +1,70 @@ +import mock + +from ddtrace.contrib.internal.flask.patch import patched_wsgi_app +from ddtrace.internal import core + +from . import BaseFlaskTestCase + + +REQUEST_STARTING_PATH = "/web-request-starting" + + +class FlaskMicrovmIdentityRefreshTestCase(BaseFlaskTestCase): + """patched_wsgi_app() must dispatch every request's method/path before tracing starts. + + The matching logic itself is tested in tests/tracer/runtime/test_runtime_id.py. + """ + + def test_microvm_run_hook_request(self): + """No route is registered at the hook path: wsgi_app() runs before routing, so it + must fire even on a 404 -- the stronger, more general form of this check (whether the + route matches doesn't change what gets dispatched). + """ + with mock.patch("ddtrace.contrib.internal.flask.patch.core.dispatch", wraps=core.dispatch) as m: + res = self.client.post(REQUEST_STARTING_PATH) + + self.assertEqual(res.status_code, 404) + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", REQUEST_STARTING_PATH)) + + def test_other_request(self): + @self.app.route("/") + def index(): + return "ok", 200 + + with mock.patch("ddtrace.contrib.internal.flask.patch.core.dispatch", wraps=core.dispatch) as m: + res = self.client.get("/") + + self.assertEqual(res.status_code, 200) + m.assert_any_call(core.WEB_REQUEST_STARTING, ("GET", "/")) + + def test_pre_request_event_dispatches_before_wsgi_middleware(self): + events = [] + environ = {"REQUEST_METHOD": "POST", "PATH_INFO": REQUEST_STARTING_PATH, "SCRIPT_NAME": ""} + + def start_response(status, headers, exc_info=None): + pass + + def wrapped(environ, start_response): + return [] + + def dispatch(name, args): + if name == core.WEB_REQUEST_STARTING: + events.append("starting") + + class WSGIMiddleware: + def __init__(self, app, tracer, integration_config): + pass + + def __call__(self, environ, start_response): + events.append("middleware") + return [] + + with ( + mock.patch("ddtrace.contrib.internal.flask.patch.core.dispatch", side_effect=dispatch), + mock.patch("ddtrace.contrib.internal.flask.patch._collect_routes_once"), + mock.patch("ddtrace.contrib.internal.flask.patch.is_tracing_enabled", return_value=True), + mock.patch("ddtrace.contrib.internal.flask.patch._FlaskWSGIMiddleware", WSGIMiddleware), + ): + patched_wsgi_app(wrapped, self.app, (environ, start_response), {}) + + assert events == ["starting", "middleware"] diff --git a/tests/contrib/http_server/__init__.py b/tests/contrib/http_server/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/tests/contrib/http_server/test_http_server_patch.py b/tests/contrib/http_server/test_http_server_patch.py new file mode 100644 index 00000000000..afe6df9ec51 --- /dev/null +++ b/tests/contrib/http_server/test_http_server_patch.py @@ -0,0 +1,26 @@ +from ddtrace.contrib.internal.http_server.patch import get_version +from ddtrace.contrib.internal.http_server.patch import patch +from ddtrace.contrib.internal.http_server.patch import unpatch +from tests.contrib.patch import PatchTestCase + + +class TestHttpServerPatch(PatchTestCase.Base): + __integration_name__ = "http_server" + __module_name__ = "http.server" + __patch_func__ = patch + __unpatch_func__ = unpatch + __get_version__ = get_version + + def assert_module_patched(self, http_server): + self.assert_wrapped(http_server.BaseHTTPRequestHandler.parse_request) + + def assert_not_module_patched(self, http_server): + self.assert_not_wrapped(http_server.BaseHTTPRequestHandler.parse_request) + + def assert_not_module_double_patched(self, http_server): + self.assert_not_double_wrapped(http_server.BaseHTTPRequestHandler.parse_request) + + def test_and_emit_get_version(self): + version = get_version() + assert isinstance(version, str) + assert version == "" diff --git a/tests/contrib/http_server/test_microvm_identity_refresh.py b/tests/contrib/http_server/test_microvm_identity_refresh.py new file mode 100644 index 00000000000..2879bf6a0d2 --- /dev/null +++ b/tests/contrib/http_server/test_microvm_identity_refresh.py @@ -0,0 +1,65 @@ +import http.server +import io + +import mock +import pytest + +from ddtrace.contrib.internal.http_server.patch import patch +from ddtrace.contrib.internal.http_server.patch import unpatch +from ddtrace.internal import core + +REQUEST_STARTING_PATH = "/web-request-starting" + + +def _handler_for(method, path): + """Build a BaseHTTPRequestHandler with just enough state for parse_request() to run, + bypassing the real socket/handle() loop (which would also dispatch to do_GET/do_POST). + """ + handler = http.server.BaseHTTPRequestHandler.__new__(http.server.BaseHTTPRequestHandler) + handler.raw_requestline = f"{method} {path} HTTP/1.1\r\n".encode() + handler.rfile = io.BytesIO(b"Host: localhost\r\n\r\n") + return handler + + +@pytest.fixture(autouse=True) +def _patched(): + patch() + yield + unpatch() + + +def test_microvm_run_hook_request(): + """parse_request() must dispatch method/path before request tracing starts. + + This covers apps that implement the hook with a raw http.server handler instead of a + supported web framework. + """ + with mock.patch("ddtrace.contrib.internal.http_server.patch.core.dispatch", wraps=core.dispatch) as m: + parsed = _handler_for("POST", REQUEST_STARTING_PATH).parse_request() + + assert parsed is True + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", REQUEST_STARTING_PATH)) + + +def test_other_request(): + with mock.patch("ddtrace.contrib.internal.http_server.patch.core.dispatch", wraps=core.dispatch) as m: + parsed = _handler_for("GET", "/").parse_request() + + assert parsed is True + m.assert_any_call(core.WEB_REQUEST_STARTING, ("GET", "/")) + + +def test_malformed_request_does_not_refresh(): + """A request line parse_request() can't parse must not emit the pre-request event. + + There is no method/path to report. + """ + handler = http.server.BaseHTTPRequestHandler.__new__(http.server.BaseHTTPRequestHandler) + handler.raw_requestline = b"" + handler.rfile = io.BytesIO(b"") + + with mock.patch("ddtrace.contrib.internal.http_server.patch.core.dispatch", wraps=core.dispatch) as m: + parsed = handler.parse_request() + + assert parsed is False + assert not any(call.args[0] == core.WEB_REQUEST_STARTING for call in m.call_args_list) diff --git a/tests/contrib/molten/test_microvm_identity_refresh.py b/tests/contrib/molten/test_microvm_identity_refresh.py new file mode 100644 index 00000000000..5769d2e39dc --- /dev/null +++ b/tests/contrib/molten/test_microvm_identity_refresh.py @@ -0,0 +1,46 @@ +import mock +import molten +from molten.testing import TestClient + +from ddtrace.contrib.internal.molten.patch import patch +from ddtrace.contrib.internal.molten.patch import unpatch +from ddtrace.internal import core +from tests.utils import TracerTestCase + + +REQUEST_STARTING_PATH = "/web-request-starting" + + +def greet(): + return "Greetings" + + +class MoltenMicrovmIdentityRefreshTestCase(TracerTestCase): + """patch_app_call() must dispatch method/path before request tracing starts.""" + + def setUp(self): + super().setUp() + patch() + self.app = molten.App(routes=[molten.Route("/greet", greet)]) + self.client = TestClient(self.app) + + def tearDown(self): + super().tearDown() + unpatch() + + def test_microvm_run_hook_request(self): + """No route is registered at the hook path: patch_app_call() wraps the raw WSGI entry + point, ahead of molten's router, so it still fires on the 404. + """ + with mock.patch("ddtrace.contrib.internal.molten.patch.core.dispatch", wraps=core.dispatch) as m: + response = self.client.request("POST", REQUEST_STARTING_PATH) + + self.assertEqual(response.status_code, 404) + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", REQUEST_STARTING_PATH)) + + def test_other_request(self): + with mock.patch("ddtrace.contrib.internal.molten.patch.core.dispatch", wraps=core.dispatch) as m: + response = self.client.request("GET", "/greet") + + self.assertEqual(response.status_code, 200) + m.assert_any_call(core.WEB_REQUEST_STARTING, ("GET", "/greet")) diff --git a/tests/contrib/pyramid/test_microvm_identity_refresh.py b/tests/contrib/pyramid/test_microvm_identity_refresh.py new file mode 100644 index 00000000000..60640fc46f1 --- /dev/null +++ b/tests/contrib/pyramid/test_microvm_identity_refresh.py @@ -0,0 +1,41 @@ +import mock + +from ddtrace.contrib.internal.pyramid.constants import SETTINGS_TRACE_ENABLED +from ddtrace.internal import core + +from .utils import PyramidTestCase + + +REQUEST_STARTING_PATH = "/web-request-starting" + + +class PyramidMicrovmIdentityRefreshTestCase(PyramidTestCase): + """trace_tween() must dispatch method/path before request tracing starts.""" + + def test_microvm_run_hook_request(self): + """No route is registered at the hook path: the tween sits above EXCVIEW, ahead of + route matching, so it still fires on the 404 (see test_404 in utils.py). + """ + with mock.patch("ddtrace.contrib.internal.pyramid.trace.core.dispatch", wraps=core.dispatch) as m: + self.app.post(REQUEST_STARTING_PATH, status=404) + + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", REQUEST_STARTING_PATH)) + + def test_other_request(self): + with mock.patch("ddtrace.contrib.internal.pyramid.trace.core.dispatch", wraps=core.dispatch) as m: + self.app.get("/", status=200) + + m.assert_any_call(core.WEB_REQUEST_STARTING, ("GET", "/")) + + def test_microvm_run_hook_request_with_tracing_disabled(self): + """Identity refresh is a process-wide concern, not a tracing concern: it must still + fire even when the app has tracing disabled, which used to skip installing the tween + entirely and return the undecorated handler. + """ + self.override_settings({"datadog_trace_service": "foobar", SETTINGS_TRACE_ENABLED: "false"}) + + with mock.patch("ddtrace.contrib.internal.pyramid.trace.core.dispatch", wraps=core.dispatch) as m: + self.app.post(REQUEST_STARTING_PATH, status=404) + + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", REQUEST_STARTING_PATH)) + assert len(self.pop_spans()) == 0 diff --git a/tests/contrib/sanic/test_sanic.py b/tests/contrib/sanic/test_sanic.py index 7e4a1ee0ce6..04b93f03e5f 100644 --- a/tests/contrib/sanic/test_sanic.py +++ b/tests/contrib/sanic/test_sanic.py @@ -2,6 +2,7 @@ import os import random import re +from unittest import mock import pytest from sanic import Sanic @@ -20,6 +21,7 @@ from ddtrace.constants import USER_KEEP from ddtrace.contrib.internal.sanic.patch import patch from ddtrace.contrib.internal.sanic.patch import unpatch +from ddtrace.internal import core from ddtrace.propagation import http as http_propagation from tests.conftest import DEFAULT_DDTRACE_SUBPROCESS_TEST_SERVICE_NAME from tests.tracer.utils_inferred_spans.test_helpers import assert_web_and_inferred_aws_api_gateway_span_data @@ -28,6 +30,9 @@ from tests.utils import override_http_config +REQUEST_STARTING_PATH = "/web-request-starting" + + # Helpers for handling response objects across sanic versions sanic_version = tuple(map(int, sanic_version.split("."))) @@ -439,6 +444,28 @@ async def test_endpoint_with_numeric_arg(tracer, client, test_spans): assert (await _response_text(response)) == '{"hello":42}' +@pytest.mark.asyncio +async def test_microvm_run_hook_request(tracer, client, test_spans): + """_create_sanic_request_span() (the shared entry point behind both + sanic_http_lifecycle_handle and patch_handle_request) must dispatch method/path before + request tracing starts. No route is registered here, so this also covers unmatched routes. + """ + with mock.patch("ddtrace.contrib.internal.sanic.patch.core.dispatch", wraps=core.dispatch) as m: + response = await client.post(REQUEST_STARTING_PATH) + + assert _response_status(response) in (404, 405) + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", REQUEST_STARTING_PATH)) + + +@pytest.mark.asyncio +async def test_other_request(tracer, client, test_spans): + with mock.patch("ddtrace.contrib.internal.sanic.patch.core.dispatch", wraps=core.dispatch) as m: + response = await client.get("/hello") + + assert _response_status(response) == 200 + m.assert_any_call(core.WEB_REQUEST_STARTING, ("GET", "/hello")) + + @pytest.mark.parametrize("service_name", [None, "mysvc"]) @pytest.mark.parametrize("schema_version", [None, "v0", "v1"]) def test_service_name_schematization(ddtrace_run_python_code_in_subprocess, schema_version, service_name): diff --git a/tests/contrib/suitespec.yml b/tests/contrib/suitespec.yml index 6d724daafa4..23ec6ad511e 100644 --- a/tests/contrib/suitespec.yml +++ b/tests/contrib/suitespec.yml @@ -116,6 +116,8 @@ components: - ddtrace/contrib/internal/grpc/* gunicorn: - ddtrace/contrib/internal/gunicorn/* + http_server: + - ddtrace/contrib/internal/http_server/* httplib: - ddtrace/contrib/internal/httplib/* httpx: @@ -871,6 +873,14 @@ suites: - tests/contrib/gunicorn/* - tests/snapshots/tests.contrib.gunicorn.* snapshot: true + http_server: + paths: + - '@bootstrap' + - '@core' + - '@contrib' + - '@tracing' + - '@http_server' + - tests/contrib/http_server/* httplib: paths: - '@bootstrap' diff --git a/tests/contrib/tornado/test_microvm_identity_refresh.py b/tests/contrib/tornado/test_microvm_identity_refresh.py new file mode 100644 index 00000000000..0543a47fd49 --- /dev/null +++ b/tests/contrib/tornado/test_microvm_identity_refresh.py @@ -0,0 +1,30 @@ +import mock + +from ddtrace.internal import core + +from .utils import TornadoTestCase + + +REQUEST_STARTING_PATH = "/web-request-starting" + + +class TornadoMicrovmIdentityRefreshTestCase(TornadoTestCase): + """execute() must dispatch method/path before request tracing starts.""" + + def test_microvm_run_hook_request(self): + """No handler is registered at the hook path: unmatched routes fall back to Tornado's + ErrorHandler, itself a RequestHandler, so execute() still fires (see test_404_handler + in test_tornado_web.py). + """ + with mock.patch("ddtrace.contrib.internal.tornado.handlers.core.dispatch", wraps=core.dispatch) as m: + response = self.fetch(REQUEST_STARTING_PATH, method="POST", body="") + + self.assertEqual(response.code, 404) + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", REQUEST_STARTING_PATH)) + + def test_other_request(self): + with mock.patch("ddtrace.contrib.internal.tornado.handlers.core.dispatch", wraps=core.dispatch) as m: + response = self.fetch("/success/") + + self.assertEqual(response.code, 200) + m.assert_any_call(core.WEB_REQUEST_STARTING, ("GET", "/success/")) From d424d555808684fb01f3159740a6bc56f6dfa902 Mon Sep 17 00:00:00 2001 From: Tianning Li Date: Thu, 20 Aug 2026 00:09:39 -0400 Subject: [PATCH 3/4] fix(runtime): refresh identity-bound consumers Rotating the runtime id is not enough by itself. Several long-lived components bake runtime or Remote Config client identity into native clients, workers, upload metadata, or tag caches. Have those components subscribe to explicit identity refreshes and rebuild only the state that captures those ids. Fork-specific cleanup remains on the existing fork hooks so this path does not drop buffers or invent fork lineage. --- ddtrace/appsec/_asm_request_context.py | 8 ++- ddtrace/appsec/_remoteconfiguration.py | 5 +- ddtrace/debugging/_probe/status.py | 7 ++- ddtrace/internal/core/crashtracking.py | 24 ++++++++ ddtrace/internal/native/_native.pyi | 3 + ddtrace/internal/remoteconfig/client.py | 16 +++++- ddtrace/internal/runtime/runtime_metrics.py | 17 ++++-- ddtrace/internal/settings/asm.py | 4 +- ddtrace/internal/symbol_db/symbols.py | 9 +++ ddtrace/internal/telemetry/writer.py | 16 ++++++ ddtrace/internal/writer/writer.py | 15 +++++ src/native/crashtracker.rs | 13 +++++ src/native/lib.rs | 1 + .../appsec/appsec/test_asm_request_context.py | 14 +++++ .../appsec/appsec/test_remoteconfiguration.py | 40 ++++++++++++++ tests/crashtracker/test_crashtracker.py | 40 ++++++++++++++ .../remoteconfig/test_remoteconfig_native.py | 43 +++++++++++++++ tests/internal/symbol_db/test_symbols.py | 25 +++++++++ tests/runtime/test_runtime_metrics_api.py | 22 ++++++++ tests/telemetry/test_writer.py | 55 +++++++++++++++++++ tests/tracer/test_writer.py | 33 +++++++++++ 21 files changed, 397 insertions(+), 13 deletions(-) diff --git a/ddtrace/appsec/_asm_request_context.py b/ddtrace/appsec/_asm_request_context.py index 0e2ba7ad4eb..452abe270f2 100644 --- a/ddtrace/appsec/_asm_request_context.py +++ b/ddtrace/appsec/_asm_request_context.py @@ -400,8 +400,12 @@ def finalize_asm_env(env: ASM_Environment) -> None: entry_span._set_attribute(APPSEC.EVENT_RULE_ERROR_COUNT, info.failed) except Exception: logger.debug("asm_context::finalize_asm_env::exception", extra=log_extra, exc_info=True) - if asm_config._rc_client_id is not None: - entry_span.set_tag(APPSEC.RC_CLIENT_ID, asm_config._rc_client_id) + if asm_config._rc_client_id_enabled: + from ddtrace.internal.remoteconfig.worker import remoteconfig_poller + + rc_client_id = remoteconfig_poller._client.id + if rc_client_id is not None: + entry_span.set_tag(APPSEC.RC_CLIENT_ID, rc_client_id) waf_adresses = env.waf_addresses req_headers = waf_adresses.get(SPAN_DATA_NAMES.REQUEST_HEADERS_NO_COOKIES, {}) if req_headers: diff --git a/ddtrace/appsec/_remoteconfiguration.py b/ddtrace/appsec/_remoteconfiguration.py index f7085e99d95..7fe2da28515 100644 --- a/ddtrace/appsec/_remoteconfiguration.py +++ b/ddtrace/appsec/_remoteconfiguration.py @@ -80,10 +80,13 @@ def enable_appsec_rc(callback: "AppSecCallback") -> None: if asm_config._asm_enabled: telemetry_writer.product_activated(TELEMETRY_APM_PRODUCT.APPSEC, True) - asm_config._rc_client_id = remoteconfig_poller._client.id + + asm_config._rc_client_id_enabled = True def disable_appsec_rc() -> None: + asm_config._rc_client_id_enabled = False + for product_name in APPSEC_PRODUCTS: remoteconfig_poller.unregister_callback(product_name) remoteconfig_poller.disable_product(product_name) diff --git a/ddtrace/debugging/_probe/status.py b/ddtrace/debugging/_probe/status.py index 47aeae81225..c646d0b0031 100644 --- a/ddtrace/debugging/_probe/status.py +++ b/ddtrace/debugging/_probe/status.py @@ -7,9 +7,10 @@ from ddtrace.debugging._metrics import metrics from ddtrace.debugging._probe.model import Probe from ddtrace.debugging._uploader import build_debugger_sender -from ddtrace.internal import runtime from ddtrace.internal.logger import get_logger from ddtrace.internal.native import DebuggerTrackType +from ddtrace.internal.runtime import get_ancestor_runtime_id +from ddtrace.internal.runtime import get_runtime_id from ddtrace.internal.utils.retry import fibonacci_backoff_with_jitter @@ -47,8 +48,8 @@ def _payload( "diagnostics": { "probeId": probe.probe_id, "probeVersion": probe.version, - "runtimeId": runtime.get_runtime_id(), - "parentId": runtime.get_ancestor_runtime_id(), + "runtimeId": get_runtime_id(), + "parentId": get_ancestor_runtime_id(), "status": status, } }, diff --git a/ddtrace/internal/core/crashtracking.py b/ddtrace/internal/core/crashtracking.py index 2c2f541792c..941bc39b486 100644 --- a/ddtrace/internal/core/crashtracking.py +++ b/ddtrace/internal/core/crashtracking.py @@ -14,6 +14,7 @@ from ddtrace.internal.compat import ensure_text from ddtrace.internal.logger import get_logger from ddtrace.internal.runtime import get_runtime_id +from ddtrace.internal.runtime import on_runtime_id_change from ddtrace.internal.settings import env from ddtrace.internal.settings._agent import config as agent_config from ddtrace.internal.settings.crashtracker import config as crashtracker_config @@ -33,6 +34,7 @@ from ddtrace.internal.native._native import StacktraceCollection from ddtrace.internal.native._native import crashtracker_init from ddtrace.internal.native._native import crashtracker_on_fork + from ddtrace.internal.native._native import crashtracker_reconfigure from ddtrace.internal.native._native import crashtracker_report_unhandled_exception from ddtrace.internal.native._native import crashtracker_status @@ -41,6 +43,13 @@ is_available = False +_identity_refresh_additional_tags: Optional[dict[str, str]] = None + + +def _on_identity_refresh(_new_runtime_id: str) -> None: + _reconfigure_for_identity_refresh(_identity_refresh_additional_tags) + + def _get_tags(additional_tags: Optional[dict[str, str]]) -> dict[str, str]: tags = { "language": "python", @@ -211,7 +220,20 @@ def is_started() -> bool: return crashtracker_status() == CrashtrackerStatus.Initialized +def _reconfigure_for_identity_refresh(additional_tags: Optional[dict[str, str]]) -> None: + if not is_started(): + return + + config, receiver_config, metadata = _get_args(additional_tags) + if config is None or receiver_config is None or metadata is None: + log.error("Failed to reconfigure crashtracker after identity refresh: failed to construct configuration") + return + crashtracker_reconfigure(config, receiver_config, metadata) + + def start(additional_tags: Optional[dict[str, str]] = None) -> bool: + global _identity_refresh_additional_tags + if not is_available: return False if not crashtracker_config.enabled: @@ -256,6 +278,8 @@ def crashtracker_fork_handler(): crashtracker_on_fork(config, receiver_config, metadata) forksafe.register(crashtracker_fork_handler) + _identity_refresh_additional_tags = additional_tags + on_runtime_id_change(_on_identity_refresh) except Exception: log.exception("Failed to start crashtracker") return False diff --git a/ddtrace/internal/native/_native.pyi b/ddtrace/internal/native/_native.pyi index 2c5855d60d7..654919701a4 100644 --- a/ddtrace/internal/native/_native.pyi +++ b/ddtrace/internal/native/_native.pyi @@ -121,6 +121,9 @@ def crashtracker_init( def crashtracker_on_fork( config: CrashtrackerConfiguration, receiver_config: CrashtrackerReceiverConfig, metadata: CrashtrackerMetadata ) -> None: ... +def crashtracker_reconfigure( + config: CrashtrackerConfiguration, receiver_config: CrashtrackerReceiverConfig, metadata: CrashtrackerMetadata +) -> None: ... def crashtracker_status() -> CrashtrackerStatus: ... def crashtracker_receiver() -> None: ... def crashtracker_report_unhandled_exception( diff --git a/ddtrace/internal/remoteconfig/client.py b/ddtrace/internal/remoteconfig/client.py index 93604bb71f9..3056fc1a7e3 100644 --- a/ddtrace/internal/remoteconfig/client.py +++ b/ddtrace/internal/remoteconfig/client.py @@ -9,7 +9,6 @@ import ddtrace from ddtrace.internal import gitmetadata from ddtrace.internal import process_tags -from ddtrace.internal import runtime from ddtrace.internal.hostname import get_hostname from ddtrace.internal.logger import get_logger from ddtrace.internal.packages import is_distribution_available @@ -17,6 +16,8 @@ from ddtrace.internal.remoteconfig import Payload from ddtrace.internal.remoteconfig import PayloadType from ddtrace.internal.remoteconfig import RCCallback +from ddtrace.internal.runtime import get_runtime_id +from ddtrace.internal.runtime import on_runtime_id_change from ddtrace.internal.settings._agent import config as agent_config from ddtrace.internal.settings._core import DDConfig from ddtrace.internal.telemetry import telemetry_writer @@ -98,6 +99,8 @@ def __init__(self) -> None: self._native: Optional[Any] = None self._reader: Optional[Any] = None + on_runtime_id_change(self._on_identity_refresh) + def ensure_native(self) -> Any: if self._native is None: from ddtrace.internal.native import RemoteConfigClient as _NativeClient @@ -109,7 +112,7 @@ def ensure_native(self) -> Any: agent_url=str(self.agent_url), tracer_version=tracer_version, client_id=self.id, - runtime_id=runtime.get_runtime_id(), + runtime_id=get_runtime_id(), service=ddtrace.config.service or "", env=ddtrace.config.env or "", app_version=ddtrace.config.version or "", @@ -125,6 +128,15 @@ def ensure_native(self) -> Any: def renew_id(self) -> None: self.id = str(uuid.uuid4()) + def _on_identity_refresh(self, new_runtime_id: str) -> None: + # Regenerate the client id and drop the native client, which bakes both ids in + # as immutable constructor arguments (get_client_id() is documented "stable for + # the process lifetime"). The next ensure_native() call rebuilds it bound to the + # fresh ids. Safe across threads: request() captures self._native into a local + # before calling .poll(), so an in-flight poll on the old client is unaffected. + self.renew_id() + self._native = None + def register_callback(self, product_name: "RemoteConfigProduct", callback: RCCallback) -> None: self._product_callbacks[product_name] = callback log.debug("[%s][P: %s] Registered callback for product %s", os.getpid(), os.getppid(), product_name) diff --git a/ddtrace/internal/runtime/runtime_metrics.py b/ddtrace/internal/runtime/runtime_metrics.py index caa095c9cdf..638f5cf5aa0 100644 --- a/ddtrace/internal/runtime/runtime_metrics.py +++ b/ddtrace/internal/runtime/runtime_metrics.py @@ -4,6 +4,7 @@ from ddtrace.internal import atexit from ddtrace.internal.constants import EXPERIMENTAL_FEATURES +from ddtrace.internal.runtime import on_runtime_id_change from ddtrace.internal.settings._agent import config as agent_config from ddtrace.internal.settings._config import config from ddtrace.internal.threads import Lock @@ -92,14 +93,22 @@ def __init__(self, interval=DEFAULT_RUNTIME_METRICS_INTERVAL, dogstatsd_url=None else: self.send_metric = self._dogstatsd_client.distribution + self._platform_tags = self._build_platform_tags() if config._runtime_metrics_runtime_id_enabled: - # Enables tagging runtime metrics with runtime-id (as well as all the v1 tags) - self._platform_tags = self._format_tags(PlatformTagsV2()) - else: - self._platform_tags = self._format_tags(PlatformTags()) + # refresh ids to ensure the tags are up to date upon MicroVM instance starts. + on_runtime_id_change(self._on_identity_refresh) self._process_tags: list[str] = list(ProcessTags()) + def _build_platform_tags(self) -> list[str]: + if config._runtime_metrics_runtime_id_enabled: + # Enables tagging runtime metrics with runtime-id (as well as all the v1 tags) + return self._format_tags(PlatformTagsV2()) + return self._format_tags(PlatformTags()) + + def _on_identity_refresh(self, _new_runtime_id: str) -> None: + self._platform_tags = self._build_platform_tags() + @classmethod def disable(cls) -> None: with cls._lock: diff --git a/ddtrace/internal/settings/asm.py b/ddtrace/internal/settings/asm.py index c129617c3f8..cba47f46631 100644 --- a/ddtrace/internal/settings/asm.py +++ b/ddtrace/internal/settings/asm.py @@ -268,7 +268,9 @@ class ASMConfig(DDConfig): sys.platform.startswith("win") or sys.platform.startswith("cygwin") ) - _rc_client_id: Optional[str] = None + # Set by enable_appsec_rc()/disable_appsec_rc(); gates _dd.rc.client_id span tagging so it's + # only emitted while AppSec RC is actually enabled, not just whenever a live RC client exists. + _rc_client_id_enabled: bool = False def __init__(self): super().__init__() diff --git a/ddtrace/internal/symbol_db/symbols.py b/ddtrace/internal/symbol_db/symbols.py index b8b9daa35e6..39cfecd6753 100644 --- a/ddtrace/internal/symbol_db/symbols.py +++ b/ddtrace/internal/symbol_db/symbols.py @@ -38,6 +38,7 @@ from ddtrace.internal.periodic import Timer from ddtrace.internal.runtime import get_ancestor_runtime_id from ddtrace.internal.runtime import get_runtime_id +from ddtrace.internal.runtime import on_runtime_id_change from ddtrace.internal.safety import _isinstance from ddtrace.internal.settings._agent import config as agent_config from ddtrace.internal.settings.dynamic_instrumentation import config as di_config @@ -561,6 +562,9 @@ def __init__(self, scopes: t.Optional[list[Scope]] = None) -> None: } forksafe.register(self._reset_on_fork) + # Same rebuild, triggered by an explicit identity refresh (e.g. an AWS Lambda + # MicroVM /run hook) rather than an actual fork. + on_runtime_id_change(self._on_identity_refresh) @cached_property def _sender(self) -> SymDBSender: @@ -577,6 +581,11 @@ def _reset_on_fork(self) -> None: self._event_data["runtimeId"] = get_runtime_id() self._event_data["parentId"] = get_ancestor_runtime_id() + def _on_identity_refresh(self, new_runtime_id: str) -> None: + # Same rebuild as _reset_on_fork(): runtimeId is baked into _event_data, so it must be + # refreshed here too or every batch keeps reporting the pre-refresh snapshot's ID. + self._reset_on_fork() + def _set_timer(self) -> None: with self._timer_lock: if self._timer is None: diff --git a/ddtrace/internal/telemetry/writer.py b/ddtrace/internal/telemetry/writer.py index 31f89b3a445..146e94c75a5 100644 --- a/ddtrace/internal/telemetry/writer.py +++ b/ddtrace/internal/telemetry/writer.py @@ -23,6 +23,7 @@ from ..runtime import get_ancestor_runtime_id from ..runtime import get_parent_runtime_id from ..runtime import get_runtime_id +from ..runtime import on_runtime_id_change from ..utils.formats import get_test_session_token from ..utils.version import version as tracer_version from .constants import TELEMETRY_APM_PRODUCT @@ -218,6 +219,9 @@ def __init__(self, agentless: Optional[bool] = None) -> None: # runtime's after_fork_child hook, which get_native_runtime() registered # during enable(), so the shared runtime is restarted before we rebuild). forksafe.register(self._fork_writer) + # Same rebuild, triggered by an explicit identity refresh (e.g. an AWS Lambda + # MicroVM /run hook) rather than an actual fork. + on_runtime_id_change(self._on_identity_refresh) get_logger("ddtrace").addHandler(DDTelemetryErrorHandler(self)) def _build_worker(self) -> "TelemetryWorker": @@ -929,6 +933,18 @@ def _fork_writer(self) -> None: # Re-discover dependencies from scratch so the child reports its own imports. self._dependency_tracker.reset() + def _on_identity_refresh(self, new_runtime_id: str) -> None: + # Same rebuild as _fork_writer(): the native worker bakes in get_runtime_id() at + # construction, so it must be dropped and lazily rebuilt on the next telemetry call. + # Unlike after a fork, the worker is still alive here -- it must be stopped (not just + # dropped) or it keeps heartbeating with the stale runtime ID until process shutdown. + if self._worker is not None: + try: + self._worker.stop(send_app_closing=False) + except Exception: + log.debug("Failed to stop the native telemetry worker during identity refresh", exc_info=True) + self._fork_writer() + def _telemetry_excepthook(self, tp, value, root_traceback) -> None: if root_traceback is not None: # Get the frame which raised the exception diff --git a/ddtrace/internal/writer/writer.py b/ddtrace/internal/writer/writer.py index 166c1e7d8a9..c4884ac4e68 100644 --- a/ddtrace/internal/writer/writer.py +++ b/ddtrace/internal/writer/writer.py @@ -19,6 +19,7 @@ from ddtrace.internal.native._native import SpanData from ddtrace.internal.native_runtime import get_native_runtime from ddtrace.internal.runtime import get_runtime_id +from ddtrace.internal.runtime import on_runtime_id_change from ddtrace.internal.settings import env from ddtrace.internal.settings._agent import config as agent_config from ddtrace.internal.settings._config import config @@ -853,6 +854,7 @@ def __init__( self._stats_opt_out = stats_opt_out self._exporter = self._create_exporter() + on_runtime_id_change(self._on_identity_refresh) @staticmethod def _parse_otlp_headers(raw: str) -> list: @@ -956,6 +958,19 @@ def set_test_session_token(self, token: Optional[str]) -> None: except Exception: _safelog(log.warning, "failed to shutdown exporter", exc_info=True) + def _on_identity_refresh(self, new_runtime_id: str) -> None: + # Rebuild the exporter so it picks up the new runtime_id (baked in at construction + # via enable_telemetry()), without touching the span buffer: unlike a fork, no spans + # were lost, so anything already buffered should still flush once the new exporter's + # connection is up. Modeled on set_test_session_token(), not recreate()/fork, which + # replace the whole writer and drop the buffer. + old_exporter = self._exporter + self._exporter = self._create_exporter() + try: + old_exporter.shutdown(3_000_000_000) + except Exception: + _safelog(log.warning, "failed to shutdown exporter", exc_info=True) + def recreate( self, appsec_enabled: Optional[bool] = None, diff --git a/src/native/crashtracker.rs b/src/native/crashtracker.rs index e2fb8149de7..ad1a35ce6fb 100644 --- a/src/native/crashtracker.rs +++ b/src/native/crashtracker.rs @@ -320,6 +320,19 @@ pub fn crashtracker_on_fork<'py>( libdd_crashtracker::on_fork(inner_config, inner_receiver_config, inner_metadata) } +#[pyfunction(name = "crashtracker_reconfigure")] +pub fn crashtracker_reconfigure<'py>( + mut config: PyRefMut<'py, CrashtrackerConfigurationPy>, + mut receiver_config: PyRefMut<'py, CrashtrackerReceiverConfigPy>, + mut metadata: PyRefMut<'py, CrashtrackerMetadataPy>, +) -> anyhow::Result<()> { + let inner_config = (*config).take_inner_or_err()?; + let inner_receiver_config = (*receiver_config).take_inner_or_err()?; + let inner_metadata = (*metadata).take_inner_or_err()?; + + libdd_crashtracker::reconfigure(inner_config, inner_receiver_config, inner_metadata) +} + #[pyfunction(name = "crashtracker_status")] pub fn crashtracker_status() -> anyhow::Result { CrashtrackerStatus::try_from(CRASHTRACKER_STATUS.load(Ordering::SeqCst)) diff --git a/src/native/lib.rs b/src/native/lib.rs index 9b16a4d07ae..236958858a6 100644 --- a/src/native/lib.rs +++ b/src/native/lib.rs @@ -54,6 +54,7 @@ fn _native(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_function(wrap_pyfunction!(crashtracker::crashtracker_init, m)?)?; m.add_function(wrap_pyfunction!(crashtracker::crashtracker_on_fork, m)?)?; + m.add_function(wrap_pyfunction!(crashtracker::crashtracker_reconfigure, m)?)?; m.add_function(wrap_pyfunction!(crashtracker::crashtracker_status, m)?)?; m.add_function(wrap_pyfunction!(crashtracker::crashtracker_receiver, m)?)?; m.add_function(wrap_pyfunction!( diff --git a/tests/appsec/appsec/test_asm_request_context.py b/tests/appsec/appsec/test_asm_request_context.py index c42a8768652..94241ee87b5 100644 --- a/tests/appsec/appsec/test_asm_request_context.py +++ b/tests/appsec/appsec/test_asm_request_context.py @@ -14,6 +14,20 @@ config_asm = {"_asm_enabled": True} +@pytest.mark.parametrize("auto_enable_crashtracking", [False]) +def test_import_does_not_load_remoteconfig_worker(run_python_code_in_subprocess, auto_enable_crashtracking): + code = """ +import sys + +import ddtrace.appsec._asm_request_context # noqa: F401 + +assert "ddtrace.internal.remoteconfig.worker" not in sys.modules +""" + + _, stderr, status, _ = run_python_code_in_subprocess(code) + assert status == 0, stderr + + def test_context_set_and_reset(): with asm_context( ip_addr=_TEST_IP, diff --git a/tests/appsec/appsec/test_remoteconfiguration.py b/tests/appsec/appsec/test_remoteconfiguration.py index f0c7f66dd17..c169626ebba 100644 --- a/tests/appsec/appsec/test_remoteconfiguration.py +++ b/tests/appsec/appsec/test_remoteconfiguration.py @@ -270,6 +270,46 @@ def test_rc_activation_validate_client_id(tracer, rc_poller, appsec_callback): disable_appsec_rc() +def test_rc_client_id_tag_reflects_live_value_not_a_stale_cache(tracer, rc_poller, appsec_callback): + """_dd.rc.client_id must be read live at span-tagging time, not cached once at RC enable time. + + Otherwise the tag would go stale after e.g. an AWS Lambda MicroVM identity refresh + regenerates the real client id. + """ + from ddtrace.internal.remoteconfig.worker import remoteconfig_poller + + with override_global_config(dict(_asm_enabled=True, _remote_config_enabled=True, api_version="v0.4")): + tracer.configure(appsec_enabled=True) + enable_appsec_rc(appsec_callback) + + with mock.patch.object(remoteconfig_poller._client, "id", "client-id-one"): + with asm_context(tracer) as span: + set_http_meta(span, {}, raw_uri="http://example.com/", status_code="200") + assert span._local_root._get_str_attribute(APPSEC.RC_CLIENT_ID) == "client-id-one" + + with mock.patch.object(remoteconfig_poller._client, "id", "client-id-two"): + with asm_context(tracer) as span: + set_http_meta(span, {}, raw_uri="http://example.com/", status_code="200") + assert span._local_root._get_str_attribute(APPSEC.RC_CLIENT_ID) == "client-id-two" + disable_appsec_rc() + + +def test_rc_client_id_tag_not_set_when_rc_disabled(tracer): + """_dd.rc.client_id must not be tagged when AppSec RC was never enabled, even though a live + RC client (and id) exists process-wide -- otherwise every ASM-tracked span would get tagged + with an id from a Remote Config subscription AppSec never activated. + """ + from ddtrace.internal.remoteconfig.worker import remoteconfig_poller + + with override_global_config(dict(_asm_enabled=True, api_version="v0.4")): + tracer.configure(appsec_enabled=True) + + with mock.patch.object(remoteconfig_poller._client, "id", "client-id-one"): + with asm_context(tracer) as span: + set_http_meta(span, {}, raw_uri="http://example.com/", status_code="200") + assert span._local_root._get_str_attribute(APPSEC.RC_CLIENT_ID) is None + + @pytest.mark.parametrize( "env_rules, expected", [ diff --git a/tests/crashtracker/test_crashtracker.py b/tests/crashtracker/test_crashtracker.py index 2c23bba2037..677c6aac461 100644 --- a/tests/crashtracker/test_crashtracker.py +++ b/tests/crashtracker/test_crashtracker.py @@ -98,6 +98,46 @@ def test_crashtracker_started(): pytest.fail("contents of stdout.log: %s, stderr.log: %s" % (stdout_msg, stderr_msg)) +@pytest.mark.skipif(not sys.platform.startswith("linux"), reason="Linux only") +@pytest.mark.subprocess(err=None) +@pytest.mark.parametrize("auto_enable_crashtracking", [False]) +def test_crashtracker_identity_refresh_reconfigures_metadata(): + from contextlib import ExitStack + + import mock + + from ddtrace.internal.core import crashtracking + import ddtrace.internal.runtime as runtime + + init_args = (object(), object(), object()) + refresh_args = (object(), object(), object()) + tags = {"service": "identity-refresh"} + initialized = object() + status = type("CrashtrackerStatus", (), {"Initialized": initialized}) + get_args = mock.Mock(side_effect=[init_args, refresh_args]) + init = mock.Mock() + reconfigure = mock.Mock() + + with ExitStack() as stack: + stack.enter_context(mock.patch.object(crashtracking, "is_available", True)) + stack.enter_context(mock.patch.object(crashtracking, "crashtracker_config", mock.Mock(enabled=True))) + stack.enter_context(mock.patch.object(crashtracking, "CrashtrackerStatus", status, create=True)) + stack.enter_context(mock.patch.object(crashtracking, "_identity_refresh_additional_tags", None)) + stack.enter_context(mock.patch.object(crashtracking, "_get_args", get_args)) + stack.enter_context(mock.patch.object(crashtracking, "crashtracker_init", init, create=True)) + stack.enter_context(mock.patch.object(crashtracking, "crashtracker_reconfigure", reconfigure, create=True)) + stack.enter_context( + mock.patch.object(crashtracking, "crashtracker_status", mock.Mock(return_value=initialized), create=True) + ) + + assert crashtracking.start(tags) + runtime.refresh_identity() + + assert get_args.call_args_list == [mock.call(tags), mock.call(tags)] + init.assert_called_once_with(*init_args) + reconfigure.assert_called_once_with(*refresh_args) + + @pytest.mark.skipif(not sys.platform.startswith("linux"), reason="Linux only") @pytest.mark.subprocess() def test_crashtracker_receiver_not_in_path(): diff --git a/tests/internal/remoteconfig/test_remoteconfig_native.py b/tests/internal/remoteconfig/test_remoteconfig_native.py index f1813266321..ff646a167f2 100644 --- a/tests/internal/remoteconfig/test_remoteconfig_native.py +++ b/tests/internal/remoteconfig/test_remoteconfig_native.py @@ -418,3 +418,46 @@ def test_enable_builds_native_runtime_before_registering_fork_hook(monkeypatch): assert poller.enable() is True assert order == ["native", "before_fork", "start"], order + + +def test_identity_refresh_renews_client_id_and_drops_native(): + # get_client_id() on the native client is documented "stable for the process lifetime", + # so refreshing must drop it (not mutate it in place) for the id to actually change. + client = RemoteConfigClient() + old_id = client.id + client.ensure_native() + assert client._native is not None + + client._on_identity_refresh("some-new-runtime-id") + + assert client.id != old_id + assert client._native is None + + +def test_identity_refresh_rebuilds_native_client_with_fresh_id(): + client = RemoteConfigClient() + native_before = client.ensure_native() + old_native_client_id = native_before.get_client_id() + + client._on_identity_refresh("some-new-runtime-id") + native_after = client.ensure_native() + + assert native_after is not native_before + assert native_after.get_client_id() == client.id + assert native_after.get_client_id() != old_native_client_id + + +@pytest.mark.subprocess +def test_identity_refresh_wired_to_runtime_id_change(): + """A RemoteConfigClient subscribes itself at construction; refresh_identity() reaches it.""" + from ddtrace.internal import runtime + from ddtrace.internal.remoteconfig.client import RemoteConfigClient + + client = RemoteConfigClient() + old_id = client.id + client.ensure_native() + + runtime.refresh_identity() + + assert client.id != old_id + assert client._native is None diff --git a/tests/internal/symbol_db/test_symbols.py b/tests/internal/symbol_db/test_symbols.py index c1c5b075312..920ef5cba70 100644 --- a/tests/internal/symbol_db/test_symbols.py +++ b/tests/internal/symbol_db/test_symbols.py @@ -526,6 +526,31 @@ def test_symbols_fork_uploads(): assert os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0, f"child {pid} exited with status {status}" +@pytest.mark.subprocess(ddtrace_run=True, err=None) +def test_symbols_identity_refresh_updates_runtime_id(): + """A non-fork identity refresh (e.g. an AWS Lambda MicroVM /run hook) must also refresh the + ScopeContext's cached runtimeId, the same way _reset_on_fork() does after an actual fork -- + otherwise every upload after a MicroVM /run keeps reporting the pre-refresh snapshot's ID. + """ + import typing as t + + import ddtrace.internal.runtime as runtime + from ddtrace.internal.symbol_db.symbols import SymbolDatabaseUploader + + SymbolDatabaseUploader.install() + + context = t.cast(SymbolDatabaseUploader, SymbolDatabaseUploader._instance)._context + old_runtime_id = runtime.get_runtime_id() + old_upload_id = context._upload_id + assert context._event_data["runtimeId"] == old_runtime_id + + runtime.refresh_identity() + + assert runtime.get_runtime_id() != old_runtime_id + assert context._event_data["runtimeId"] == runtime.get_runtime_id() + assert context._upload_id != old_upload_id + + @pytest.mark.subprocess(ddtrace_run=True, err=None) def test_symbols_fork_forces_reenable_and_install(): """ diff --git a/tests/runtime/test_runtime_metrics_api.py b/tests/runtime/test_runtime_metrics_api.py index 963d8deeae9..b0e19778190 100644 --- a/tests/runtime/test_runtime_metrics_api.py +++ b/tests/runtime/test_runtime_metrics_api.py @@ -232,6 +232,28 @@ def test_runtime_metrics_experimental_runtime_tag(): ) +@pytest.mark.subprocess(env={"DD_RUNTIME_METRICS_RUNTIME_ID_ENABLED": "true"}, err=None) +def test_runtime_metrics_runtime_id_tag_refreshes_on_identity_refresh(): + from ddtrace.internal import runtime + from ddtrace.internal.runtime.runtime_metrics import RuntimeWorker + + try: + RuntimeWorker.enable() + assert RuntimeWorker._instance is not None + + worker_instance = RuntimeWorker._instance + runtime_id_tag = f"runtime-id:{runtime.get_runtime_id()}" + assert runtime_id_tag in worker_instance._platform_tags, worker_instance._platform_tags + + runtime.refresh_identity() + + refreshed_runtime_id_tag = f"runtime-id:{runtime.get_runtime_id()}" + assert refreshed_runtime_id_tag in worker_instance._platform_tags, worker_instance._platform_tags + assert runtime_id_tag not in worker_instance._platform_tags, worker_instance._platform_tags + finally: + RuntimeWorker.disable() + + @pytest.mark.subprocess( parametrize={"DD_TRACE_EXPERIMENTAL_FEATURES_ENABLED": ["DD_RUNTIME_METRICS_ENABLED,someotherfeature", ""]}, err=None, diff --git a/tests/telemetry/test_writer.py b/tests/telemetry/test_writer.py index 7bbede8e4c6..ce5b71a4c98 100644 --- a/tests/telemetry/test_writer.py +++ b/tests/telemetry/test_writer.py @@ -570,6 +570,61 @@ def test_telemetry_writer_agent_setup(): assert new_telemetry_writer._agentless is False +def test_identity_refresh_rebuilds_native_worker(): + """Same rebuild as after a fork: the native worker bakes in get_runtime_id() at construction.""" + with override_global_config( + {"_dd_site": "datad0g.com", "_dd_api_key": "foobarkey", "_ci_visibility_agentless_enabled": False} + ): + writer = ddtrace.internal.telemetry.TelemetryWriter(agentless=False) + assert writer._worker is not None + + writer._on_identity_refresh("some-new-runtime-id") + + assert writer._worker is None + assert writer.started is False + + +def test_identity_refresh_stops_live_worker_before_dropping(): + """Unlike after a fork, the worker is still alive here and must be explicitly stopped, or it + keeps heartbeating with the stale runtime ID until process shutdown. + """ + with override_global_config( + {"_dd_site": "datad0g.com", "_dd_api_key": "foobarkey", "_ci_visibility_agentless_enabled": False} + ): + writer = ddtrace.internal.telemetry.TelemetryWriter(agentless=False) + assert writer._worker is not None + + # TelemetryWorker is a native extension type -- its methods can't be patched in place, + # so swap in a mock to observe the stop() call instead. + fake_worker = mock.Mock() + writer._worker = fake_worker + + writer._on_identity_refresh("some-new-runtime-id") + + fake_worker.stop.assert_called_once_with(send_app_closing=False) + assert writer._worker is None + + +@pytest.mark.subprocess( + env={"DD_SITE": "datad0g.com", "DD_API_KEY": "foobarkey", "DD_CIVISIBILITY_AGENTLESS_ENABLED": "false"} +) +def test_identity_refresh_wired_to_runtime_id_change(): + """Drives the refresh through runtime.refresh_identity() instead of calling + _on_identity_refresh directly (as the test above does), so a dropped + on_runtime_id_change() subscription would actually fail this. + """ + from ddtrace.internal import runtime + import ddtrace.internal.telemetry + + writer = ddtrace.internal.telemetry.TelemetryWriter(agentless=False) + assert writer._worker is not None + + runtime.refresh_identity() + + assert writer._worker is None + assert writer.started is False + + @pytest.mark.parametrize( "env_agentless,arg_agentless", [ diff --git a/tests/tracer/test_writer.py b/tests/tracer/test_writer.py index a13958ba833..eaf60adb565 100644 --- a/tests/tracer/test_writer.py +++ b/tests/tracer/test_writer.py @@ -418,6 +418,20 @@ def test_on_shutdown_before_start(self): # Call shutdown without ever calling start() writer.on_shutdown() + def test_identity_refresh_rebuilds_exporter_without_recreating_writer(self): + """Same trigger as an AWS Lambda MicroVM /run hook: rebuild the exporter (it bakes in + get_runtime_id() at construction) without touching the writer/buffer, unlike recreate() + (used on fork), which replaces the whole writer and drops anything already written. + """ + writer = NativeWriter("http://dne:1234") + old_exporter = writer._exporter + old_clients = writer._clients + + writer._on_identity_refresh("some-new-runtime-id") + + assert writer._exporter is not old_exporter + assert writer._clients is old_clients + # Http related metrics are sent by the native code def test_drop_reason_bad_endpoint(self): pytest.skip() @@ -427,6 +441,25 @@ def test_gzip_compression_exception_logging_and_metrics(self): pytest.skip() +@pytest.mark.subprocess +def test_native_writer_identity_refresh_wired_to_runtime_id_change(): + """Drives the refresh through runtime.refresh_identity() instead of calling + _on_identity_refresh directly (as the test above does), so a dropped + on_runtime_id_change() subscription would actually fail this. + """ + from ddtrace.internal import runtime + from ddtrace.internal.writer import NativeWriter + + writer = NativeWriter("http://dne:1234") + old_exporter = writer._exporter + old_clients = writer._clients + + runtime.refresh_identity() + + assert writer._exporter is not old_exporter + assert writer._clients is old_clients + + class CIVisibilityWriterTests(NativeWriterTests): WRITER_CLASS = CIVisibilityWriter From 621c07416d0add9ddfba6981a0d39ae008841281 Mon Sep 17 00:00:00 2001 From: Tianning Li Date: Thu, 20 Aug 2026 00:10:29 -0400 Subject: [PATCH 4/4] feat(aws-lambda-microvm): refresh identity on run hook AWS Lambda MicroVM instances restored from the same image start with the same in-memory runtime id and Remote Config client id. The platform's /run lifecycle request is the earliest common signal that a restored instance is becoming active. Register the web request listener only inside MicroVM images, match the fixed POST /run hook, and refresh identity once per process. Stacked web layers can observe the same request without rotating identity more than once. --- ddtrace/internal/runtime/__init__.py | 94 ++++++++++-- ...ovm-identity-refresh-3a672cd6bcbad16d.yaml | 7 + .../asgi/test_microvm_identity_refresh.py | 10 +- .../bottle/test_microvm_identity_refresh.py | 8 +- .../cherrypy/test_microvm_identity_refresh.py | 8 +- .../django/test_microvm_identity_refresh.py | 7 +- .../falcon/test_microvm_identity_refresh.py | 8 +- .../flask/test_microvm_identity_refresh.py | 10 +- .../test_microvm_identity_refresh.py | 7 +- .../molten/test_microvm_identity_refresh.py | 8 +- .../pyramid/test_microvm_identity_refresh.py | 12 +- tests/contrib/sanic/test_sanic.py | 8 +- .../tornado/test_microvm_identity_refresh.py | 8 +- tests/tracer/runtime/test_runtime_id.py | 138 +++++++++++++++++- 14 files changed, 258 insertions(+), 75 deletions(-) create mode 100644 releasenotes/notes/aws-lambda-microvm-identity-refresh-3a672cd6bcbad16d.yaml diff --git a/ddtrace/internal/runtime/__init__.py b/ddtrace/internal/runtime/__init__.py index 0cd60046b6c..10e2818558e 100644 --- a/ddtrace/internal/runtime/__init__.py +++ b/ddtrace/internal/runtime/__init__.py @@ -1,3 +1,4 @@ +import threading import typing as t import uuid import weakref @@ -17,6 +18,8 @@ "get_parent_runtime_id", "get_runtime_propagation_envs", "refresh_identity", + "maybe_refresh_identity", + "listen_for_identity_refresh_hooks", ] @@ -43,14 +46,19 @@ def _generate_runtime_id() -> str: def on_runtime_id_change(cb: t.Callable[[str], None]) -> None: - """Register a callback to be called when refresh_identity() runs. - - refresh_identity() is the non-fork trigger for a new logical process - instance. It is deliberately not called after a plain fork: forked children - already get a fresh runtime ID silently (see _set_runtime_id()), and code - that needs to react to a fork specifically should use forksafe.register(). - Only a weak reference to cb is kept, so the caller must keep it alive for - it to keep firing. + """Register a callback to be called when ``refresh_identity()`` runs. + + ``refresh_identity()`` is the non-fork trigger (e.g. an AWS Lambda MicroVM + ``/run`` hook) for a new logical process instance. It is deliberately not + called after a plain fork: forked children already get a fresh runtime ID + silently (see ``_set_runtime_id()``), and code that needs to react to a + fork specifically should use ``forksafe.register()`` instead -- unlike a + fork, no state (spans, SHM-backed native clients, ...) was inherited from + a different process here, so subscribers generally only need to rebuild + what bakes the runtime/client id in at construction, not reset buffers or + handles the way a fork hook would. Only a weak reference to ``cb`` is + kept, so the caller must keep it alive (e.g. by registering a bound + method of a long-lived object) for it to keep firing. """ global _ON_RUNTIME_ID_CHANGE try: @@ -83,8 +91,8 @@ def _notify_runtime_id_subscribers() -> None: try: cb(_RUNTIME_ID) except Exception: - # One broken subscriber must not prevent other subscribers from seeing the - # refreshed runtime ID. + # This can run on a web framework's request-dispatch path (maybe_refresh_identity()); + # one broken subscriber must not take down the request or block the others. log.debug("Error notifying on_runtime_id_change() subscriber", exc_info=True) _ON_RUNTIME_ID_CHANGE -= dead @@ -108,16 +116,72 @@ def _set_runtime_id() -> None: def refresh_identity() -> None: """Regenerate the runtime ID without recording fork lineage. - Unlike a fork, this does not update _PARENT_RUNTIME_ID / _ANCESTOR_RUNTIME_ID: - the previous runtime ID was not a real parent process, so recording it there - would make get_process_role() and friends misreport a fork lineage that never - existed. Use this when a new logical process instance is created by a mechanism - other than fork(). + Unlike a fork, this does not update ``_PARENT_RUNTIME_ID`` / + ``_ANCESTOR_RUNTIME_ID``: the previous runtime ID was not a real parent + process, so recording it there would make ``get_process_role()`` and + friends misreport a fork lineage that never existed. Use this when a new + logical process instance is created by a mechanism other than ``fork()`` + -- e.g. an AWS Lambda MicroVM instance launched from a shared image + snapshot. """ _regenerate_runtime_id() _notify_runtime_id_subscribers() +# Fixed platform path for the AWS Lambda MicroVM "/run" lifecycle hook. Never the "/resume" +# hook path -- see refresh_identity()'s docstring for why. Exported (no leading underscore) +# so tests can reference the same constant instead of duplicating the literal. +MICROVM_RUN_HOOK_METHOD = "POST" +MICROVM_RUN_HOOK_PATH = "/aws/lambda-microvms/runtime/v1/run" + +# Same env var #18017 uses to detect a MicroVM at runtime (rand64bits() OS-entropy fallback). +# Read once at import, like that fix does, so listener registration is skipped outside a +# MicroVM. This exact method+path is otherwise just an unauthenticated trigger on every +# ddtrace user's request-dispatch path, MicroVM or not. +_IS_AWS_LAMBDA_MICROVM = env.get("AWS_LAMBDA_MICROVM_IMAGE_ARN") is not None +# Multiple instrumented request layers can observe the same MicroVM /run hook +# (for example, Werkzeug's http.server layer plus Flask). Refresh identity once +# per process so a single logical MicroVM instance gets one runtime-id rotation. +_IDENTITY_REFRESH_HOOK_REFRESHED = threading.Event() +_IDENTITY_REFRESH_HOOK_REFRESH_LOCK = threading.Lock() + + +def listen_for_identity_refresh_hooks() -> None: + """Refresh MicroVM identity from request events emitted before root span creation.""" + if not _IS_AWS_LAMBDA_MICROVM: + return + + from ddtrace.internal import core + + core.on(core.WEB_REQUEST_STARTING, maybe_refresh_identity) + + +def maybe_refresh_identity(method: t.Optional[str], path: t.Optional[str]) -> None: + """Call refresh_identity() if this request is the AWS Lambda MicroVM "/run" hook. + + Called from every instrumented web framework's request-dispatch patch in a MicroVM + with that request's method and path, so the platform-defined hook path only needs + to be matched in one place. + ``method``/``path`` may be ``None`` -- some callers read them straight off a raw + request/environ mapping (e.g. ``environ.get("REQUEST_METHOD")``) that has no guarantee + either key is present. + """ + if not method or not path: + return + if method != MICROVM_RUN_HOOK_METHOD or path != MICROVM_RUN_HOOK_PATH: + return + + with _IDENTITY_REFRESH_HOOK_REFRESH_LOCK: + if _IDENTITY_REFRESH_HOOK_REFRESHED.is_set(): + return + _IDENTITY_REFRESH_HOOK_REFRESHED.set() + + refresh_identity() + + +listen_for_identity_refresh_hooks() + + def get_runtime_id() -> str: """Return a unique string identifier for this runtime. diff --git a/releasenotes/notes/aws-lambda-microvm-identity-refresh-3a672cd6bcbad16d.yaml b/releasenotes/notes/aws-lambda-microvm-identity-refresh-3a672cd6bcbad16d.yaml new file mode 100644 index 00000000000..1158e079966 --- /dev/null +++ b/releasenotes/notes/aws-lambda-microvm-identity-refresh-3a672cd6bcbad16d.yaml @@ -0,0 +1,7 @@ +--- +fixes: + - | + stable id uniqueness: Regenerates stable identifiers, including the runtime ID and + Remote Config client ID, when a new AWS Lambda MicroVM instance starts to ensure + they remain unique. This behavior only applies to AWS Lambda MicroVM deployments; + behavior in other environments is unchanged. diff --git a/tests/contrib/asgi/test_microvm_identity_refresh.py b/tests/contrib/asgi/test_microvm_identity_refresh.py index 0ca6c3cf99f..51593ee343b 100644 --- a/tests/contrib/asgi/test_microvm_identity_refresh.py +++ b/tests/contrib/asgi/test_microvm_identity_refresh.py @@ -4,13 +4,11 @@ from ddtrace.contrib.internal.asgi.middleware import TraceMiddleware from ddtrace.internal import core +from ddtrace.internal.runtime import MICROVM_RUN_HOOK_PATH from .test_asgi import basic_app -REQUEST_STARTING_PATH = "/web-request-starting" - - def _scope(method, path): return { "client": ("127.0.0.1", 32767), @@ -32,14 +30,14 @@ async def test_microvm_run_hook_request(test_spans): three. """ app = TraceMiddleware(basic_app) - instance = ApplicationCommunicator(app, _scope("POST", REQUEST_STARTING_PATH)) + instance = ApplicationCommunicator(app, _scope("POST", MICROVM_RUN_HOOK_PATH)) with mock.patch("ddtrace.contrib.internal.asgi.middleware.core.dispatch", wraps=core.dispatch) as m: await instance.send_input({"type": "http.request", "body": b""}) await instance.receive_output(1) await instance.receive_output(1) - m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", REQUEST_STARTING_PATH)) + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", MICROVM_RUN_HOOK_PATH)) @pytest.mark.asyncio @@ -61,7 +59,7 @@ async def test_sub_app_does_not_double_refresh(test_spans): (matches the existing not-is_subapp guard around route collection/distributed headers). """ app = TraceMiddleware(basic_app) - scope = _scope("POST", REQUEST_STARTING_PATH) + scope = _scope("POST", MICROVM_RUN_HOOK_PATH) # marks this as a sub-app request, per TraceMiddleware.__call__; request_spans matches # the shape _on_asgi_request always creates the dict with (ddtrace/_trace/trace_handlers.py) scope["datadog"] = {"request_spans": []} diff --git a/tests/contrib/bottle/test_microvm_identity_refresh.py b/tests/contrib/bottle/test_microvm_identity_refresh.py index aa1ad09864d..2d5ccf87a28 100644 --- a/tests/contrib/bottle/test_microvm_identity_refresh.py +++ b/tests/contrib/bottle/test_microvm_identity_refresh.py @@ -5,12 +5,10 @@ from ddtrace.contrib.internal.bottle.patch import TracePlugin from ddtrace.contrib.internal.bottle.patch import patch from ddtrace.internal import core +from ddtrace.internal.runtime import MICROVM_RUN_HOOK_PATH from tests.utils import TracerTestCase -REQUEST_STARTING_PATH = "/web-request-starting" - - class BottleMicrovmIdentityRefreshTestCase(TracerTestCase): """traced_wsgi() wraps Bottle.wsgi() -- the WSGI entry point, run before routing -- so it emits every request's real method/path before request tracing starts. Bottle has no @@ -34,10 +32,10 @@ def test_microvm_run_hook_request(self): self._trace_app() with mock.patch("ddtrace.contrib.internal.bottle.trace.core.dispatch", wraps=core.dispatch) as m: - resp = self.app.post(REQUEST_STARTING_PATH, expect_errors=True) + resp = self.app.post(MICROVM_RUN_HOOK_PATH, expect_errors=True) assert resp.status_int == 404 - m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", REQUEST_STARTING_PATH)) + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", MICROVM_RUN_HOOK_PATH)) def test_other_request(self): @self.app.route("/hi/") diff --git a/tests/contrib/cherrypy/test_microvm_identity_refresh.py b/tests/contrib/cherrypy/test_microvm_identity_refresh.py index c6f566c1d64..1aab78764b5 100644 --- a/tests/contrib/cherrypy/test_microvm_identity_refresh.py +++ b/tests/contrib/cherrypy/test_microvm_identity_refresh.py @@ -4,14 +4,12 @@ from ddtrace.contrib.internal.cherrypy.patch import TraceMiddleware from ddtrace.internal import core +from ddtrace.internal.runtime import MICROVM_RUN_HOOK_PATH from tests.utils import TracerTestCase from .web import StubApp -REQUEST_STARTING_PATH = "/web-request-starting" - - class CherrypyMicrovmIdentityRefreshTestCase(TracerTestCase, helper.CPWebCase): """TraceTool._on_start_resource() must dispatch method/path before request tracing starts. @@ -38,10 +36,10 @@ def test_microvm_run_hook_request(self): 404 (see test_404 in test_middleware.py). """ with mock.patch("ddtrace.contrib.internal.cherrypy.patch.core.dispatch", wraps=core.dispatch) as m: - self.getPage(REQUEST_STARTING_PATH, method="POST") + self.getPage(MICROVM_RUN_HOOK_PATH, method="POST") self.assertStatus("404 Not Found") - m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", REQUEST_STARTING_PATH)) + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", MICROVM_RUN_HOOK_PATH)) def test_other_request(self): with mock.patch("ddtrace.contrib.internal.cherrypy.patch.core.dispatch", wraps=core.dispatch) as m: diff --git a/tests/contrib/django/test_microvm_identity_refresh.py b/tests/contrib/django/test_microvm_identity_refresh.py index 59fbb2dbb89..34d9e257904 100644 --- a/tests/contrib/django/test_microvm_identity_refresh.py +++ b/tests/contrib/django/test_microvm_identity_refresh.py @@ -1,8 +1,7 @@ import mock from ddtrace.internal import core - -REQUEST_STARTING_PATH = "/web-request-starting" +from ddtrace.internal.runtime import MICROVM_RUN_HOOK_PATH def test_microvm_run_hook_request(client): @@ -12,10 +11,10 @@ def test_microvm_run_hook_request(client): test_django_request_not_found). """ with mock.patch("ddtrace.contrib.internal.django.response.core.dispatch", wraps=core.dispatch) as m: - resp = client.post(REQUEST_STARTING_PATH) + resp = client.post(MICROVM_RUN_HOOK_PATH) assert resp.status_code == 404 - m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", REQUEST_STARTING_PATH)) + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", MICROVM_RUN_HOOK_PATH)) def test_other_request(client): diff --git a/tests/contrib/falcon/test_microvm_identity_refresh.py b/tests/contrib/falcon/test_microvm_identity_refresh.py index baa6fc5706c..35c6f7d5b4f 100644 --- a/tests/contrib/falcon/test_microvm_identity_refresh.py +++ b/tests/contrib/falcon/test_microvm_identity_refresh.py @@ -2,13 +2,11 @@ import mock from ddtrace.internal import core +from ddtrace.internal.runtime import MICROVM_RUN_HOOK_PATH from .app import get_app -REQUEST_STARTING_PATH = "/web-request-starting" - - def _client(): return testing.TestClient(get_app()) @@ -20,10 +18,10 @@ def test_microvm_run_hook_request(): fires on the 404 (see test_404 in test_suite.py). """ with mock.patch("ddtrace.contrib.internal.falcon.middleware.core.dispatch", wraps=core.dispatch) as m: - response = _client().simulate_post(REQUEST_STARTING_PATH) + response = _client().simulate_post(MICROVM_RUN_HOOK_PATH) assert response.status[:3] == "404" - m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", REQUEST_STARTING_PATH)) + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", MICROVM_RUN_HOOK_PATH)) def test_other_request(): diff --git a/tests/contrib/flask/test_microvm_identity_refresh.py b/tests/contrib/flask/test_microvm_identity_refresh.py index a248288fb5a..449cacbbcef 100644 --- a/tests/contrib/flask/test_microvm_identity_refresh.py +++ b/tests/contrib/flask/test_microvm_identity_refresh.py @@ -2,13 +2,11 @@ from ddtrace.contrib.internal.flask.patch import patched_wsgi_app from ddtrace.internal import core +from ddtrace.internal.runtime import MICROVM_RUN_HOOK_PATH from . import BaseFlaskTestCase -REQUEST_STARTING_PATH = "/web-request-starting" - - class FlaskMicrovmIdentityRefreshTestCase(BaseFlaskTestCase): """patched_wsgi_app() must dispatch every request's method/path before tracing starts. @@ -21,10 +19,10 @@ def test_microvm_run_hook_request(self): route matches doesn't change what gets dispatched). """ with mock.patch("ddtrace.contrib.internal.flask.patch.core.dispatch", wraps=core.dispatch) as m: - res = self.client.post(REQUEST_STARTING_PATH) + res = self.client.post(MICROVM_RUN_HOOK_PATH) self.assertEqual(res.status_code, 404) - m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", REQUEST_STARTING_PATH)) + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", MICROVM_RUN_HOOK_PATH)) def test_other_request(self): @self.app.route("/") @@ -39,7 +37,7 @@ def index(): def test_pre_request_event_dispatches_before_wsgi_middleware(self): events = [] - environ = {"REQUEST_METHOD": "POST", "PATH_INFO": REQUEST_STARTING_PATH, "SCRIPT_NAME": ""} + environ = {"REQUEST_METHOD": "POST", "PATH_INFO": MICROVM_RUN_HOOK_PATH, "SCRIPT_NAME": ""} def start_response(status, headers, exc_info=None): pass diff --git a/tests/contrib/http_server/test_microvm_identity_refresh.py b/tests/contrib/http_server/test_microvm_identity_refresh.py index 2879bf6a0d2..54f2b1cd3c7 100644 --- a/tests/contrib/http_server/test_microvm_identity_refresh.py +++ b/tests/contrib/http_server/test_microvm_identity_refresh.py @@ -7,8 +7,7 @@ from ddtrace.contrib.internal.http_server.patch import patch from ddtrace.contrib.internal.http_server.patch import unpatch from ddtrace.internal import core - -REQUEST_STARTING_PATH = "/web-request-starting" +from ddtrace.internal.runtime import MICROVM_RUN_HOOK_PATH def _handler_for(method, path): @@ -35,10 +34,10 @@ def test_microvm_run_hook_request(): supported web framework. """ with mock.patch("ddtrace.contrib.internal.http_server.patch.core.dispatch", wraps=core.dispatch) as m: - parsed = _handler_for("POST", REQUEST_STARTING_PATH).parse_request() + parsed = _handler_for("POST", MICROVM_RUN_HOOK_PATH).parse_request() assert parsed is True - m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", REQUEST_STARTING_PATH)) + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", MICROVM_RUN_HOOK_PATH)) def test_other_request(): diff --git a/tests/contrib/molten/test_microvm_identity_refresh.py b/tests/contrib/molten/test_microvm_identity_refresh.py index 5769d2e39dc..20cfa02e7c5 100644 --- a/tests/contrib/molten/test_microvm_identity_refresh.py +++ b/tests/contrib/molten/test_microvm_identity_refresh.py @@ -5,12 +5,10 @@ from ddtrace.contrib.internal.molten.patch import patch from ddtrace.contrib.internal.molten.patch import unpatch from ddtrace.internal import core +from ddtrace.internal.runtime import MICROVM_RUN_HOOK_PATH from tests.utils import TracerTestCase -REQUEST_STARTING_PATH = "/web-request-starting" - - def greet(): return "Greetings" @@ -33,10 +31,10 @@ def test_microvm_run_hook_request(self): point, ahead of molten's router, so it still fires on the 404. """ with mock.patch("ddtrace.contrib.internal.molten.patch.core.dispatch", wraps=core.dispatch) as m: - response = self.client.request("POST", REQUEST_STARTING_PATH) + response = self.client.request("POST", MICROVM_RUN_HOOK_PATH) self.assertEqual(response.status_code, 404) - m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", REQUEST_STARTING_PATH)) + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", MICROVM_RUN_HOOK_PATH)) def test_other_request(self): with mock.patch("ddtrace.contrib.internal.molten.patch.core.dispatch", wraps=core.dispatch) as m: diff --git a/tests/contrib/pyramid/test_microvm_identity_refresh.py b/tests/contrib/pyramid/test_microvm_identity_refresh.py index 60640fc46f1..b9cccfcf7f1 100644 --- a/tests/contrib/pyramid/test_microvm_identity_refresh.py +++ b/tests/contrib/pyramid/test_microvm_identity_refresh.py @@ -2,13 +2,11 @@ from ddtrace.contrib.internal.pyramid.constants import SETTINGS_TRACE_ENABLED from ddtrace.internal import core +from ddtrace.internal.runtime import MICROVM_RUN_HOOK_PATH from .utils import PyramidTestCase -REQUEST_STARTING_PATH = "/web-request-starting" - - class PyramidMicrovmIdentityRefreshTestCase(PyramidTestCase): """trace_tween() must dispatch method/path before request tracing starts.""" @@ -17,9 +15,9 @@ def test_microvm_run_hook_request(self): route matching, so it still fires on the 404 (see test_404 in utils.py). """ with mock.patch("ddtrace.contrib.internal.pyramid.trace.core.dispatch", wraps=core.dispatch) as m: - self.app.post(REQUEST_STARTING_PATH, status=404) + self.app.post(MICROVM_RUN_HOOK_PATH, status=404) - m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", REQUEST_STARTING_PATH)) + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", MICROVM_RUN_HOOK_PATH)) def test_other_request(self): with mock.patch("ddtrace.contrib.internal.pyramid.trace.core.dispatch", wraps=core.dispatch) as m: @@ -35,7 +33,7 @@ def test_microvm_run_hook_request_with_tracing_disabled(self): self.override_settings({"datadog_trace_service": "foobar", SETTINGS_TRACE_ENABLED: "false"}) with mock.patch("ddtrace.contrib.internal.pyramid.trace.core.dispatch", wraps=core.dispatch) as m: - self.app.post(REQUEST_STARTING_PATH, status=404) + self.app.post(MICROVM_RUN_HOOK_PATH, status=404) - m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", REQUEST_STARTING_PATH)) + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", MICROVM_RUN_HOOK_PATH)) assert len(self.pop_spans()) == 0 diff --git a/tests/contrib/sanic/test_sanic.py b/tests/contrib/sanic/test_sanic.py index 04b93f03e5f..3ef1a81405f 100644 --- a/tests/contrib/sanic/test_sanic.py +++ b/tests/contrib/sanic/test_sanic.py @@ -22,6 +22,7 @@ from ddtrace.contrib.internal.sanic.patch import patch from ddtrace.contrib.internal.sanic.patch import unpatch from ddtrace.internal import core +from ddtrace.internal.runtime import MICROVM_RUN_HOOK_PATH from ddtrace.propagation import http as http_propagation from tests.conftest import DEFAULT_DDTRACE_SUBPROCESS_TEST_SERVICE_NAME from tests.tracer.utils_inferred_spans.test_helpers import assert_web_and_inferred_aws_api_gateway_span_data @@ -30,9 +31,6 @@ from tests.utils import override_http_config -REQUEST_STARTING_PATH = "/web-request-starting" - - # Helpers for handling response objects across sanic versions sanic_version = tuple(map(int, sanic_version.split("."))) @@ -451,10 +449,10 @@ async def test_microvm_run_hook_request(tracer, client, test_spans): request tracing starts. No route is registered here, so this also covers unmatched routes. """ with mock.patch("ddtrace.contrib.internal.sanic.patch.core.dispatch", wraps=core.dispatch) as m: - response = await client.post(REQUEST_STARTING_PATH) + response = await client.post(MICROVM_RUN_HOOK_PATH) assert _response_status(response) in (404, 405) - m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", REQUEST_STARTING_PATH)) + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", MICROVM_RUN_HOOK_PATH)) @pytest.mark.asyncio diff --git a/tests/contrib/tornado/test_microvm_identity_refresh.py b/tests/contrib/tornado/test_microvm_identity_refresh.py index 0543a47fd49..da2270ba195 100644 --- a/tests/contrib/tornado/test_microvm_identity_refresh.py +++ b/tests/contrib/tornado/test_microvm_identity_refresh.py @@ -1,13 +1,11 @@ import mock from ddtrace.internal import core +from ddtrace.internal.runtime import MICROVM_RUN_HOOK_PATH from .utils import TornadoTestCase -REQUEST_STARTING_PATH = "/web-request-starting" - - class TornadoMicrovmIdentityRefreshTestCase(TornadoTestCase): """execute() must dispatch method/path before request tracing starts.""" @@ -17,10 +15,10 @@ def test_microvm_run_hook_request(self): in test_tornado_web.py). """ with mock.patch("ddtrace.contrib.internal.tornado.handlers.core.dispatch", wraps=core.dispatch) as m: - response = self.fetch(REQUEST_STARTING_PATH, method="POST", body="") + response = self.fetch(MICROVM_RUN_HOOK_PATH, method="POST", body="") self.assertEqual(response.code, 404) - m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", REQUEST_STARTING_PATH)) + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", MICROVM_RUN_HOOK_PATH)) def test_other_request(self): with mock.patch("ddtrace.contrib.internal.tornado.handlers.core.dispatch", wraps=core.dispatch) as m: diff --git a/tests/tracer/runtime/test_runtime_id.py b/tests/tracer/runtime/test_runtime_id.py index 40d57733798..59bfd0ec2ae 100644 --- a/tests/tracer/runtime/test_runtime_id.py +++ b/tests/tracer/runtime/test_runtime_id.py @@ -214,7 +214,7 @@ def test_get_process_role_spawn_child() -> None: @pytest.mark.subprocess def test_refresh_identity_changes_runtime_id(): - """refresh_identity() is the non-fork trigger for a new logical process instance.""" + """refresh_identity() is the non-fork trigger used by e.g. an AWS Lambda MicroVM /run hook.""" import ddtrace.internal.runtime as runtime runtime_id = runtime.get_runtime_id() @@ -236,8 +236,9 @@ def test_refresh_identity_changes_runtime_id(): def test_refresh_identity_does_not_record_fork_lineage(): """Unlike a fork, refresh_identity() must not make get_process_role() report a fake worker. - The previous runtime ID was not a real parent process, so recording it as one would - corrupt process-lineage telemetry. + The previous runtime ID was not a real parent process (e.g. it's the shared image + snapshot's ID on an AWS Lambda MicroVM /run), so recording it as one would corrupt + process-lineage telemetry. """ import ddtrace.internal.runtime as runtime @@ -325,3 +326,134 @@ def on_change(self, new_id): runtime.refresh_identity() assert len(runtime._ON_RUNTIME_ID_CHANGE) == baseline + + +@pytest.mark.parametrize("auto_enable_crashtracking", [False]) +def test_listen_for_identity_refresh_hooks_noop_does_not_import_core(monkeypatch, auto_enable_crashtracking): + import builtins + + import ddtrace.internal.runtime as runtime + + monkeypatch.setattr(runtime, "_IS_AWS_LAMBDA_MICROVM", False) + + real_import = builtins.__import__ + + def fail_core_import(name, *args, **kwargs): + fromlist = kwargs.get("fromlist", ()) + if len(args) >= 3: + fromlist = args[2] + if name == "ddtrace.internal" and "core" in fromlist: + raise AssertionError("listen_for_identity_refresh_hooks() imported core outside a MicroVM") + return real_import(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", fail_core_import) + + runtime.listen_for_identity_refresh_hooks() + + +@pytest.mark.subprocess(env={"AWS_LAMBDA_MICROVM_IMAGE_ARN": "arn:aws:lambda:us-east-1::runtime:python3.12"}, err=None) +def test_maybe_refresh_identity_matches_microvm_run_hook(): + """Only the exact AWS Lambda MicroVM "/run" hook request triggers a refresh.""" + import ddtrace.internal.runtime as runtime + + runtime_id = runtime.get_runtime_id() + + runtime.maybe_refresh_identity(runtime.MICROVM_RUN_HOOK_METHOD, runtime.MICROVM_RUN_HOOK_PATH) + + refreshed_runtime_id = runtime.get_runtime_id() + assert refreshed_runtime_id != runtime_id + + runtime.maybe_refresh_identity(runtime.MICROVM_RUN_HOOK_METHOD, runtime.MICROVM_RUN_HOOK_PATH) + + assert runtime.get_runtime_id() == refreshed_runtime_id + + +@pytest.mark.subprocess(env={"AWS_LAMBDA_MICROVM_IMAGE_ARN": "arn:aws:lambda:us-east-1::runtime:python3.12"}, err=None) +def test_identity_refresh_hook_runs_before_root_span_creation(): + """The pre-request hook must refresh runtime-id before a web root span reads it.""" + from ddtrace import tracer + from ddtrace.internal import core + import ddtrace.internal.runtime as runtime + + runtime_id = runtime.get_runtime_id() + core.dispatch(core.WEB_REQUEST_STARTING, (runtime.MICROVM_RUN_HOOK_METHOD, runtime.MICROVM_RUN_HOOK_PATH)) + + refreshed_runtime_id = runtime.get_runtime_id() + assert refreshed_runtime_id != runtime_id + + with tracer.trace("web.request") as span: + assert span.get_tag("runtime-id") == refreshed_runtime_id + + core.dispatch(core.WEB_REQUEST_STARTING, (runtime.MICROVM_RUN_HOOK_METHOD, runtime.MICROVM_RUN_HOOK_PATH)) + + assert runtime.get_runtime_id() == refreshed_runtime_id + + +@pytest.mark.subprocess(env={"AWS_LAMBDA_MICROVM_IMAGE_ARN": "arn:aws:lambda:us-east-1::runtime:python3.12"}, err=None) +def test_maybe_refresh_identity_is_thread_safe(): + """Concurrent observations of the same MicroVM "/run" hook refresh identity once.""" + import threading + import time + + import ddtrace.internal.runtime as runtime + + calls = [] + + def refresh_identity(): + calls.append(1) + time.sleep(0.01) + + runtime.refresh_identity = refresh_identity + + workers = 16 + barrier = threading.Barrier(workers) + errors = [] + threads = [] + + def refresh_from_request_layer(): + try: + barrier.wait() + runtime.maybe_refresh_identity(runtime.MICROVM_RUN_HOOK_METHOD, runtime.MICROVM_RUN_HOOK_PATH) + except Exception as e: + errors.append(e) + + for _ in range(workers): + thread = threading.Thread(target=refresh_from_request_layer) + thread.start() + threads.append(thread) + + for thread in threads: + thread.join() + + assert errors == [] + assert len(calls) == 1 + + +@pytest.mark.subprocess(env={"AWS_LAMBDA_MICROVM_IMAGE_ARN": "arn:aws:lambda:us-east-1::runtime:python3.12"}, err=None) +def test_maybe_refresh_identity_ignores_other_requests(): + """A different method/path, or the "/resume" hook, must not trigger a refresh.""" + import ddtrace.internal.runtime as runtime + + runtime_id = runtime.get_runtime_id() + + runtime.maybe_refresh_identity("GET", runtime.MICROVM_RUN_HOOK_PATH) + runtime.maybe_refresh_identity(runtime.MICROVM_RUN_HOOK_METHOD, "/aws/lambda-microvms/runtime/v1/resume") + runtime.maybe_refresh_identity(runtime.MICROVM_RUN_HOOK_METHOD, "/some/other/path") + + assert runtime.get_runtime_id() == runtime_id + + +@pytest.mark.subprocess(env={"AWS_LAMBDA_MICROVM_IMAGE_ARN": None}, err=None) +def test_listen_for_identity_refresh_hooks_noop_outside_microvm(): + """Outside a MicroVM, do not register the request-event listener.""" + from ddtrace.internal import core + import ddtrace.internal.runtime as runtime + + core.reset_listeners(core.WEB_REQUEST_STARTING) + runtime.listen_for_identity_refresh_hooks() + + runtime_id = runtime.get_runtime_id() + + core.dispatch(core.WEB_REQUEST_STARTING, (runtime.MICROVM_RUN_HOOK_METHOD, runtime.MICROVM_RUN_HOOK_PATH)) + + assert runtime.get_runtime_id() == runtime_id