fix(id): drop pending telemetry state and re-add RC client-id on MicroVM identity refresh

Buffered/aggregated telemetry recorded before a MicroVM snapshot would
otherwise export or flush under every clone's refreshed identity instead
of being dropped with the rest of the pre-clone state. Reset it as part
of the identity-refresh path, in each of the affected subsystems:

- dogstatsd: MetricsAggregationClient drops pending counters/gauges/
  histograms when the wrapped client's tags actually change
- agentless exporter: Writer#resetPendingBatch() discards the pending
  encoded trace batch
- OTLP logs: BatchLogRecordProcessor#resetPendingState() discards
  queued log records and clears the batch timer
- OTLP metrics: PeriodicMetricReader#resetPendingState() discards
  queued measurements and rebases sync Counter/Histogram cumulative
  state
- span stats: SpanStatsProcessor replaces its bucket map
- runtime metrics: rebase CPU/event-loop/ELU sampler baselines so the
  next collection reports a delta since the resume, not one spanning
  the snapshot pause

Also fixes a separate identity-refresh gap: an RC lib-config update
rebuilds config.tags from tracked sources (config/remote_config.js's
tracing_tags transformer), dropping the directly-set _dd.rc.client_id
key. refreshIdentity()'s guard only wrote the refreshed value back when
the tag was already present, so once that sequence happened,
config.tags (and the DogStatsD/OTLP tags built from it) permanently
lost _dd.rc.client_id after an identity refresh, even though the RC
client's own id field kept updating correctly. Gate the write on the RC
client existing instead, and write it unconditionally in that case.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

Author: litianningdatadog

packages/dd-trace/src/dogstatsd.js

@@ -61,18 +61,21 @@ class DogStatsDClient {
    * before the swap.
    *
    * @param {string[]} tags - DogStatsD-formatted tags (e.g. `['key:value']`)
+   * @returns {boolean} True if the tag prefix actually changed (and buffered lines were dropped)
    */
   updateTags (tags) {
     const tagsPrefix = tags.length ? `|#${tags.join(',')}` : ''
 
     this._tags = tags
 
-    if (tagsPrefix === this.#tagsPrefix) return
+    if (tagsPrefix === this.#tagsPrefix) return false
 
     this.#tagsPrefix = tagsPrefix
     this._queue = []
     this._buffer = ''
     this._offset = 0
+
+    return true
   }
 
   increment (stat, value, tags) {
@@ -241,11 +244,16 @@ class MetricsAggregationClient {
   }
 
   /**
-   * Recomputes the wrapped client's cached tags (e.g. after a MicroVM clone resume).
+   * Recomputes the wrapped client's cached tags (e.g. after a MicroVM clone resume). Pending
+   * counters/gauges/histograms were aggregated under the old identity, so they're reset along
+   * with the client's buffered lines — but only if the tags actually changed, so a no-op resume
+   * doesn't discard in-flight aggregation for nothing.
    * @param {string[]} tags - DogStatsD-formatted tags (e.g. `['key:value']`)
    */
   updateTags (tags) {
-    this._client.updateTags(tags)
+    if (this._client.updateTags(tags)) {
+      this.reset()
+    }
   }
 
   flush () {

packages/dd-trace/src/exporters/agentless/index.js

@@ -3,12 +3,16 @@
 const { URL } = require('node:url')
 const os = require('node:os')
 
+const { channel } = require('dc-polyfill')
+
 const log = require('../../log')
 const { entityId } = require('../common/docker')
 const tracerVersion = require('../../../../../package.json').version
 const Writer = require('./writer')
 const { computeIntakeUrl } = require('./intake')
 
+const identityRefreshChannel = channel('datadog:identity:refresh')
+
 /**
  * Agentless exporter for APM trace intake.
  * Sends traces directly to the Datadog intake without requiring a local agent.
@@ -56,6 +60,10 @@ class AgentlessExporter {
       metadata,
     })
 
+    // Drop traces buffered before a MicroVM snapshot instead of flushing them under every
+    // clone's identity. No stop() hook here, same as this exporter's other process-lifetime state.
+    identityRefreshChannel.subscribe(() => this._writer.resetPendingBatch())
+
     const ddTrace = globalThis[Symbol.for('dd-trace')]
     if (ddTrace?.beforeExitHandlers) {
       ddTrace.beforeExitHandlers.add(this.flush.bind(this))

packages/dd-trace/src/exporters/common/writer.js

@@ -70,6 +70,15 @@ class Writer {
   setUrl (url) {
     this._url = url
   }
+
+  /**
+   * Discards the pending encoded batch, e.g. on a MicroVM clone resume so traces buffered
+   * before the snapshot don't flush under every clone's identity.
+   * @returns {void}
+   */
+  resetPendingBatch () {
+    this._encoder.reset()
+  }
 }
 
 module.exports = Writer

packages/dd-trace/src/opentelemetry/logs/batch_log_processor.js

@@ -60,6 +60,16 @@ class BatchLogRecordProcessor {
     this.#export()
   }
 
+  /**
+   * Discards queued log records, e.g. on a MicroVM clone resume so records buffered before the
+   * snapshot don't export under every clone's identity.
+   * @returns {void}
+   */
+  resetPendingState () {
+    this.#logRecords = []
+    this.#clearTimer()
+  }
+
   /**
    * Starts the batch timeout timer.
    * @private

packages/dd-trace/src/opentelemetry/logs/index.js

@@ -1,11 +1,18 @@
 'use strict'
 
+const { channel } = require('dc-polyfill')
+
 const { buildResourceAttributes, registerResourceAttributeRefresh } = require('../resource-attributes')
 
 /**
  * @typedef {import('../../config')} Config
  */
 
+const identityRefreshChannel = channel('datadog:identity:refresh')
+
+// Replaces the previous subscription on each call, so restarting doesn't accumulate listeners.
+let unsubscribeLogsPendingStateReset = null
+
 /**
  * OpenTelemetry Logs Implementation for `dd-trace-js`
  *
@@ -59,6 +66,13 @@ function initializeOpenTelemetryLogs (config) {
   loggerProvider.register()
 
   registerResourceAttributeRefresh(exporter, () => buildResourceAttributes(config))
+
+  // Drop log records queued before a MicroVM snapshot instead of exporting them under every
+  // clone's identity.
+  unsubscribeLogsPendingStateReset?.()
+  const onIdentityRefresh = () => processor.resetPendingState()
+  identityRefreshChannel.subscribe(onIdentityRefresh)
+  unsubscribeLogsPendingStateReset = () => identityRefreshChannel.unsubscribe(onIdentityRefresh)
 }
 
 module.exports = {

packages/dd-trace/src/opentelemetry/metrics/index.js

@@ -3,6 +3,7 @@
 const os = require('os')
 
 const { metrics } = require('@opentelemetry/api')
+const { channel } = require('dc-polyfill')
 
 const { VERSION } = require('../../../../../version')
 const processTags = require('../../process-tags')
@@ -18,6 +19,11 @@ const OtlpHttpMetricExporter = require('./otlp_http_metric_exporter')
  * @typedef {import('../../config')} Config
  */
 
+const identityRefreshChannel = channel('datadog:identity:refresh')
+
+// Replaces the previous subscription on each call, so restarting doesn't accumulate listeners.
+let unsubscribeMetricsPendingStateReset = null
+
 /**
  * @file OpenTelemetry Metrics Implementation for dd-trace-js
  *
@@ -64,6 +70,13 @@ function initializeOpenTelemetryMetrics (config) {
   metrics.setGlobalMeterProvider(meterProvider)
 
   registerResourceAttributeRefresh(exporter, () => buildGeneralResourceAttributes(config))
+
+  // Drop measurements queued before a MicroVM snapshot instead of exporting them under every
+  // clone's identity.
+  unsubscribeMetricsPendingStateReset?.()
+  const onIdentityRefresh = () => reader.resetPendingState()
+  identityRefreshChannel.subscribe(onIdentityRefresh)
+  unsubscribeMetricsPendingStateReset = () => identityRefreshChannel.unsubscribe(onIdentityRefresh)
 }
 
 function buildResourceAttributes (tags, { reportHostname, otelSemanticsEnabled, service, env, serviceVersion } = {}) {

packages/dd-trace/src/opentelemetry/metrics/periodic_metric_reader.js

@@ -207,6 +207,25 @@ class PeriodicMetricReader {
     this.#collectAndExport()
   }
 
+  /**
+   * Discards queued measurements and sync-instrument cumulative state, e.g. on a MicroVM clone
+   * resume, so measurements recorded before the snapshot don't export under every clone's identity.
+   *
+   * Only clears `#lastExportedState` entries that have a matching `#cumulativeState` entry (sync
+   * Counter/Histogram delta baselines). An ObservableCounter's baseline lives only in
+   * `#lastExportedState` — clearing it too would turn its next export into an absolute reading
+   * instead of a delta.
+   * @returns {void}
+   */
+  resetPendingState () {
+    this.#measurements = []
+
+    for (const key of this.#cumulativeState.keys()) {
+      this.#lastExportedState.delete(key)
+    }
+    this.#cumulativeState.clear()
+  }
+
   /**
    * Shuts down the reader and stops periodic collection.
    * @returns {void}

packages/dd-trace/src/remote_config/index.js

@@ -601,10 +601,11 @@ function getTagsString (config, repositoryUrl, commitSHA) {
  */
 function refreshIdentity (config) {
   clientId = uuid()
-  if (config.tags['_dd.rc.client_id']) {
-    config.tags['_dd.rc.client_id'] = clientId
-  }
   if (client !== undefined) {
+    // Written unconditionally: an RC lib-config update replaces `config.tags` with a fresh
+    // object (see `config/remote_config.js`'s `tracing_tags` transformer), dropping this
+    // directly-set key even though a client already exists.
+    config.tags['_dd.rc.client_id'] = clientId
     client.id = clientId
     client.client_tracer.runtime_id = config.tags['runtime-id']
     const { commitSHA, repositoryUrl } = getGitMetadata(config)

packages/dd-trace/src/runtime_metrics/client.js

@@ -48,10 +48,15 @@ function createMetricsClient (config) {
  *
  * @param {MetricsAggregationClient} client - The client returned by `createMetricsClient()`
  * @param {import('../config/config-base')} config - Tracer configuration
+ * @param {() => void} [onRefresh] - Called after the tag update, e.g. to rebase sampler
+ *   baselines (CPU usage, event-loop delay) that would otherwise span the snapshot pause
  * @returns {() => void} Unsubscribe function; call it from the owning module's `stop()`
  */
-function subscribeToIdentityRefresh (client, config) {
-  const onIdentityRefresh = () => client.updateTags(buildClientConfig(config).tags)
+function subscribeToIdentityRefresh (client, config, onRefresh) {
+  const onIdentityRefresh = () => {
+    client.updateTags(buildClientConfig(config).tags)
+    onRefresh?.()
+  }
   identityRefreshChannel.subscribe(onIdentityRefresh)
   return () => identityRefreshChannel.unsubscribe(onIdentityRefresh)
 }

packages/dd-trace/src/runtime_metrics/otlp_runtime_metrics.js

@@ -63,7 +63,7 @@ module.exports = {
     this.stop()
 
     client = createMetricsClient(config)
-    unsubscribeIdentityRefresh = subscribeToIdentityRefresh(client, config)
+    unsubscribeIdentityRefresh = subscribeToIdentityRefresh(client, config, resetSamplerBaselines)
     flushInterval = setInterval(() => {
       client.flush()
     }, config.DD_RUNTIME_METRICS_FLUSH_INTERVAL ?? 10_000)
@@ -251,6 +251,20 @@ module.exports = {
   },
 }
 
+/**
+ * Rebases the event-loop-delay and ELU sampler baselines to the current reading, e.g. on a
+ * MicroVM clone resume, so the next collection reports the delta since the resume instead of
+ * one spanning the snapshot pause.
+ * @returns {void}
+ */
+function resetSamplerBaselines () {
+  eventLoopHistogram?.reset()
+
+  if (lastElu !== null) {
+    lastElu = performance.eventLoopUtilization()
+  }
+}
+
 /**
  * @param {Function} callback
  * @param {object} instrument