fix(id): drop stale telemetry and RC tag on 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.

SpanStatsProcessor and AgentlessExporter also subscribed to the
identity-refresh channel per instance with no cleanup. Harmless in
production (both are process-lifetime singletons), but each of the
many instances constructed across a test run stayed subscribed
forever. Now replace the previous subscription on construction,
matching the pattern already used for the OTel logs/metrics
initializers.

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,20 @@
 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')
+
+// Only one AgentlessExporter is ever live in a real process, so replacing the subscription on
+// construction is safe - it just keeps tests (which build several) from piling up listeners.
+let unsubscribeBatchReset = null
+
 /**
  * Agentless exporter for APM trace intake.
  * Sends traces directly to the Datadog intake without requiring a local agent.
@@ -56,6 +64,12 @@ class AgentlessExporter {
       metadata,
     })
 
+    // A clone resume shouldn't flush spans buffered before the snapshot under its own identity.
+    unsubscribeBatchReset?.()
+    const onIdentityRefresh = () => this._writer.resetPendingBatch()
+    identityRefreshChannel.subscribe(onIdentityRefresh)
+    unsubscribeBatchReset = () => identityRefreshChannel.unsubscribe(onIdentityRefresh)
+
     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 whatever's queued in the encoder. Used on a MicroVM clone resume, where anything
+   * buffered before the snapshot would otherwise 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 whatever's queued. Used on a MicroVM clone resume, where log records buffered
+   * before the snapshot would otherwise 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,19 @@
 'use strict'
 
+const { channel } = require('dc-polyfill')
+
 const { buildResourceAttributes, registerResourceAttributeRefresh } = require('../resource-attributes')
 
 /**
  * @typedef {import('../../config')} Config
  */
 
+const identityRefreshChannel = channel('datadog:identity:refresh')
+
+// initializeOpenTelemetryLogs() can be called again (e.g. re-init); drop the old subscription
+// first so it doesn't stack.
+let unsubscribeLogsPendingStateReset = null
+
 /**
  * OpenTelemetry Logs Implementation for `dd-trace-js`
  *
@@ -59,6 +67,12 @@ function initializeOpenTelemetryLogs (config) {
   loggerProvider.register()
 
   registerResourceAttributeRefresh(exporter, () => buildResourceAttributes(config))
+
+  // A clone resume shouldn't export log records queued before the snapshot under its own 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,12 @@ const OtlpHttpMetricExporter = require('./otlp_http_metric_exporter')
  * @typedef {import('../../config')} Config
  */
 
+const identityRefreshChannel = channel('datadog:identity:refresh')
+
+// initializeOpenTelemetryMetrics() can be called again (e.g. re-init); drop the old subscription
+// first so it doesn't stack.
+let unsubscribeMetricsPendingStateReset = null
+
 /**
  * @file OpenTelemetry Metrics Implementation for dd-trace-js
  *
@@ -64,6 +71,12 @@ function initializeOpenTelemetryMetrics (config) {
   metrics.setGlobalMeterProvider(meterProvider)
 
   registerResourceAttributeRefresh(exporter, () => buildGeneralResourceAttributes(config))
+
+  // A clone resume shouldn't export measurements queued before the snapshot under its own 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,26 @@ class PeriodicMetricReader {
     this.#collectAndExport()
   }
 
+  /**
+   * Discards queued measurements and sync-instrument cumulative state. Used on a MicroVM clone
+   * resume so measurements recorded before the snapshot don't get exported under the 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`, and 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,10 @@ 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) {
+    // Unconditional, because an RC lib-config update rebuilds config.tags from scratch (see
+    // tracing_tags() in config/remote_config.js) and drops this directly-set key.
+    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 now. Without this, a MicroVM clone's
+ * first collection after resume would report a delta spanning the snapshot pause instead of just
+ * the time since resume.
+ * @returns {void}
+ */
+function resetSamplerBaselines () {
+  eventLoopHistogram?.reset()
+
+  if (lastElu !== null) {
+    lastElu = performance.eventLoopUtilization()
+  }
+}
+
 /**
  * @param {Function} callback
  * @param {object} instrument