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/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/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/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/__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/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/__init__.py b/ddtrace/internal/runtime/__init__.py index 3bfb2ff28aa..10e2818558e 100644 --- a/ddtrace/internal/runtime/__init__.py +++ b/ddtrace/internal/runtime/__init__.py @@ -1,9 +1,14 @@ +import threading 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 +17,9 @@ "get_runtime_id", "get_parent_runtime_id", "get_runtime_propagation_envs", + "refresh_identity", + "maybe_refresh_identity", + "listen_for_identity_refresh_hooks", ] @@ -30,30 +38,148 @@ 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 (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 - _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: + # 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 @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()`` + -- 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: 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/_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/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/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/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/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/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/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/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/contrib/asgi/test_microvm_identity_refresh.py b/tests/contrib/asgi/test_microvm_identity_refresh.py new file mode 100644 index 00000000000..51593ee343b --- /dev/null +++ b/tests/contrib/asgi/test_microvm_identity_refresh.py @@ -0,0 +1,73 @@ +from asgiref.testing import ApplicationCommunicator +import mock +import pytest + +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 + + +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", 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", MICROVM_RUN_HOOK_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", 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": []} + 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..2d5ccf87a28 --- /dev/null +++ b/tests/contrib/bottle/test_microvm_identity_refresh.py @@ -0,0 +1,51 @@ +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 ddtrace.internal.runtime import MICROVM_RUN_HOOK_PATH +from tests.utils import TracerTestCase + + +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(MICROVM_RUN_HOOK_PATH, expect_errors=True) + + assert resp.status_int == 404 + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", MICROVM_RUN_HOOK_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..1aab78764b5 --- /dev/null +++ b/tests/contrib/cherrypy/test_microvm_identity_refresh.py @@ -0,0 +1,49 @@ +import cherrypy +from cherrypy.test import helper +import mock + +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 + + +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(MICROVM_RUN_HOOK_PATH, method="POST") + + self.assertStatus("404 Not Found") + 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: + 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..34d9e257904 --- /dev/null +++ b/tests/contrib/django/test_microvm_identity_refresh.py @@ -0,0 +1,25 @@ +import mock + +from ddtrace.internal import core +from ddtrace.internal.runtime import MICROVM_RUN_HOOK_PATH + + +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(MICROVM_RUN_HOOK_PATH) + + assert resp.status_code == 404 + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", MICROVM_RUN_HOOK_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..35c6f7d5b4f --- /dev/null +++ b/tests/contrib/falcon/test_microvm_identity_refresh.py @@ -0,0 +1,32 @@ +from falcon import testing +import mock + +from ddtrace.internal import core +from ddtrace.internal.runtime import MICROVM_RUN_HOOK_PATH + +from .app import get_app + + +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(MICROVM_RUN_HOOK_PATH) + + assert response.status[:3] == "404" + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", MICROVM_RUN_HOOK_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..449cacbbcef --- /dev/null +++ b/tests/contrib/flask/test_microvm_identity_refresh.py @@ -0,0 +1,68 @@ +import mock + +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 + + +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(MICROVM_RUN_HOOK_PATH) + + self.assertEqual(res.status_code, 404) + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", MICROVM_RUN_HOOK_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": MICROVM_RUN_HOOK_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..54f2b1cd3c7 --- /dev/null +++ b/tests/contrib/http_server/test_microvm_identity_refresh.py @@ -0,0 +1,64 @@ +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 +from ddtrace.internal.runtime import MICROVM_RUN_HOOK_PATH + + +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", MICROVM_RUN_HOOK_PATH).parse_request() + + assert parsed is True + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", MICROVM_RUN_HOOK_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..20cfa02e7c5 --- /dev/null +++ b/tests/contrib/molten/test_microvm_identity_refresh.py @@ -0,0 +1,44 @@ +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 ddtrace.internal.runtime import MICROVM_RUN_HOOK_PATH +from tests.utils import TracerTestCase + + +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", MICROVM_RUN_HOOK_PATH) + + self.assertEqual(response.status_code, 404) + 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: + 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..b9cccfcf7f1 --- /dev/null +++ b/tests/contrib/pyramid/test_microvm_identity_refresh.py @@ -0,0 +1,39 @@ +import mock + +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 + + +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(MICROVM_RUN_HOOK_PATH, status=404) + + 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: + 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(MICROVM_RUN_HOOK_PATH, status=404) + + 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 7e4a1ee0ce6..3ef1a81405f 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,8 @@ 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.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 @@ -439,6 +442,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(MICROVM_RUN_HOOK_PATH) + + assert _response_status(response) in (404, 405) + m.assert_any_call(core.WEB_REQUEST_STARTING, ("POST", MICROVM_RUN_HOOK_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..da2270ba195 --- /dev/null +++ b/tests/contrib/tornado/test_microvm_identity_refresh.py @@ -0,0 +1,28 @@ +import mock + +from ddtrace.internal import core +from ddtrace.internal.runtime import MICROVM_RUN_HOOK_PATH + +from .utils import TornadoTestCase + + +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(MICROVM_RUN_HOOK_PATH, method="POST", body="") + + self.assertEqual(response.code, 404) + 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: + response = self.fetch("/success/") + + self.assertEqual(response.code, 200) + m.assert_any_call(core.WEB_REQUEST_STARTING, ("GET", "/success/")) 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/runtime/test_runtime_id.py b/tests/tracer/runtime/test_runtime_id.py index 0b44fccb902..59bfd0ec2ae 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,250 @@ 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 used by e.g. an AWS Lambda MicroVM /run hook.""" + 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 (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 + + 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 + + +@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 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