fix(id): drop stale telemetry and RC tag on identity refresh

benchmark/sirun/test-optimization/index.js

@@ -128,7 +128,7 @@ assert.ok(shape, `unknown VARIANT: ${VARIANT}`)
 const trace = buildTrace(shape.tests, shape.suites, shape.wide)
 const encoder = new AgentlessCiVisibilityEncoder(
   { flush () {} },
-  { runtimeId: 'a1b2c3d4-0000-0000-0000-000000000000', service: 'my-service', env: 'ci' }
+  { tags: { 'runtime-id': 'a1b2c3d4-0000-0000-0000-000000000000', service: 'my-service', env: 'ci' } }
 )
 
 // Preflight: encode once and confirm the encoder buffered bytes and counted the

packages/dd-trace/src/ci-visibility/exporters/agentless/writer.js

@@ -19,9 +19,8 @@ const {
 class Writer extends BaseWriter {
   constructor ({ url, tags, evpProxyPrefix = '' }) {
     super(...arguments)
-    const { 'runtime-id': runtimeId, env, service } = tags
     this._url = url
-    this._encoder = new AgentlessCiVisibilityEncoder(this, { runtimeId, env, service })
+    this._encoder = new AgentlessCiVisibilityEncoder(this, { tags })
     this._evpProxyPrefix = evpProxyPrefix
   }
 

packages/dd-trace/src/ci-visibility/exporters/ci-validation/writer.js

@@ -11,9 +11,8 @@ class CiValidationWriter {
    * @param {object} options.tags tracer tags
    */
   constructor ({ sink, tags }) {
-    const { 'runtime-id': runtimeId, env, service } = tags
     this._sink = sink
-    this._encoder = new AgentlessCiVisibilityEncoder(this, { runtimeId, env, service })
+    this._encoder = new AgentlessCiVisibilityEncoder(this, { tags })
   }
 
   /**

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

@@ -4,6 +4,8 @@ const fs = require('node:fs')
 const os = require('node:os')
 const { URL, format } = require('node:url')
 
+const { channel } = require('dc-polyfill')
+
 const exporters = require('../../../../ext/exporters')
 const rfdc = require('../../../../vendor/dist/rfdc')({ proto: false, circles: false })
 const uuid = require('../../../../vendor/dist/crypto-randomuuid') // we need to keep the old uuid dep because of cypress
@@ -42,7 +44,6 @@ const {
 const { normalizeService } = require('./normalize-service')
 const { programmaticTypeCoercions, transformers } = require('./parsers')
 
-const RUNTIME_ID = uuid()
 const TEST_OPTIMIZATION_WORKER_EXPORTERS = new Set([
   exporters.CUCUMBER_WORKER,
   exporters.JEST_WORKER,
@@ -51,6 +52,21 @@ const TEST_OPTIMIZATION_WORKER_EXPORTERS = new Set([
   exporters.VITEST_WORKER,
 ])
 
+let runtimeId
+
+channel('datadog:identity:update').subscribe(refreshRuntimeId)
+
+/**
+ * Lazily generates the process-wide runtime ID on first access instead of at module load,
+ * so modules that merely require this file without constructing a Config never pay for it.
+ *
+ * @returns {string}
+ */
+function getRuntimeId () {
+  runtimeId ??= uuid()
+  return runtimeId
+}
+
 const tracerMetrics = telemetryMetrics.manager.namespace('tracers')
 
 /**
@@ -597,7 +613,7 @@ class Config extends ConfigBase {
     if (this.version) {
       this.tags.version = this.version
     }
-    this.tags['runtime-id'] = RUNTIME_ID
+    this.tags['runtime-id'] = getRuntimeId()
     const platformTags = getServerlessPlatformTags()
     if (platformTags) {
       for (let i = 0; i < platformTags.length; i += 2) {
@@ -783,3 +799,15 @@ function getConfig (options) {
   }
   return configInstance
 }
+
+/**
+ * Regenerates the runtime ID.
+ *
+ * Used for Lambda MicroVM `/run` lifecycle hooks, giving each clone a distinct runtime identity.
+ *
+ * @param {import('./config-base')} config
+ */
+function refreshRuntimeId (config) {
+  runtimeId = uuid()
+  config.tags['runtime-id'] = runtimeId
+}

packages/dd-trace/src/crashtracking/crashtracker.js

@@ -6,10 +6,13 @@ const { EOL, platform } = require('node:os')
 const libdatadog = require('@datadog/libdatadog')
 const binding = libdatadog.load('crashtracker')
 
+const { channel } = require('dc-polyfill')
 const log = require('../log')
 const pkg = require('../../../../package.json')
 const processTags = require('../process-tags')
 
+const identityRefreshChannel = channel('datadog:identity:refresh')
+
 class Crashtracker {
   #started = false
 
@@ -38,6 +41,7 @@ class Crashtracker {
       )
       this.#started = true
       this.#trackUnhandledExceptions()
+      identityRefreshChannel.subscribe((config) => this.configure(config))
     } catch (e) {
       log.error('Error initializing crashtracker', e)
     }

packages/dd-trace/src/dogstatsd.js

@@ -3,6 +3,7 @@
 const dgram = require('dgram')
 const isIP = require('net').isIP
 
+const { channel } = require('dc-polyfill')
 const { storage } = require('../../datadog-core')
 const request = require('./exporters/common/request')
 const log = require('./log')
@@ -18,6 +19,8 @@ const TYPE_GAUGE = 'g'
 const TYPE_DISTRIBUTION = 'd'
 const TYPE_HISTOGRAM = 'h'
 
+const identityRefreshChannel = channel('datadog:identity:refresh')
+
 /**
  * @import { DogStatsD } from "../../../index.d.ts"
  * @implements {DogStatsD}
@@ -39,14 +42,42 @@ class DogStatsDClient {
     this._family = isIP(this._host)
     this._port = options.port
     this._tags = options.tags
-    this.#tagsPrefix = this._tags?.length ? `|#${this._tags.join(',')}` : ''
+    this.#tagsPrefix = this._tags.length ? `|#${this._tags.join(',')}` : ''
     this._queue = []
     this._buffer = ''
     this._offset = 0
     this._udp4 = this._socket('udp4')
     this._udp6 = this._socket('udp6')
   }
 
+  /**
+   * Recomputes the cached tags and tag-prefix (mirrors the constructor) after a `config.tags`
+   * change, e.g. a MicroVM clone resume.
+   *
+   * Buffered lines have the old prefix baked in, and on a clone resume they were produced during
+   * the image build, so every clone holds the same bytes — flushing them would submit one identical
+   * copy per clone. Dropping is right here for that reason only: for a tag change on a live process
+   * the buffer holds unique data whose old tags are still correct, so that case wants a flush
+   * 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 false
+
+    this.#tagsPrefix = tagsPrefix
+    this._queue = []
+    this._buffer = ''
+    this._offset = 0
+
+    return true
+  }
+
   increment (stat, value, tags) {
     this._add(stat, value, TYPE_COUNTER, tags)
   }
@@ -212,6 +243,19 @@ class MetricsAggregationClient {
     this.reset()
   }
 
+  /**
+   * 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) {
+    if (this._client.updateTags(tags)) {
+      this.reset()
+    }
+  }
+
   flush () {
     this._captureCounters()
     this._captureGauges()
@@ -364,6 +408,11 @@ class CustomMetrics {
     const clientConfig = DogStatsDClient.generateClientConfig(config)
     this.#client = new MetricsAggregationClient(new DogStatsDClient(clientConfig))
 
+    // CustomMetrics has process-lifetime flush handlers and no stop hook, so this shares that lifetime.
+    identityRefreshChannel.subscribe(() => {
+      this.#client.updateTags(DogStatsDClient.generateClientConfig(config).tags)
+    })
+
     const flush = this.flush.bind(this)
 
     // TODO(bengl) this magic number should be configurable

packages/dd-trace/src/encode/agentless-ci-visibility.js

@@ -69,11 +69,11 @@ function truncateTestLevelMetadataTags (tags) {
 }
 
 class AgentlessCiVisibilityEncoder extends AgentEncoder {
-  constructor (writer, { runtimeId, service, env }) {
+  constructor (writer, { tags }) {
     super(writer, INTAKE_SOFT_LIMIT)
-    this.runtimeId = runtimeId
-    this.service = service
-    this.env = env
+    // Holds a reference to the live `tags` object (instead of copying `env`/`runtime-id` out of it)
+    // so a later change (e.g. a MicroVM clone resume) is picked up at flush time.
+    this.tags = tags
 
     // Used to keep track of the number of encoded events to update the
     // length of `payload.events` when calling `makePayload`
@@ -406,11 +406,12 @@ class AgentlessCiVisibilityEncoder extends AgentEncoder {
       events: [],
     }
 
-    if (this.env) {
-      payload.metadata['*'].env = this.env
+    if (this.tags.env) {
+      payload.metadata['*'].env = this.tags.env
     }
-    if (this.runtimeId) {
-      payload.metadata['*']['runtime-id'] = this.runtimeId
+    const runtimeId = this.tags['runtime-id']
+    if (runtimeId) {
+      payload.metadata['*']['runtime-id'] = runtimeId
     }
 
     bytes.writeMapPrefix(Object.keys(payload).length)

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.
@@ -23,7 +31,7 @@ class AgentlessExporter {
    * @param {string} [config.site] - The Datadog site. Defaults to 'datadoghq.com'.
    * @param {number} [config.flushInterval] - Batch flush interval in ms
    * @param {string} [config.env] - Environment name
-   * @param {object} [config.tags] - Tags including runtime-id
+   * @param {object} config.tags - Tags including runtime-id
    */
   constructor (config) {
     this.#config = config
@@ -40,11 +48,13 @@ class AgentlessExporter {
 
     const metadata = {
       hostname: os.hostname(),
-      env: config.env,
       languageName: 'nodejs',
       languageVersion: process.version,
       tracerVersion,
-      runtimeID: config.tags?.['runtime-id'],
+      // Read live off `config` (instead of copying the value) so a later change
+      // (e.g. a MicroVM clone resume) is picked up by the next `JSON.stringify` in the encoder.
+      get env () { return config.env },
+      get runtimeID () { return config.tags['runtime-id'] },
       ...(entityId ? { containerID: entityId } : {}),
     }
 
@@ -54,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/id.js

@@ -2,13 +2,17 @@
 
 const { randomFillSync } = require('crypto')
 
+const { channel } = require('dc-polyfill')
+
 const UINT_MAX = 4_294_967_296
 
 const data = new Uint8Array(8 * 8192)
 const zeroId = new Uint8Array(8)
 
 let batch = 0
 
+channel('datadog:identity:update').subscribe(reseed)
+
 // Internal representation of a trace or span ID.
 class Identifier {
   /** @type {number[] | Uint8Array} */
@@ -254,6 +258,16 @@ function writeUInt32BE (buffer, value, offset) {
   buffer[0 + offset] = value & 255
 }
 
+/**
+ * Resets the batch cursor, forcing the next ID batch to draw a fresh
+ * randomFillSync() call on MicroVM clone resume. Node's crypto RNG is
+ * re-seeded from the kernel CSPRNG on snapshot resume, so re-invoking it
+ * is sufficient — no need to read /dev/urandom directly.
+ */
+function reseed () {
+  batch = 0
+}
+
 /**
  * @param {string} [value]
  * @param {number} [radix]

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