diff --git a/ddtrace/contrib/_events/web_framework.py b/ddtrace/contrib/_events/web_framework.py index 97794868c6b..dcabfe9a927 100644 --- a/ddtrace/contrib/_events/web_framework.py +++ b/ddtrace/contrib/_events/web_framework.py @@ -6,13 +6,16 @@ from ddtrace.contrib._events.http import HttpRequestBaseEvent from ddtrace.ext import SpanKind from ddtrace.ext import SpanTypes +from ddtrace.internal.core.event_names import WEB_REQUEST as WEB_REQUEST_EVENT +from ddtrace.internal.core.event_names import WEB_REQUEST_STARTING as WEB_REQUEST_STARTING_EVENT from ddtrace.internal.core.events import event_field from ddtrace.internal.schema import SpanDirection from ddtrace.internal.schema import schematize_url_operation class WebFrameworkEvents(str, Enum): - WEB_REQUEST = "web.request" + WEB_REQUEST = WEB_REQUEST_EVENT + WEB_REQUEST_STARTING = WEB_REQUEST_STARTING_EVENT @dataclass diff --git a/ddtrace/contrib/internal/flask/patch.py b/ddtrace/contrib/internal/flask/patch.py index 12ceeb8ec9f..fdb9bc5d7ac 100644 --- a/ddtrace/contrib/internal/flask/patch.py +++ b/ddtrace/contrib/internal/flask/patch.py @@ -7,6 +7,7 @@ from werkzeug.exceptions import NotFound from ddtrace.contrib import trace_utils +from ddtrace.contrib._events.web_framework import WebFrameworkEvents from ddtrace.ext import SpanTypes from ddtrace.internal import core from ddtrace.internal.constants import COMPONENT @@ -391,6 +392,9 @@ def unpatch(): def patched_wsgi_app(wrapped, instance, args, kwargs): environ, start_response = args + core.dispatch( + WebFrameworkEvents.WEB_REQUEST_STARTING.value, (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/internal/core/crashtracking.py b/ddtrace/internal/core/crashtracking.py index 2c2f541792c..87185946961 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,26 @@ is_available = False +# Runtime ID callbacks only receive the new ID, so keep the start() tags here +# to rebuild crashtracker metadata on explicit identity refreshes. +_identity_refresh_additional_tags: Optional[dict[str, str]] = None +_identity_refresh_fork_generation = forksafe.get_generation() + + +def _on_identity_refresh(_new_runtime_id: str) -> None: + global _identity_refresh_fork_generation + + fork_generation = forksafe.get_generation() + if fork_generation != _identity_refresh_fork_generation: + # Runtime ID callbacks also run from the forksafe hook, before crashtracker's + # own fork handler. Skip that notification so the native crashtracker is + # reset only by crashtracker_on_fork; explicit later identity refreshes in + # the child still reconfigure below. + _identity_refresh_fork_generation = fork_generation + return + _reconfigure_for_identity_refresh(_identity_refresh_additional_tags) + + def _get_tags(additional_tags: Optional[dict[str, str]]) -> dict[str, str]: tags = { "language": "python", @@ -211,12 +233,32 @@ def is_started() -> bool: return crashtracker_status() == CrashtrackerStatus.Initialized +def _reconfigure_for_identity_refresh(additional_tags: Optional[dict[str, str]]) -> None: + # Native reconfigure is best-effort during identity refresh: if it fails, + # do not break runtime.refresh_identity() or later runtime ID callbacks. + try: + 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) + except Exception: + log.debug("Failed to reconfigure crashtracker after identity refresh", exc_info=True) + + def start(additional_tags: Optional[dict[str, str]] = None) -> bool: + global _identity_refresh_additional_tags, _identity_refresh_fork_generation + if not is_available: return False if not crashtracker_config.enabled: return False + additional_tags = dict(additional_tags) if additional_tags is not None else None + try: config, receiver_config, metadata = _get_args(additional_tags) if config is None or receiver_config is None or metadata is None: @@ -256,6 +298,9 @@ def crashtracker_fork_handler(): crashtracker_on_fork(config, receiver_config, metadata) forksafe.register(crashtracker_fork_handler) + _identity_refresh_additional_tags = additional_tags + _identity_refresh_fork_generation = forksafe.get_generation() + on_runtime_id_change(_on_identity_refresh) except Exception: log.exception("Failed to start crashtracker") return False diff --git a/ddtrace/internal/core/event_names.py b/ddtrace/internal/core/event_names.py new file mode 100644 index 00000000000..b0c20033e2e --- /dev/null +++ b/ddtrace/internal/core/event_names.py @@ -0,0 +1,9 @@ +"""Shared names for events dispatched through ddtrace.internal.core.""" + +WEB_REQUEST = "web.request" +WEB_REQUEST_STARTING = "web.request.starting" + +__all__ = [ + "WEB_REQUEST", + "WEB_REQUEST_STARTING", +] diff --git a/ddtrace/internal/native/_native.pyi b/ddtrace/internal/native/_native.pyi index 56ba431263f..b60c9519a96 100644 --- a/ddtrace/internal/native/_native.pyi +++ b/ddtrace/internal/native/_native.pyi @@ -122,6 +122,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/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 61cb6005a9f..0ca9d530751 100644 --- a/src/native/lib.rs +++ b/src/native/lib.rs @@ -56,6 +56,7 @@ fn _native(m: &Bound<'_, PyModule>) -> PyResult<()> { m.add_class::()?; m.add_function(wrap_pyfunction!(crashtracker::crashtracker_init, m)?)?; m.add_function(wrap_pyfunction!(crashtracker::crashtracker_on_fork, m)?)?; + m.add_function(wrap_pyfunction!(crashtracker::crashtracker_reconfigure, m)?)?; m.add_function(wrap_pyfunction!(crashtracker::crashtracker_status, m)?)?; m.add_function(wrap_pyfunction!(crashtracker::crashtracker_receiver, m)?)?; m.add_function(wrap_pyfunction!( diff --git a/tests/contrib/flask/test_microvm_identity_refresh.py b/tests/contrib/flask/test_microvm_identity_refresh.py new file mode 100644 index 00000000000..db55a8d9fdb --- /dev/null +++ b/tests/contrib/flask/test_microvm_identity_refresh.py @@ -0,0 +1,71 @@ +import mock + +from ddtrace.contrib._events.web_framework import WebFrameworkEvents +from ddtrace.contrib.internal.flask.patch import patched_wsgi_app +from ddtrace.internal import core + +from . import BaseFlaskTestCase + + +REQUEST_STARTING_PATH = "/web-request-starting" + + +class FlaskMicrovmIdentityRefreshTestCase(BaseFlaskTestCase): + """patched_wsgi_app() must dispatch every request's method/path before tracing starts. + + The matching logic itself is tested in tests/tracer/runtime/test_runtime_id.py. + """ + + def test_microvm_run_hook_request(self): + """No route is registered at the hook path: wsgi_app() runs before routing, so it + must fire even on a 404 -- the stronger, more general form of this check (whether the + route matches doesn't change what gets dispatched). + """ + with mock.patch("ddtrace.contrib.internal.flask.patch.core.dispatch", wraps=core.dispatch) as m: + res = self.client.post(REQUEST_STARTING_PATH) + + self.assertEqual(res.status_code, 404) + m.assert_any_call(WebFrameworkEvents.WEB_REQUEST_STARTING.value, ("POST", REQUEST_STARTING_PATH)) + + def test_other_request(self): + @self.app.route("/") + def index(): + return "ok", 200 + + with mock.patch("ddtrace.contrib.internal.flask.patch.core.dispatch", wraps=core.dispatch) as m: + res = self.client.get("/") + + self.assertEqual(res.status_code, 200) + m.assert_any_call(WebFrameworkEvents.WEB_REQUEST_STARTING.value, ("GET", "/")) + + def test_pre_request_event_dispatches_before_wsgi_middleware(self): + events = [] + environ = {"REQUEST_METHOD": "POST", "PATH_INFO": REQUEST_STARTING_PATH, "SCRIPT_NAME": ""} + + def start_response(status, headers, exc_info=None): + pass + + def wrapped(environ, start_response): + return [] + + def dispatch(name, args): + if name == WebFrameworkEvents.WEB_REQUEST_STARTING.value: + 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/crashtracker/test_crashtracker.py b/tests/crashtracker/test_crashtracker.py index 2c23bba2037..098a2922999 100644 --- a/tests/crashtracker/test_crashtracker.py +++ b/tests/crashtracker/test_crashtracker.py @@ -98,6 +98,159 @@ 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(err=None) +@pytest.mark.parametrize("auto_enable_crashtracking", [False]) +def test_crashtracker_identity_refresh_reconfigure_errors_are_best_effort(): + from contextlib import ExitStack + + import mock + + from ddtrace.internal.core import crashtracking + + initialized = object() + status = type("CrashtrackerStatus", (), {"Initialized": initialized}) + + with ExitStack() as stack: + stack.enter_context(mock.patch.object(crashtracking, "is_available", True)) + stack.enter_context(mock.patch.object(crashtracking, "CrashtrackerStatus", status, create=True)) + stack.enter_context( + mock.patch.object(crashtracking, "_get_args", mock.Mock(return_value=(object(), object(), object()))) + ) + stack.enter_context( + mock.patch.object(crashtracking, "crashtracker_status", mock.Mock(return_value=initialized), create=True) + ) + stack.enter_context( + mock.patch.object( + crashtracking, "crashtracker_reconfigure", mock.Mock(side_effect=RuntimeError("boom")), create=True + ) + ) + + crashtracking._reconfigure_for_identity_refresh({"service": "identity-refresh"}) + + +@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_uses_initial_additional_tags(): + from contextlib import ExitStack + + import mock + + from ddtrace.internal.core import crashtracking + import ddtrace.internal.runtime as runtime + + args = (object(), object(), object()) + tags = {"service": "identity-refresh"} + seen_tags = [] + initialized = object() + status = type("CrashtrackerStatus", (), {"Initialized": initialized}) + + def get_args(additional_tags): + seen_tags.append(dict(additional_tags)) + return args + + 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", mock.Mock(), create=True)) + stack.enter_context(mock.patch.object(crashtracking, "crashtracker_reconfigure", mock.Mock(), create=True)) + stack.enter_context( + mock.patch.object(crashtracking, "crashtracker_status", mock.Mock(return_value=initialized), create=True) + ) + + assert crashtracking.start(tags) + tags["service"] = "mutated" + tags["extra"] = "mutated" + runtime.refresh_identity() + + assert seen_tags == [{"service": "identity-refresh"}, {"service": "identity-refresh"}] + + +@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_skips_first_new_fork_generation(): + 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()) + initialized = object() + status = type("CrashtrackerStatus", (), {"Initialized": initialized}) + get_args = mock.Mock(side_effect=[init_args, refresh_args]) + 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", mock.Mock(), 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) + ) + stack.enter_context( + mock.patch.object(crashtracking.forksafe, "get_generation", mock.Mock(side_effect=[0, 1, 1])) + ) + + assert crashtracking.start({"service": "identity-refresh"}) + runtime.refresh_identity() + runtime.refresh_identity() + + assert get_args.call_count == 2 + 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():