fix(runtime): refresh identity-bound consumers

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:

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)

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,
                 }
             },

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

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(

ddtrace/internal/remoteconfig/client.py

@@ -9,14 +9,15 @@
 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
 from ddtrace.internal.remoteconfig import ConfigMetadata
 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)

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:

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__()

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:

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

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,

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> {
     CrashtrackerStatus::try_from(CRASHTRACKER_STATUS.load(Ordering::SeqCst))

src/native/lib.rs

@@ -56,6 +56,7 @@ fn _native(m: &Bound<'_, PyModule>) -> PyResult<()> {
         m.add_class::<crashtracker::CrashtrackerStatus>()?;
         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!(

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,