From 8faa264e9a7da7fc00709978230f505e4adbfd55 Mon Sep 17 00:00:00 2001 From: Tianning Li Date: Thu, 20 Aug 2026 00:09:39 -0400 Subject: [PATCH] fix(runtime): refresh identity-bound consumers Rotating the runtime id is not enough by itself. Several long-lived components bake runtime or Remote Config client identity into native clients, workers, upload metadata, or tag caches. Have those components subscribe to explicit identity refreshes and rebuild only the state that captures those ids. Fork-specific cleanup remains on the existing fork hooks so this path does not drop buffers or invent fork lineage. --- ddtrace/appsec/_asm_request_context.py | 8 ++- ddtrace/appsec/_remoteconfiguration.py | 5 +- ddtrace/debugging/_probe/status.py | 7 ++- ddtrace/internal/core/crashtracking.py | 24 ++++++++ ddtrace/internal/native/_native.pyi | 3 + ddtrace/internal/remoteconfig/client.py | 16 +++++- ddtrace/internal/runtime/runtime_metrics.py | 17 ++++-- ddtrace/internal/settings/asm.py | 4 +- ddtrace/internal/symbol_db/symbols.py | 9 +++ ddtrace/internal/telemetry/writer.py | 16 ++++++ ddtrace/internal/writer/writer.py | 15 +++++ src/native/crashtracker.rs | 13 +++++ src/native/lib.rs | 1 + .../appsec/appsec/test_asm_request_context.py | 14 +++++ .../appsec/appsec/test_remoteconfiguration.py | 40 ++++++++++++++ tests/crashtracker/test_crashtracker.py | 40 ++++++++++++++ .../remoteconfig/test_remoteconfig_native.py | 43 +++++++++++++++ tests/internal/symbol_db/test_symbols.py | 25 +++++++++ tests/runtime/test_runtime_metrics_api.py | 22 ++++++++ tests/telemetry/test_writer.py | 55 +++++++++++++++++++ tests/tracer/test_writer.py | 33 +++++++++++ 21 files changed, 397 insertions(+), 13 deletions(-) diff --git a/ddtrace/appsec/_asm_request_context.py b/ddtrace/appsec/_asm_request_context.py index 0e2ba7ad4eb..452abe270f2 100644 --- a/ddtrace/appsec/_asm_request_context.py +++ b/ddtrace/appsec/_asm_request_context.py @@ -400,8 +400,12 @@ def finalize_asm_env(env: ASM_Environment) -> None: entry_span._set_attribute(APPSEC.EVENT_RULE_ERROR_COUNT, info.failed) except Exception: logger.debug("asm_context::finalize_asm_env::exception", extra=log_extra, exc_info=True) - if asm_config._rc_client_id is not None: - entry_span.set_tag(APPSEC.RC_CLIENT_ID, asm_config._rc_client_id) + if asm_config._rc_client_id_enabled: + from ddtrace.internal.remoteconfig.worker import remoteconfig_poller + + rc_client_id = remoteconfig_poller._client.id + if rc_client_id is not None: + entry_span.set_tag(APPSEC.RC_CLIENT_ID, rc_client_id) waf_adresses = env.waf_addresses req_headers = waf_adresses.get(SPAN_DATA_NAMES.REQUEST_HEADERS_NO_COOKIES, {}) if req_headers: diff --git a/ddtrace/appsec/_remoteconfiguration.py b/ddtrace/appsec/_remoteconfiguration.py index f7085e99d95..7fe2da28515 100644 --- a/ddtrace/appsec/_remoteconfiguration.py +++ b/ddtrace/appsec/_remoteconfiguration.py @@ -80,10 +80,13 @@ def enable_appsec_rc(callback: "AppSecCallback") -> None: if asm_config._asm_enabled: telemetry_writer.product_activated(TELEMETRY_APM_PRODUCT.APPSEC, True) - asm_config._rc_client_id = remoteconfig_poller._client.id + + asm_config._rc_client_id_enabled = True def disable_appsec_rc() -> None: + asm_config._rc_client_id_enabled = False + for product_name in APPSEC_PRODUCTS: remoteconfig_poller.unregister_callback(product_name) remoteconfig_poller.disable_product(product_name) diff --git a/ddtrace/debugging/_probe/status.py b/ddtrace/debugging/_probe/status.py index 47aeae81225..c646d0b0031 100644 --- a/ddtrace/debugging/_probe/status.py +++ b/ddtrace/debugging/_probe/status.py @@ -7,9 +7,10 @@ from ddtrace.debugging._metrics import metrics from ddtrace.debugging._probe.model import Probe from ddtrace.debugging._uploader import build_debugger_sender -from ddtrace.internal import runtime from ddtrace.internal.logger import get_logger from ddtrace.internal.native import DebuggerTrackType +from ddtrace.internal.runtime import get_ancestor_runtime_id +from ddtrace.internal.runtime import get_runtime_id from ddtrace.internal.utils.retry import fibonacci_backoff_with_jitter @@ -47,8 +48,8 @@ def _payload( "diagnostics": { "probeId": probe.probe_id, "probeVersion": probe.version, - "runtimeId": runtime.get_runtime_id(), - "parentId": runtime.get_ancestor_runtime_id(), + "runtimeId": get_runtime_id(), + "parentId": get_ancestor_runtime_id(), "status": status, } }, diff --git a/ddtrace/internal/core/crashtracking.py b/ddtrace/internal/core/crashtracking.py index 2c2f541792c..941bc39b486 100644 --- a/ddtrace/internal/core/crashtracking.py +++ b/ddtrace/internal/core/crashtracking.py @@ -14,6 +14,7 @@ from ddtrace.internal.compat import ensure_text from ddtrace.internal.logger import get_logger from ddtrace.internal.runtime import get_runtime_id +from ddtrace.internal.runtime import on_runtime_id_change from ddtrace.internal.settings import env from ddtrace.internal.settings._agent import config as agent_config from ddtrace.internal.settings.crashtracker import config as crashtracker_config @@ -33,6 +34,7 @@ from ddtrace.internal.native._native import StacktraceCollection from ddtrace.internal.native._native import crashtracker_init from ddtrace.internal.native._native import crashtracker_on_fork + from ddtrace.internal.native._native import crashtracker_reconfigure from ddtrace.internal.native._native import crashtracker_report_unhandled_exception from ddtrace.internal.native._native import crashtracker_status @@ -41,6 +43,13 @@ is_available = False +_identity_refresh_additional_tags: Optional[dict[str, str]] = None + + +def _on_identity_refresh(_new_runtime_id: str) -> None: + _reconfigure_for_identity_refresh(_identity_refresh_additional_tags) + + def _get_tags(additional_tags: Optional[dict[str, str]]) -> dict[str, str]: tags = { "language": "python", @@ -211,7 +220,20 @@ def is_started() -> bool: return crashtracker_status() == CrashtrackerStatus.Initialized +def _reconfigure_for_identity_refresh(additional_tags: Optional[dict[str, str]]) -> None: + if not is_started(): + return + + config, receiver_config, metadata = _get_args(additional_tags) + if config is None or receiver_config is None or metadata is None: + log.error("Failed to reconfigure crashtracker after identity refresh: failed to construct configuration") + return + crashtracker_reconfigure(config, receiver_config, metadata) + + def start(additional_tags: Optional[dict[str, str]] = None) -> bool: + global _identity_refresh_additional_tags + if not is_available: return False if not crashtracker_config.enabled: @@ -256,6 +278,8 @@ def crashtracker_fork_handler(): crashtracker_on_fork(config, receiver_config, metadata) forksafe.register(crashtracker_fork_handler) + _identity_refresh_additional_tags = additional_tags + on_runtime_id_change(_on_identity_refresh) except Exception: log.exception("Failed to start crashtracker") return False diff --git a/ddtrace/internal/native/_native.pyi b/ddtrace/internal/native/_native.pyi index 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/ddtrace/internal/remoteconfig/client.py b/ddtrace/internal/remoteconfig/client.py index 93604bb71f9..3056fc1a7e3 100644 --- a/ddtrace/internal/remoteconfig/client.py +++ b/ddtrace/internal/remoteconfig/client.py @@ -9,7 +9,6 @@ import ddtrace from ddtrace.internal import gitmetadata from ddtrace.internal import process_tags -from ddtrace.internal import runtime from ddtrace.internal.hostname import get_hostname from ddtrace.internal.logger import get_logger from ddtrace.internal.packages import is_distribution_available @@ -17,6 +16,8 @@ from ddtrace.internal.remoteconfig import Payload from ddtrace.internal.remoteconfig import PayloadType from ddtrace.internal.remoteconfig import RCCallback +from ddtrace.internal.runtime import get_runtime_id +from ddtrace.internal.runtime import on_runtime_id_change from ddtrace.internal.settings._agent import config as agent_config from ddtrace.internal.settings._core import DDConfig from ddtrace.internal.telemetry import telemetry_writer @@ -98,6 +99,8 @@ def __init__(self) -> None: self._native: Optional[Any] = None self._reader: Optional[Any] = None + on_runtime_id_change(self._on_identity_refresh) + def ensure_native(self) -> Any: if self._native is None: from ddtrace.internal.native import RemoteConfigClient as _NativeClient @@ -109,7 +112,7 @@ def ensure_native(self) -> Any: agent_url=str(self.agent_url), tracer_version=tracer_version, client_id=self.id, - runtime_id=runtime.get_runtime_id(), + runtime_id=get_runtime_id(), service=ddtrace.config.service or "", env=ddtrace.config.env or "", app_version=ddtrace.config.version or "", @@ -125,6 +128,15 @@ def ensure_native(self) -> Any: def renew_id(self) -> None: self.id = str(uuid.uuid4()) + def _on_identity_refresh(self, new_runtime_id: str) -> None: + # Regenerate the client id and drop the native client, which bakes both ids in + # as immutable constructor arguments (get_client_id() is documented "stable for + # the process lifetime"). The next ensure_native() call rebuilds it bound to the + # fresh ids. Safe across threads: request() captures self._native into a local + # before calling .poll(), so an in-flight poll on the old client is unaffected. + self.renew_id() + self._native = None + def register_callback(self, product_name: "RemoteConfigProduct", callback: RCCallback) -> None: self._product_callbacks[product_name] = callback log.debug("[%s][P: %s] Registered callback for product %s", os.getpid(), os.getppid(), product_name) diff --git a/ddtrace/internal/runtime/runtime_metrics.py b/ddtrace/internal/runtime/runtime_metrics.py index 17ef94bfec4..05c99a220a7 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 @@ -93,11 +94,10 @@ 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()) # Only dd.internal.entity_id needs preserving here: service/env/version are already @@ -108,6 +108,15 @@ def __init__(self, interval=DEFAULT_RUNTIME_METRICS_INTERVAL, dogstatsd_url=None tag for tag in (self._dogstatsd_client.constant_tags or []) if tag.startswith(entity_id_prefix) ] + def _build_platform_tags(self) -> list[str]: + if config._runtime_metrics_runtime_id_enabled: + # Enables tagging runtime metrics with runtime-id (as well as all the v1 tags) + return self._format_tags(PlatformTagsV2()) + return self._format_tags(PlatformTags()) + + def _on_identity_refresh(self, _new_runtime_id: str) -> None: + self._platform_tags = self._build_platform_tags() + @classmethod def disable(cls) -> None: with cls._lock: diff --git a/ddtrace/internal/settings/asm.py b/ddtrace/internal/settings/asm.py index c129617c3f8..cba47f46631 100644 --- a/ddtrace/internal/settings/asm.py +++ b/ddtrace/internal/settings/asm.py @@ -268,7 +268,9 @@ class ASMConfig(DDConfig): sys.platform.startswith("win") or sys.platform.startswith("cygwin") ) - _rc_client_id: Optional[str] = None + # Set by enable_appsec_rc()/disable_appsec_rc(); gates _dd.rc.client_id span tagging so it's + # only emitted while AppSec RC is actually enabled, not just whenever a live RC client exists. + _rc_client_id_enabled: bool = False def __init__(self): super().__init__() diff --git a/ddtrace/internal/symbol_db/symbols.py b/ddtrace/internal/symbol_db/symbols.py index b8b9daa35e6..39cfecd6753 100644 --- a/ddtrace/internal/symbol_db/symbols.py +++ b/ddtrace/internal/symbol_db/symbols.py @@ -38,6 +38,7 @@ from ddtrace.internal.periodic import Timer from ddtrace.internal.runtime import get_ancestor_runtime_id from ddtrace.internal.runtime import get_runtime_id +from ddtrace.internal.runtime import on_runtime_id_change from ddtrace.internal.safety import _isinstance from ddtrace.internal.settings._agent import config as agent_config from ddtrace.internal.settings.dynamic_instrumentation import config as di_config @@ -561,6 +562,9 @@ def __init__(self, scopes: t.Optional[list[Scope]] = None) -> None: } forksafe.register(self._reset_on_fork) + # Same rebuild, triggered by an explicit identity refresh (e.g. an AWS Lambda + # MicroVM /run hook) rather than an actual fork. + on_runtime_id_change(self._on_identity_refresh) @cached_property def _sender(self) -> SymDBSender: @@ -577,6 +581,11 @@ def _reset_on_fork(self) -> None: self._event_data["runtimeId"] = get_runtime_id() self._event_data["parentId"] = get_ancestor_runtime_id() + def _on_identity_refresh(self, new_runtime_id: str) -> None: + # Same rebuild as _reset_on_fork(): runtimeId is baked into _event_data, so it must be + # refreshed here too or every batch keeps reporting the pre-refresh snapshot's ID. + self._reset_on_fork() + def _set_timer(self) -> None: with self._timer_lock: if self._timer is None: diff --git a/ddtrace/internal/telemetry/writer.py b/ddtrace/internal/telemetry/writer.py index e5ab691ed75..863da58eafb 100644 --- a/ddtrace/internal/telemetry/writer.py +++ b/ddtrace/internal/telemetry/writer.py @@ -22,6 +22,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 @@ -203,6 +204,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": @@ -915,6 +919,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 652cca4b15c..4772fd293ff 100644 --- a/ddtrace/internal/writer/writer.py +++ b/ddtrace/internal/writer/writer.py @@ -20,6 +20,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 @@ -815,6 +816,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: @@ -930,6 +932,19 @@ def shutdown_exporter(self) -> None: except Exception: _safelog(log.warning, "failed to shutdown exporter", exc_info=True) + def _on_identity_refresh(self, new_runtime_id: str) -> None: + # Rebuild the exporter so it picks up the new runtime_id (baked in at construction + # via enable_telemetry()), without touching the span buffer: unlike a fork, no spans + # were lost, so anything already buffered should still flush once the new exporter's + # connection is up. Modeled on set_test_session_token(), not recreate()/fork, which + # replace the whole writer and drop the buffer. + old_exporter = self._exporter + self._exporter = self._create_exporter() + try: + old_exporter.shutdown(3_000_000_000) + except Exception: + _safelog(log.warning, "failed to shutdown exporter", exc_info=True) + def recreate( self, appsec_enabled: Optional[bool] = None, diff --git a/src/native/crashtracker.rs b/src/native/crashtracker.rs index e2fb8149de7..ad1a35ce6fb 100644 --- a/src/native/crashtracker.rs +++ b/src/native/crashtracker.rs @@ -320,6 +320,19 @@ pub fn crashtracker_on_fork<'py>( libdd_crashtracker::on_fork(inner_config, inner_receiver_config, inner_metadata) } +#[pyfunction(name = "crashtracker_reconfigure")] +pub fn crashtracker_reconfigure<'py>( + mut config: PyRefMut<'py, CrashtrackerConfigurationPy>, + mut receiver_config: PyRefMut<'py, CrashtrackerReceiverConfigPy>, + mut metadata: PyRefMut<'py, CrashtrackerMetadataPy>, +) -> anyhow::Result<()> { + let inner_config = (*config).take_inner_or_err()?; + let inner_receiver_config = (*receiver_config).take_inner_or_err()?; + let inner_metadata = (*metadata).take_inner_or_err()?; + + libdd_crashtracker::reconfigure(inner_config, inner_receiver_config, inner_metadata) +} + #[pyfunction(name = "crashtracker_status")] pub fn crashtracker_status() -> anyhow::Result { CrashtrackerStatus::try_from(CRASHTRACKER_STATUS.load(Ordering::SeqCst)) diff --git a/src/native/lib.rs b/src/native/lib.rs index 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/appsec/appsec/test_asm_request_context.py b/tests/appsec/appsec/test_asm_request_context.py index c42a8768652..94241ee87b5 100644 --- a/tests/appsec/appsec/test_asm_request_context.py +++ b/tests/appsec/appsec/test_asm_request_context.py @@ -14,6 +14,20 @@ config_asm = {"_asm_enabled": True} +@pytest.mark.parametrize("auto_enable_crashtracking", [False]) +def test_import_does_not_load_remoteconfig_worker(run_python_code_in_subprocess, auto_enable_crashtracking): + code = """ +import sys + +import ddtrace.appsec._asm_request_context # noqa: F401 + +assert "ddtrace.internal.remoteconfig.worker" not in sys.modules +""" + + _, stderr, status, _ = run_python_code_in_subprocess(code) + assert status == 0, stderr + + def test_context_set_and_reset(): with asm_context( ip_addr=_TEST_IP, diff --git a/tests/appsec/appsec/test_remoteconfiguration.py b/tests/appsec/appsec/test_remoteconfiguration.py index f0c7f66dd17..c169626ebba 100644 --- a/tests/appsec/appsec/test_remoteconfiguration.py +++ b/tests/appsec/appsec/test_remoteconfiguration.py @@ -270,6 +270,46 @@ def test_rc_activation_validate_client_id(tracer, rc_poller, appsec_callback): disable_appsec_rc() +def test_rc_client_id_tag_reflects_live_value_not_a_stale_cache(tracer, rc_poller, appsec_callback): + """_dd.rc.client_id must be read live at span-tagging time, not cached once at RC enable time. + + Otherwise the tag would go stale after e.g. an AWS Lambda MicroVM identity refresh + regenerates the real client id. + """ + from ddtrace.internal.remoteconfig.worker import remoteconfig_poller + + with override_global_config(dict(_asm_enabled=True, _remote_config_enabled=True, api_version="v0.4")): + tracer.configure(appsec_enabled=True) + enable_appsec_rc(appsec_callback) + + with mock.patch.object(remoteconfig_poller._client, "id", "client-id-one"): + with asm_context(tracer) as span: + set_http_meta(span, {}, raw_uri="http://example.com/", status_code="200") + assert span._local_root._get_str_attribute(APPSEC.RC_CLIENT_ID) == "client-id-one" + + with mock.patch.object(remoteconfig_poller._client, "id", "client-id-two"): + with asm_context(tracer) as span: + set_http_meta(span, {}, raw_uri="http://example.com/", status_code="200") + assert span._local_root._get_str_attribute(APPSEC.RC_CLIENT_ID) == "client-id-two" + disable_appsec_rc() + + +def test_rc_client_id_tag_not_set_when_rc_disabled(tracer): + """_dd.rc.client_id must not be tagged when AppSec RC was never enabled, even though a live + RC client (and id) exists process-wide -- otherwise every ASM-tracked span would get tagged + with an id from a Remote Config subscription AppSec never activated. + """ + from ddtrace.internal.remoteconfig.worker import remoteconfig_poller + + with override_global_config(dict(_asm_enabled=True, api_version="v0.4")): + tracer.configure(appsec_enabled=True) + + with mock.patch.object(remoteconfig_poller._client, "id", "client-id-one"): + with asm_context(tracer) as span: + set_http_meta(span, {}, raw_uri="http://example.com/", status_code="200") + assert span._local_root._get_str_attribute(APPSEC.RC_CLIENT_ID) is None + + @pytest.mark.parametrize( "env_rules, expected", [ diff --git a/tests/crashtracker/test_crashtracker.py b/tests/crashtracker/test_crashtracker.py index 2c23bba2037..677c6aac461 100644 --- a/tests/crashtracker/test_crashtracker.py +++ b/tests/crashtracker/test_crashtracker.py @@ -98,6 +98,46 @@ def test_crashtracker_started(): pytest.fail("contents of stdout.log: %s, stderr.log: %s" % (stdout_msg, stderr_msg)) +@pytest.mark.skipif(not sys.platform.startswith("linux"), reason="Linux only") +@pytest.mark.subprocess(err=None) +@pytest.mark.parametrize("auto_enable_crashtracking", [False]) +def test_crashtracker_identity_refresh_reconfigures_metadata(): + from contextlib import ExitStack + + import mock + + from ddtrace.internal.core import crashtracking + import ddtrace.internal.runtime as runtime + + init_args = (object(), object(), object()) + refresh_args = (object(), object(), object()) + tags = {"service": "identity-refresh"} + initialized = object() + status = type("CrashtrackerStatus", (), {"Initialized": initialized}) + get_args = mock.Mock(side_effect=[init_args, refresh_args]) + init = mock.Mock() + reconfigure = mock.Mock() + + with ExitStack() as stack: + stack.enter_context(mock.patch.object(crashtracking, "is_available", True)) + stack.enter_context(mock.patch.object(crashtracking, "crashtracker_config", mock.Mock(enabled=True))) + stack.enter_context(mock.patch.object(crashtracking, "CrashtrackerStatus", status, create=True)) + stack.enter_context(mock.patch.object(crashtracking, "_identity_refresh_additional_tags", None)) + stack.enter_context(mock.patch.object(crashtracking, "_get_args", get_args)) + stack.enter_context(mock.patch.object(crashtracking, "crashtracker_init", init, create=True)) + stack.enter_context(mock.patch.object(crashtracking, "crashtracker_reconfigure", reconfigure, create=True)) + stack.enter_context( + mock.patch.object(crashtracking, "crashtracker_status", mock.Mock(return_value=initialized), create=True) + ) + + assert crashtracking.start(tags) + runtime.refresh_identity() + + assert get_args.call_args_list == [mock.call(tags), mock.call(tags)] + init.assert_called_once_with(*init_args) + reconfigure.assert_called_once_with(*refresh_args) + + @pytest.mark.skipif(not sys.platform.startswith("linux"), reason="Linux only") @pytest.mark.subprocess() def test_crashtracker_receiver_not_in_path(): diff --git a/tests/internal/remoteconfig/test_remoteconfig_native.py b/tests/internal/remoteconfig/test_remoteconfig_native.py index f1813266321..ff646a167f2 100644 --- a/tests/internal/remoteconfig/test_remoteconfig_native.py +++ b/tests/internal/remoteconfig/test_remoteconfig_native.py @@ -418,3 +418,46 @@ def test_enable_builds_native_runtime_before_registering_fork_hook(monkeypatch): assert poller.enable() is True assert order == ["native", "before_fork", "start"], order + + +def test_identity_refresh_renews_client_id_and_drops_native(): + # get_client_id() on the native client is documented "stable for the process lifetime", + # so refreshing must drop it (not mutate it in place) for the id to actually change. + client = RemoteConfigClient() + old_id = client.id + client.ensure_native() + assert client._native is not None + + client._on_identity_refresh("some-new-runtime-id") + + assert client.id != old_id + assert client._native is None + + +def test_identity_refresh_rebuilds_native_client_with_fresh_id(): + client = RemoteConfigClient() + native_before = client.ensure_native() + old_native_client_id = native_before.get_client_id() + + client._on_identity_refresh("some-new-runtime-id") + native_after = client.ensure_native() + + assert native_after is not native_before + assert native_after.get_client_id() == client.id + assert native_after.get_client_id() != old_native_client_id + + +@pytest.mark.subprocess +def test_identity_refresh_wired_to_runtime_id_change(): + """A RemoteConfigClient subscribes itself at construction; refresh_identity() reaches it.""" + from ddtrace.internal import runtime + from ddtrace.internal.remoteconfig.client import RemoteConfigClient + + client = RemoteConfigClient() + old_id = client.id + client.ensure_native() + + runtime.refresh_identity() + + assert client.id != old_id + assert client._native is None diff --git a/tests/internal/symbol_db/test_symbols.py b/tests/internal/symbol_db/test_symbols.py index c1c5b075312..920ef5cba70 100644 --- a/tests/internal/symbol_db/test_symbols.py +++ b/tests/internal/symbol_db/test_symbols.py @@ -526,6 +526,31 @@ def test_symbols_fork_uploads(): assert os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0, f"child {pid} exited with status {status}" +@pytest.mark.subprocess(ddtrace_run=True, err=None) +def test_symbols_identity_refresh_updates_runtime_id(): + """A non-fork identity refresh (e.g. an AWS Lambda MicroVM /run hook) must also refresh the + ScopeContext's cached runtimeId, the same way _reset_on_fork() does after an actual fork -- + otherwise every upload after a MicroVM /run keeps reporting the pre-refresh snapshot's ID. + """ + import typing as t + + import ddtrace.internal.runtime as runtime + from ddtrace.internal.symbol_db.symbols import SymbolDatabaseUploader + + SymbolDatabaseUploader.install() + + context = t.cast(SymbolDatabaseUploader, SymbolDatabaseUploader._instance)._context + old_runtime_id = runtime.get_runtime_id() + old_upload_id = context._upload_id + assert context._event_data["runtimeId"] == old_runtime_id + + runtime.refresh_identity() + + assert runtime.get_runtime_id() != old_runtime_id + assert context._event_data["runtimeId"] == runtime.get_runtime_id() + assert context._upload_id != old_upload_id + + @pytest.mark.subprocess(ddtrace_run=True, err=None) def test_symbols_fork_forces_reenable_and_install(): """ diff --git a/tests/runtime/test_runtime_metrics_api.py b/tests/runtime/test_runtime_metrics_api.py index 963d8deeae9..b0e19778190 100644 --- a/tests/runtime/test_runtime_metrics_api.py +++ b/tests/runtime/test_runtime_metrics_api.py @@ -232,6 +232,28 @@ def test_runtime_metrics_experimental_runtime_tag(): ) +@pytest.mark.subprocess(env={"DD_RUNTIME_METRICS_RUNTIME_ID_ENABLED": "true"}, err=None) +def test_runtime_metrics_runtime_id_tag_refreshes_on_identity_refresh(): + from ddtrace.internal import runtime + from ddtrace.internal.runtime.runtime_metrics import RuntimeWorker + + try: + RuntimeWorker.enable() + assert RuntimeWorker._instance is not None + + worker_instance = RuntimeWorker._instance + runtime_id_tag = f"runtime-id:{runtime.get_runtime_id()}" + assert runtime_id_tag in worker_instance._platform_tags, worker_instance._platform_tags + + runtime.refresh_identity() + + refreshed_runtime_id_tag = f"runtime-id:{runtime.get_runtime_id()}" + assert refreshed_runtime_id_tag in worker_instance._platform_tags, worker_instance._platform_tags + assert runtime_id_tag not in worker_instance._platform_tags, worker_instance._platform_tags + finally: + RuntimeWorker.disable() + + @pytest.mark.subprocess( parametrize={"DD_TRACE_EXPERIMENTAL_FEATURES_ENABLED": ["DD_RUNTIME_METRICS_ENABLED,someotherfeature", ""]}, err=None, diff --git a/tests/telemetry/test_writer.py b/tests/telemetry/test_writer.py index 7bbede8e4c6..ce5b71a4c98 100644 --- a/tests/telemetry/test_writer.py +++ b/tests/telemetry/test_writer.py @@ -570,6 +570,61 @@ def test_telemetry_writer_agent_setup(): assert new_telemetry_writer._agentless is False +def test_identity_refresh_rebuilds_native_worker(): + """Same rebuild as after a fork: the native worker bakes in get_runtime_id() at construction.""" + with override_global_config( + {"_dd_site": "datad0g.com", "_dd_api_key": "foobarkey", "_ci_visibility_agentless_enabled": False} + ): + writer = ddtrace.internal.telemetry.TelemetryWriter(agentless=False) + assert writer._worker is not None + + writer._on_identity_refresh("some-new-runtime-id") + + assert writer._worker is None + assert writer.started is False + + +def test_identity_refresh_stops_live_worker_before_dropping(): + """Unlike after a fork, the worker is still alive here and must be explicitly stopped, or it + keeps heartbeating with the stale runtime ID until process shutdown. + """ + with override_global_config( + {"_dd_site": "datad0g.com", "_dd_api_key": "foobarkey", "_ci_visibility_agentless_enabled": False} + ): + writer = ddtrace.internal.telemetry.TelemetryWriter(agentless=False) + assert writer._worker is not None + + # TelemetryWorker is a native extension type -- its methods can't be patched in place, + # so swap in a mock to observe the stop() call instead. + fake_worker = mock.Mock() + writer._worker = fake_worker + + writer._on_identity_refresh("some-new-runtime-id") + + fake_worker.stop.assert_called_once_with(send_app_closing=False) + assert writer._worker is None + + +@pytest.mark.subprocess( + env={"DD_SITE": "datad0g.com", "DD_API_KEY": "foobarkey", "DD_CIVISIBILITY_AGENTLESS_ENABLED": "false"} +) +def test_identity_refresh_wired_to_runtime_id_change(): + """Drives the refresh through runtime.refresh_identity() instead of calling + _on_identity_refresh directly (as the test above does), so a dropped + on_runtime_id_change() subscription would actually fail this. + """ + from ddtrace.internal import runtime + import ddtrace.internal.telemetry + + writer = ddtrace.internal.telemetry.TelemetryWriter(agentless=False) + assert writer._worker is not None + + runtime.refresh_identity() + + assert writer._worker is None + assert writer.started is False + + @pytest.mark.parametrize( "env_agentless,arg_agentless", [ diff --git a/tests/tracer/test_writer.py b/tests/tracer/test_writer.py index cc90b6403b2..d9f07704e74 100644 --- a/tests/tracer/test_writer.py +++ b/tests/tracer/test_writer.py @@ -419,6 +419,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() @@ -428,6 +442,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