diff --git a/benchmark/sirun/test-optimization/index.js b/benchmark/sirun/test-optimization/index.js index b70476ed08e..469d11dafbb 100644 --- a/benchmark/sirun/test-optimization/index.js +++ b/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 diff --git a/packages/dd-trace/src/ci-visibility/exporters/agentless/writer.js b/packages/dd-trace/src/ci-visibility/exporters/agentless/writer.js index 2fad9a633a5..8d494c86d1d 100644 --- a/packages/dd-trace/src/ci-visibility/exporters/agentless/writer.js +++ b/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 } diff --git a/packages/dd-trace/src/ci-visibility/exporters/ci-validation/writer.js b/packages/dd-trace/src/ci-visibility/exporters/ci-validation/writer.js index f097cbee045..d856c21e1c1 100644 --- a/packages/dd-trace/src/ci-visibility/exporters/ci-validation/writer.js +++ b/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 }) } /** diff --git a/packages/dd-trace/src/config/index.js b/packages/dd-trace/src/config/index.js index 0c5a71cc982..bb69815c54c 100644 --- a/packages/dd-trace/src/config/index.js +++ b/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 +} diff --git a/packages/dd-trace/src/crashtracking/crashtracker.js b/packages/dd-trace/src/crashtracking/crashtracker.js index 5929abb4337..b53dbb9b3bb 100644 --- a/packages/dd-trace/src/crashtracking/crashtracker.js +++ b/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) } diff --git a/packages/dd-trace/src/dogstatsd.js b/packages/dd-trace/src/dogstatsd.js index 19a580ad2db..c2898b4a797 100644 --- a/packages/dd-trace/src/dogstatsd.js +++ b/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,7 +42,7 @@ 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 @@ -47,6 +50,34 @@ class DogStatsDClient { 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 diff --git a/packages/dd-trace/src/encode/agentless-ci-visibility.js b/packages/dd-trace/src/encode/agentless-ci-visibility.js index c494622d757..ac58c33ee42 100644 --- a/packages/dd-trace/src/encode/agentless-ci-visibility.js +++ b/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) diff --git a/packages/dd-trace/src/exporters/agentless/index.js b/packages/dd-trace/src/exporters/agentless/index.js index b38a48f7688..e2958da85c6 100644 --- a/packages/dd-trace/src/exporters/agentless/index.js +++ b/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)) diff --git a/packages/dd-trace/src/exporters/common/writer.js b/packages/dd-trace/src/exporters/common/writer.js index 38d12562a88..044d57d4933 100644 --- a/packages/dd-trace/src/exporters/common/writer.js +++ b/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 diff --git a/packages/dd-trace/src/id.js b/packages/dd-trace/src/id.js index b0006e44baf..688d9eebcb9 100644 --- a/packages/dd-trace/src/id.js +++ b/packages/dd-trace/src/id.js @@ -2,6 +2,8 @@ const { randomFillSync } = require('crypto') +const { channel } = require('dc-polyfill') + const UINT_MAX = 4_294_967_296 const data = new Uint8Array(8 * 8192) @@ -9,6 +11,8 @@ 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] diff --git a/packages/dd-trace/src/opentelemetry/logs/batch_log_processor.js b/packages/dd-trace/src/opentelemetry/logs/batch_log_processor.js index 46e8ba6c16a..5e6ce1a3808 100644 --- a/packages/dd-trace/src/opentelemetry/logs/batch_log_processor.js +++ b/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 diff --git a/packages/dd-trace/src/opentelemetry/logs/index.js b/packages/dd-trace/src/opentelemetry/logs/index.js index d6a40ad0122..de463f24f4f 100644 --- a/packages/dd-trace/src/opentelemetry/logs/index.js +++ b/packages/dd-trace/src/opentelemetry/logs/index.js @@ -1,11 +1,19 @@ 'use strict' -const os = require('os') +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` * @@ -36,35 +44,13 @@ const OtlpHttpLogExporter = require('./otlp_http_log_exporter') * @param {import('../../config/config-base')} config - Tracer configuration instance */ function initializeOpenTelemetryLogs (config) { - // Build resource attributes - const resourceAttributes = { - 'service.name': config.service, - 'service.version': config.version, - 'deployment.environment': config.env, - } - - // Add all tracer tags (includes DD_TAGS, OTEL_RESOURCE_ATTRIBUTES, DD_TRACE_TAGS, etc.) - // Exclude Datadog-style keys that duplicate OpenTelemetry standard keys - if (config.tags) { - const filteredTags = { ...config.tags } - delete filteredTags.service - delete filteredTags.version - delete filteredTags.env - Object.assign(resourceAttributes, filteredTags) - } - - // Add host.name if reportHostname is enabled - if (config.reportHostname) { - resourceAttributes['host.name'] = os.hostname() - } - // Create OTLP exporter using resolved config values const exporter = new OtlpHttpLogExporter( config.OTEL_EXPORTER_OTLP_LOGS_ENDPOINT, config.OTEL_EXPORTER_OTLP_LOGS_HEADERS, config.OTEL_EXPORTER_OTLP_LOGS_TIMEOUT, config.OTEL_EXPORTER_OTLP_LOGS_PROTOCOL, - resourceAttributes + buildResourceAttributes(config) ) // Create batch processor for exporting logs to Datadog Agent @@ -79,6 +65,14 @@ function initializeOpenTelemetryLogs (config) { // Register the logger provider globally with OpenTelemetry API 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 = { diff --git a/packages/dd-trace/src/opentelemetry/logs/otlp_http_log_exporter.js b/packages/dd-trace/src/opentelemetry/logs/otlp_http_log_exporter.js index d01b0cec5c2..510e002a849 100644 --- a/packages/dd-trace/src/opentelemetry/logs/otlp_http_log_exporter.js +++ b/packages/dd-trace/src/opentelemetry/logs/otlp_http_log_exporter.js @@ -4,7 +4,7 @@ const OtlpHttpExporterBase = require('../otlp/otlp_http_exporter_base') const OtlpTransformer = require('./otlp_transformer') /** - * @typedef {import('@opentelemetry/resources').Resource} Resource + * @typedef {import('@opentelemetry/api').Attributes} Attributes * @typedef {import('@opentelemetry/api-logs').LogRecord} LogRecord */ @@ -26,11 +26,21 @@ class OtlpHttpLogExporter extends OtlpHttpExporterBase { * corresponding `OTEL_EXPORTER_OTLP_*_HEADERS` env by the MAP parser. * @param {number} timeout - Request timeout in milliseconds * @param {string} protocol - OTLP protocol (http/protobuf or http/json) - * @param {Resource} resource - Resource attributes + * @param {Attributes} resourceAttributes - Resource attributes */ - constructor (url, headers, timeout, protocol, resource) { + constructor (url, headers, timeout, protocol, resourceAttributes) { super(url, headers, timeout, protocol, 'logs') - this.transformer = new OtlpTransformer(resource, protocol) + this.transformer = new OtlpTransformer(resourceAttributes, protocol) + } + + /** + * Recomputes the resource attributes baked into the transformer (e.g. after a MicroVM clone + * resume regenerates `runtime-id`). + * + * @param {Attributes} resourceAttributes - Resource attributes + */ + updateResourceAttributes (resourceAttributes) { + this.transformer.updateResourceAttributes(resourceAttributes) } /** diff --git a/packages/dd-trace/src/opentelemetry/metrics/index.js b/packages/dd-trace/src/opentelemetry/metrics/index.js index e9e16910bc7..7ddfac7d43f 100644 --- a/packages/dd-trace/src/opentelemetry/metrics/index.js +++ b/packages/dd-trace/src/opentelemetry/metrics/index.js @@ -3,9 +3,14 @@ const os = require('os') const { metrics } = require('@opentelemetry/api') +const { channel } = require('dc-polyfill') const { VERSION } = require('../../../../../version') const processTags = require('../../process-tags') +const { + buildResourceAttributes: buildGeneralResourceAttributes, + registerResourceAttributeRefresh, +} = require('../resource-attributes') const MeterProvider = require('./meter_provider') const PeriodicMetricReader = require('./periodic_metric_reader') const OtlpHttpMetricExporter = require('./otlp_http_metric_exporter') @@ -14,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 * @@ -41,30 +52,12 @@ const OtlpHttpMetricExporter = require('./otlp_http_metric_exporter') * @param {import('../../config/config-base')} config - Tracer configuration instance */ function initializeOpenTelemetryMetrics (config) { - const resourceAttributes = { - 'service.name': config.service, - 'service.version': config.version, - 'deployment.environment': config.env, - } - - if (config.tags) { - const filteredTags = { ...config.tags } - delete filteredTags.service - delete filteredTags.version - delete filteredTags.env - Object.assign(resourceAttributes, filteredTags) - } - - if (config.reportHostname) { - resourceAttributes['host.name'] = os.hostname() - } - const exporter = new OtlpHttpMetricExporter( config.OTEL_EXPORTER_OTLP_METRICS_ENDPOINT, config.OTEL_EXPORTER_OTLP_METRICS_HEADERS, config.OTEL_EXPORTER_OTLP_METRICS_TIMEOUT, config.OTEL_EXPORTER_OTLP_METRICS_PROTOCOL, - resourceAttributes + buildGeneralResourceAttributes(config) ) const reader = new PeriodicMetricReader( @@ -76,6 +69,14 @@ function initializeOpenTelemetryMetrics (config) { const meterProvider = new MeterProvider({ reader }) 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 } = {}) { @@ -104,27 +105,30 @@ function buildResourceAttributes (tags, { reportHostname, otelSemanticsEnabled, function createOtlpSpanStatsExporter (config) { const { OtlpStatsExporter } = require('./otlp_span_stats_exporter') const protocol = config.OTEL_EXPORTER_OTLP_METRICS_PROTOCOL || 'http/json' - const resourceAttributes = buildResourceAttributes(config.tags, { + const buildSpanStatsResourceAttributes = () => buildResourceAttributes(config.tags, { reportHostname: config.reportHostname, otelSemanticsEnabled: config.DD_TRACE_OTEL_SEMANTICS_ENABLED, service: config.service, env: config.env, serviceVersion: config.version, }) - return new OtlpStatsExporter( + const exporter = new OtlpStatsExporter( config.OTEL_EXPORTER_OTLP_METRICS_ENDPOINT, protocol, - resourceAttributes, + buildSpanStatsResourceAttributes(), config.DD_TRACE_OTEL_SEMANTICS_ENABLED, config.service, config.OTEL_EXPORTER_OTLP_METRICS_HEADERS, config.OTEL_EXPORTER_OTLP_METRICS_TIMEOUT ) + + registerResourceAttributeRefresh(exporter, buildSpanStatsResourceAttributes) + + return exporter } module.exports = { MeterProvider, initializeOpenTelemetryMetrics, - buildResourceAttributes, createOtlpSpanStatsExporter, } diff --git a/packages/dd-trace/src/opentelemetry/metrics/otlp_http_metric_exporter.js b/packages/dd-trace/src/opentelemetry/metrics/otlp_http_metric_exporter.js index 8af42b70854..185225bc796 100644 --- a/packages/dd-trace/src/opentelemetry/metrics/otlp_http_metric_exporter.js +++ b/packages/dd-trace/src/opentelemetry/metrics/otlp_http_metric_exporter.js @@ -4,7 +4,7 @@ const OtlpHttpExporterBase = require('../otlp/otlp_http_exporter_base') const OtlpTransformer = require('./otlp_transformer') /** - * @typedef {import('@opentelemetry/resources').Resource} Resource + * @typedef {import('@opentelemetry/api').Attributes} Attributes * @typedef {import('./periodic_metric_reader').AggregatedMetric} AggregatedMetric */ @@ -22,11 +22,21 @@ class OtlpHttpMetricExporter extends OtlpHttpExporterBase { * corresponding `OTEL_EXPORTER_OTLP_*_HEADERS` env by the MAP parser. * @param {number} timeout - Request timeout in milliseconds * @param {string} protocol - OTLP protocol (http/protobuf or http/json) - * @param {Resource} resource - Resource attributes + * @param {Attributes} resourceAttributes - Resource attributes */ - constructor (url, headers, timeout, protocol, resource) { + constructor (url, headers, timeout, protocol, resourceAttributes) { super(url, headers, timeout, protocol, 'metrics') - this.transformer = new OtlpTransformer(resource, protocol) + this.transformer = new OtlpTransformer(resourceAttributes, protocol) + } + + /** + * Recomputes the resource attributes baked into the transformer (e.g. after a MicroVM clone + * resume regenerates `runtime-id`). Leaves the reader's export interval untouched. + * + * @param {Attributes} resourceAttributes - Resource attributes + */ + updateResourceAttributes (resourceAttributes) { + this.transformer.updateResourceAttributes(resourceAttributes) } /** diff --git a/packages/dd-trace/src/opentelemetry/metrics/otlp_span_stats_exporter.js b/packages/dd-trace/src/opentelemetry/metrics/otlp_span_stats_exporter.js index 018493dbf5c..63197a870c9 100644 --- a/packages/dd-trace/src/opentelemetry/metrics/otlp_span_stats_exporter.js +++ b/packages/dd-trace/src/opentelemetry/metrics/otlp_span_stats_exporter.js @@ -22,6 +22,16 @@ class OtlpStatsExporter extends OtlpHttpExporterBase { this.#transformer = new OtlpStatsTransformer(resourceAttributes, protocol, otelSemanticsEnabled, defaultService) } + /** + * Recomputes the resource attributes baked into the transformer (e.g. after a MicroVM clone + * resume regenerates `runtime-id`). + * + * @param {import('@opentelemetry/api').Attributes} resourceAttributes + */ + updateResourceAttributes (resourceAttributes) { + this.#transformer.updateResourceAttributes(resourceAttributes) + } + /** * @param {Array<{timeNs: number, bucket: import('../../span_stats').SpanBuckets}>} drained * @param {number} bucketSizeNs diff --git a/packages/dd-trace/src/opentelemetry/metrics/periodic_metric_reader.js b/packages/dd-trace/src/opentelemetry/metrics/periodic_metric_reader.js index a97b5cb6c99..8607fee6c08 100644 --- a/packages/dd-trace/src/opentelemetry/metrics/periodic_metric_reader.js +++ b/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} diff --git a/packages/dd-trace/src/opentelemetry/otlp/otlp_transformer_base.js b/packages/dd-trace/src/opentelemetry/otlp/otlp_transformer_base.js index cd522c8facc..10263733542 100644 --- a/packages/dd-trace/src/opentelemetry/otlp/otlp_transformer_base.js +++ b/packages/dd-trace/src/opentelemetry/otlp/otlp_transformer_base.js @@ -75,6 +75,16 @@ class OtlpTransformerBase { } } + /** + * Recomputes the cached OTLP resource attributes (e.g. after `config.tags` changes post- + * construction, such as a MicroVM clone resume regenerating its runtime-id). Leaves the + * reader/exporter/protocol otherwise untouched. + * @param {Attributes} resourceAttributes - Resource attributes + */ + updateResourceAttributes (resourceAttributes) { + this.#resourceAttributes = this.transformAttributes(resourceAttributes) + } + /** * Transforms attributes to OTLP KeyValue format. * @param {Attributes} attributes - Attributes to transform diff --git a/packages/dd-trace/src/opentelemetry/resource-attributes.js b/packages/dd-trace/src/opentelemetry/resource-attributes.js new file mode 100644 index 00000000000..ae9baefcf98 --- /dev/null +++ b/packages/dd-trace/src/opentelemetry/resource-attributes.js @@ -0,0 +1,55 @@ +'use strict' + +const os = require('node:os') + +const { channel } = require('dc-polyfill') + +const identityRefreshChannel = channel('datadog:identity:refresh') +const resourceAttributeRefreshers = new Map() + +function refreshActiveResourceAttributes () { + for (const refreshResourceAttributes of resourceAttributeRefreshers.values()) { + refreshResourceAttributes() + } +} + +/** + * @typedef {import('@opentelemetry/api').Attributes} Attributes + * @typedef {{ + * signalType: string, + * updateResourceAttributes: (resourceAttributes: Attributes) => void + * }} ResourceAttributeExporter + */ + +/** + * @param {import('../config/config-base')} config + * @returns {Attributes} + */ +function buildResourceAttributes (config) { + const { service, version, env, ...tags } = config.tags + const resourceAttributes = { + 'service.name': config.service, + 'service.version': config.version, + 'deployment.environment': config.env, + ...tags, + } + + if (config.reportHostname) resourceAttributes['host.name'] = os.hostname() + + return resourceAttributes +} + +/** + * @param {ResourceAttributeExporter} exporter + * @param {() => Attributes} buildResourceAttributes + */ +function registerResourceAttributeRefresh (exporter, buildResourceAttributes) { + if (resourceAttributeRefreshers.size === 0) { + identityRefreshChannel.subscribe(refreshActiveResourceAttributes) + } + resourceAttributeRefreshers.set(exporter.signalType, () => { + exporter.updateResourceAttributes(buildResourceAttributes()) + }) +} + +module.exports = { buildResourceAttributes, registerResourceAttributeRefresh } diff --git a/packages/dd-trace/src/opentelemetry/trace/index.js b/packages/dd-trace/src/opentelemetry/trace/index.js index e3013cd2c51..6c5a9ae7f46 100644 --- a/packages/dd-trace/src/opentelemetry/trace/index.js +++ b/packages/dd-trace/src/opentelemetry/trace/index.js @@ -1,6 +1,7 @@ 'use strict' const { VERSION } = require('../../../../../version') +const { registerResourceAttributeRefresh } = require('../resource-attributes') const OtlpHttpTraceExporter = require('./otlp_http_trace_exporter') /** @@ -59,13 +60,17 @@ function buildResourceAttributes (config) { * @returns {OtlpHttpTraceExporter} The OTLP HTTP/JSON exporter */ function createOtlpTraceExporter (config) { - return new OtlpHttpTraceExporter( + const exporter = new OtlpHttpTraceExporter( config.OTEL_EXPORTER_OTLP_TRACES_ENDPOINT, config.OTEL_EXPORTER_OTLP_TRACES_HEADERS, config.OTEL_EXPORTER_OTLP_TRACES_TIMEOUT, buildResourceAttributes(config), config.DD_TRACE_OTEL_SEMANTICS_ENABLED ) + + registerResourceAttributeRefresh(exporter, () => buildResourceAttributes(config)) + + return exporter } module.exports = { diff --git a/packages/dd-trace/src/opentelemetry/trace/otlp_http_trace_exporter.js b/packages/dd-trace/src/opentelemetry/trace/otlp_http_trace_exporter.js index 2afdb454b4f..6c87c64f499 100644 --- a/packages/dd-trace/src/opentelemetry/trace/otlp_http_trace_exporter.js +++ b/packages/dd-trace/src/opentelemetry/trace/otlp_http_trace_exporter.js @@ -43,6 +43,16 @@ class OtlpHttpTraceExporter extends OtlpHttpExporterBase { this.#transformer = new OtlpTraceTransformer(resourceAttributes, otelTraceSemanticsEnabled) } + /** + * Recomputes the resource attributes baked into the transformer (e.g. after a MicroVM clone + * resume regenerates `runtime-id`). + * + * @param {import('@opentelemetry/api').Attributes} resourceAttributes - Resource attributes + */ + updateResourceAttributes (resourceAttributes) { + this.#transformer.updateResourceAttributes(resourceAttributes) + } + /** * Exports DD-formatted spans via OTLP over HTTP. * diff --git a/packages/dd-trace/src/profiling/config.js b/packages/dd-trace/src/profiling/config.js index 87a25c797df..07ece161e7f 100644 --- a/packages/dd-trace/src/profiling/config.js +++ b/packages/dd-trace/src/profiling/config.js @@ -222,4 +222,5 @@ function buildProfilingRuntime (config) { module.exports = { buildProfilingRuntime, + getProfilingTags, } diff --git a/packages/dd-trace/src/profiling/profiler.js b/packages/dd-trace/src/profiling/profiler.js index e071cd93148..43ca8f41fab 100644 --- a/packages/dd-trace/src/profiling/profiler.js +++ b/packages/dd-trace/src/profiling/profiler.js @@ -4,7 +4,7 @@ const { EventEmitter } = require('events') const dc = require('dc-polyfill') const crashtracker = require('../crashtracking') const log = require('../log') -const { buildProfilingRuntime } = require('./config') +const { buildProfilingRuntime, getProfilingTags } = require('./config') const { snapshotKinds } = require('./constants') const { threadNamePrefix } = require('./profilers/shared') const { isWebServerSpan, endpointNameFromTags, getStartedSpans } = require('./webspan-utils') @@ -54,6 +54,7 @@ class Profiler extends EventEmitter { #compressionFn #compressionFnInitialized = false #compressionOptions + #config #customLabelKeys = new Set() #enabled = false #endpointCounts = new Map() @@ -64,7 +65,6 @@ class Profiler extends EventEmitter { #profilers #spanFinishListener #systemInfoReport - #tags #timer #uploadCompression @@ -172,9 +172,9 @@ class Profiler extends EventEmitter { if (this.enabled) return true this.#enabled = true - const { tags, exporters, flushInterval, profilers, uploadCompression, systemInfoReport } = + const { exporters, flushInterval, profilers, uploadCompression, systemInfoReport } = buildProfilingRuntime(config) - this.#tags = tags + this.#config = config this.#exporters = exporters this.#flushInterval = flushInterval this.#profilers = profilers @@ -389,7 +389,7 @@ class Profiler extends EventEmitter { } #submit (profiles, infos, start, end, snapshotKind) { - const tags = this.#tags + const tags = getProfilingTags(this.#config) // Flatten endpoint counts const endpointCounts = {} diff --git a/packages/dd-trace/src/profiling/profilers/space.js b/packages/dd-trace/src/profiling/profilers/space.js index a5c8ba25641..d3578963186 100644 --- a/packages/dd-trace/src/profiling/profilers/space.js +++ b/packages/dd-trace/src/profiling/profilers/space.js @@ -1,5 +1,8 @@ 'use strict' +const { channel } = require('dc-polyfill') + +const log = require('../../log') const { oomExportStrategies, ensureOOMExportStrategies, strategiesToCallbackMode, buildExportCommand } = require('../oom') const { encodeProfileAsync, getThreadLabels } = require('./shared') @@ -11,10 +14,12 @@ const { encodeProfileAsync, getThreadLabels } = require('./shared') */ const STACK_DEPTH = 64 +const identityRefreshChannel = channel('datadog:identity:refresh') class NativeSpaceProfiler { #config #exporters + #identityRefreshListener #mapper #pprof #samplingInterval @@ -39,20 +44,36 @@ class NativeSpaceProfiler { start ({ mapper, nearOOMCallback } = {}) { if (this.#started) return - const config = this.#config this.#mapper = mapper this.#pprof = require('@datadog/pprof') - this.#pprof.heap.start(this.#samplingInterval, STACK_DEPTH, config.DD_PROFILING_ALLOCATION_ENABLED) - if (config.DD_PROFILING_EXPERIMENTAL_OOM_MONITORING_ENABLED) { + this.#pprof.heap.start(this.#samplingInterval, STACK_DEPTH, this.#config.DD_PROFILING_ALLOCATION_ENABLED) + + if (this.#config.DD_PROFILING_EXPERIMENTAL_OOM_MONITORING_ENABLED) { + const config = this.#config const strategies = ensureOOMExportStrategies(config.DD_PROFILING_EXPERIMENTAL_OOM_EXPORT_STRATEGIES) - this.#pprof.heap.monitorOutOfMemory( - config.DD_PROFILING_EXPERIMENTAL_OOM_HEAP_LIMIT_EXTENSION_SIZE, - config.DD_PROFILING_EXPERIMENTAL_OOM_MAX_HEAP_EXTENSION_COUNT, - strategies.includes(oomExportStrategies.LOGS), - strategies.includes(oomExportStrategies.PROCESS) ? buildExportCommand(this.#exporters, this.#tags) : [], - (profile) => nearOOMCallback(this.type, this.#pprof.encodeSync(profile), this.getInfo()), - strategiesToCallbackMode(strategies, this.#pprof.heap.CallbackMode) - ) + const monitorOutOfMemory = () => { + this.#pprof.heap.monitorOutOfMemory( + config.DD_PROFILING_EXPERIMENTAL_OOM_HEAP_LIMIT_EXTENSION_SIZE, + config.DD_PROFILING_EXPERIMENTAL_OOM_MAX_HEAP_EXTENSION_COUNT, + strategies.includes(oomExportStrategies.LOGS), + strategies.includes(oomExportStrategies.PROCESS) ? buildExportCommand(this.#exporters, this.#tags) : [], + (profile) => nearOOMCallback(this.type, this.#pprof.encodeSync(profile), this.getInfo()), + strategiesToCallbackMode(strategies, this.#pprof.heap.CallbackMode) + ) + } + monitorOutOfMemory() + + if (strategies.includes(oomExportStrategies.PROCESS)) { + this.#identityRefreshListener = () => { + try { + Object.assign(this.#tags, config.tags) + monitorOutOfMemory() + } catch (error) { + log.error(error) + } + } + identityRefreshChannel.subscribe(this.#identityRefreshListener) + } } this.#started = true @@ -76,6 +97,12 @@ class NativeSpaceProfiler { stop () { if (!this.#started) return + + if (this.#identityRefreshListener !== undefined) { + identityRefreshChannel.unsubscribe(this.#identityRefreshListener) + this.#identityRefreshListener = undefined + } + this.#pprof.heap.stop() this.#started = false } diff --git a/packages/dd-trace/src/proxy.js b/packages/dd-trace/src/proxy.js index 443bc3a4872..3c128ac1618 100644 --- a/packages/dd-trace/src/proxy.js +++ b/packages/dd-trace/src/proxy.js @@ -1,5 +1,7 @@ 'use strict' +const { channel } = require('dc-polyfill') +const uuid = require('../../../vendor/dist/crypto-randomuuid') const NoopProxy = require('./noop/proxy') const { features } = require('./feature-registry') const DatadogTracer = require('./tracer') @@ -13,7 +15,7 @@ const telemetry = require('./telemetry') const nomenclature = require('./service-naming') const PluginManager = require('./plugin_manager') const NoopDogStatsDClient = require('./noop/dogstatsd') -const { IS_SERVERLESS } = require('./serverless') +const { IS_AWS_LAMBDA_MICROVM, IS_SERVERLESS, NODE_BUNDLES_OPENSSL } = require('./serverless') const processTags = require('./process-tags') const { isTrue } = require('./util') const { @@ -42,6 +44,13 @@ const FEATURE_STATE_NOOP = 0 const FEATURE_STATE_LAZY = 1 const FEATURE_STATE_ACTIVE = 2 +const UUID_POOL_SIZE = 128 + +const BUNDLED_OPENSSL_WARNING = 'This Node.js build bundles its own OpenSSL, so its random number ' + + 'generator is not reseeded when a MicroVM clone resumes and trace IDs may repeat across clones. ' + + 'Install Node.js from the Amazon Linux 2023 repositories (for example `dnf install nodejs22`) so it ' + + 'links the base image\'s snapsafe OpenSSL.' + class LazyModule { constructor (provider) { this.provider = provider @@ -139,6 +148,10 @@ class Tracer extends NoopProxy { try { const config = getConfig(options) // TODO: support dynamic code config + if (IS_AWS_LAMBDA_MICROVM) { + this.#registerMicroVmRunHook(config) + } + // Add config dependent process tags processTags.initialize(config) @@ -288,6 +301,41 @@ class Tracer extends NoopProxy { return this } + /** + * Listens for the MicroVM /run lifecycle event and triggers a one-time + * identity reset on first fire. + * + * @param {import('./config/config-base')} config + */ + #registerMicroVmRunHook (config) { + // Node reseeds its CSPRNG from the kernel on clone resume only when it links the base image's + // snapsafe libcrypto; a bundled OpenSSL keeps the snapshot's DRBG state, so the refresh below + // cannot produce distinct IDs. Logged at registration, during the image build, where the + // Dockerfile can still be fixed — not once per clone. + if (NODE_BUNDLES_OPENSSL) { + log.warn(BUNDLED_OPENSSL_WARNING) + } + + const ch = channel('http.server.request.start') + + const onHttpRequest = ({ request }) => { + if (request.method === 'POST' && request.url === '/aws/lambda-microvms/runtime/v1/run') { + ch.unsubscribe(onHttpRequest) + drainUuidPool() + channel('datadog:identity:update').publish(config) + if (this._tracingInitialized) { + const metadata = require('./tracer_metadata')(config) + if (metadata === undefined) { + log.warn('Could not store tracer configuration for service discovery') + } + } + channel('datadog:identity:refresh').publish(config) + } + } + + ch.subscribe(onHttpRequest) + } + /** * @param {import('./config/config-base')} config - Tracer configuration */ @@ -478,4 +526,19 @@ function isOfflineValidationExporter (options) { return OFFLINE_VALIDATION_EXPORTERS.has(options?.experimental?.exporter) } +/** + * Discards Node's buffered UUID entropy, so the identity subscribers draw from bytes generated + * after the clone resumed. + * + * `crypto.randomUUID()` serves `kBatchSize` (128, see `lib/internal/crypto/random.js`) UUIDs from one + * process-wide pool and refills it only when its cursor wraps back to 0. A full cycle crosses the + * cursor exactly once from any starting position, and the position is not observable, so the count + * has to be the batch size rather than the number of UUIDs we need. + */ +function drainUuidPool () { + for (let index = 0; index < UUID_POOL_SIZE; index++) { + uuid() + } +} + module.exports = Tracer diff --git a/packages/dd-trace/src/remote_config/index.js b/packages/dd-trace/src/remote_config/index.js index ebbce0b8efe..db2c17b3119 100644 --- a/packages/dd-trace/src/remote_config/index.js +++ b/packages/dd-trace/src/remote_config/index.js @@ -1,5 +1,7 @@ 'use strict' +const { channel } = require('dc-polyfill') + const uuid = require('../../../../vendor/dist/crypto-randomuuid') const tracerVersion = require('../../../../package.json').version const request = require('../exporters/common/request') @@ -12,7 +14,11 @@ const processTags = require('../process-tags') const Scheduler = require('./scheduler') const { UNACKNOWLEDGED, ACKNOWLEDGED, ERROR } = require('./apply_states') -const clientId = uuid() +let clientId = uuid() +/** @type {{ id: string, client_tracer: { runtime_id: string, tags: string[] } } | undefined} */ +let client + +channel('datadog:identity:update').subscribe(refreshIdentity) const DEFAULT_CAPABILITY = Buffer.alloc(1).toString('base64') // 0x00 @@ -38,13 +44,6 @@ class RemoteConfig { }) const { commitSHA, repositoryUrl } = getGitMetadata(config) - const tags = repositoryUrl - ? { - ...config.tags, - [GIT_REPOSITORY_URL]: repositoryUrl, - [GIT_COMMIT_SHA]: commitSHA, - } - : config.tags const appliedConfigs = this.appliedConfigs = new Map() @@ -84,13 +83,15 @@ class RemoteConfig { env: config.env, app_version: config.version, extra_services: /** @type {string[]} */ ([]), - tags: Object.entries(tags).map((pair) => pair.join(':')), + tags: getTagsString(config, repositoryUrl, commitSHA), [processTags.REMOTE_CONFIG_FIELD_NAME]: processTags.tagsArray, }, capabilities: DEFAULT_CAPABILITY, // updated by `updateCapabilities()` }, cached_target_files: /** @type {RcCachedTargetFile[]} */ ([]), // updated by `parseConfig()` } + + client = this.state.client } /** @@ -575,4 +576,40 @@ function supportsAckCallback (handler) { return result } +/** + * @param {import('../config/config-base')} config + * @param {string} repositoryUrl + * @param {string} commitSHA + * @returns {string[]} + */ +function getTagsString (config, repositoryUrl, commitSHA) { + const tags = repositoryUrl + ? { + ...config.tags, + [GIT_REPOSITORY_URL]: repositoryUrl, + [GIT_COMMIT_SHA]: commitSHA, + } + : config.tags + return Object.entries(tags).map((pair) => pair.join(':')) +} + +/** + * Regenerates the RC client ID and rebuilds the RC tag list, so subsequent RC polls report the + * clone's identity. No-ops on the tags before the first `RemoteConfig` is constructed. + * + * @param {import('../config/config-base')} config + */ +function refreshIdentity (config) { + clientId = uuid() + 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) + client.client_tracer.tags = getTagsString(config, repositoryUrl, commitSHA) + } +} + module.exports = RemoteConfig diff --git a/packages/dd-trace/src/runtime_metrics/client.js b/packages/dd-trace/src/runtime_metrics/client.js index 1009f15a04d..114a90aafcb 100644 --- a/packages/dd-trace/src/runtime_metrics/client.js +++ b/packages/dd-trace/src/runtime_metrics/client.js @@ -1,21 +1,24 @@ 'use strict' +const { channel } = require('dc-polyfill') const { DogStatsDClient, MetricsAggregationClient } = require('../dogstatsd') const processTags = require('../process-tags') +const identityRefreshChannel = channel('datadog:identity:refresh') + /** - * Builds the aggregating DogStatsD client used to emit DD-proprietary tracer - * metrics (runtime.node.*, datadog.tracer.*). Shared by both runtime-metrics - * paths (DogStatsD and OTLP) so their client construction can't drift apart. + * Builds the DogStatsD tags (with process tags applied) for the runtime-metrics client. + * Shared by `createMetricsClient()` and `subscribeToIdentityRefresh()` so their tag + * composition can't drift apart. * * Process tags are applied here, not via config/generateClientConfig, so they only * reach this bounded set of runtime metrics. Putting them on the global tags would * also tag user-facing custom metrics, inflating their cardinality (and billing). * * @param {import('../config/config-base')} config - Tracer configuration - * @returns {MetricsAggregationClient} + * @returns {{ host: string, port: number, tags: string[], lookup: Function, metricsProxyUrl?: URL }} */ -function createMetricsClient (config) { +function buildClientConfig (config) { const clientConfig = DogStatsDClient.generateClientConfig(config) if (config.DD_EXPERIMENTAL_PROPAGATE_PROCESS_TAGS_ENABLED) { @@ -24,7 +27,38 @@ function createMetricsClient (config) { } } - return new MetricsAggregationClient(new DogStatsDClient(clientConfig)) + return clientConfig +} + +/** + * Builds the aggregating DogStatsD client used to emit DD-proprietary tracer + * metrics (runtime.node.*, datadog.tracer.*). Shared by both runtime-metrics + * paths (DogStatsD and OTLP) so their client construction can't drift apart. + * + * @param {import('../config/config-base')} config - Tracer configuration + * @returns {MetricsAggregationClient} + */ +function createMetricsClient (config) { + return new MetricsAggregationClient(new DogStatsDClient(buildClientConfig(config))) +} + +/** + * Subscribes a runtime-metrics client to the identity-refresh channel so its DogStatsD tags + * (runtime-id, RC client id) reflect `config.tags` after a MicroVM clone resume. + * + * @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, onRefresh) { + const onIdentityRefresh = () => { + client.updateTags(buildClientConfig(config).tags) + onRefresh?.() + } + identityRefreshChannel.subscribe(onIdentityRefresh) + return () => identityRefreshChannel.unsubscribe(onIdentityRefresh) } -module.exports = { createMetricsClient } +module.exports = { createMetricsClient, subscribeToIdentityRefresh } diff --git a/packages/dd-trace/src/runtime_metrics/otlp_runtime_metrics.js b/packages/dd-trace/src/runtime_metrics/otlp_runtime_metrics.js index fbb6e1d7675..9d077790e52 100644 --- a/packages/dd-trace/src/runtime_metrics/otlp_runtime_metrics.js +++ b/packages/dd-trace/src/runtime_metrics/otlp_runtime_metrics.js @@ -5,7 +5,7 @@ const process = require('node:process') const { performance, monitorEventLoopDelay, PerformanceObserver, constants } = require('node:perf_hooks') const { metrics } = require('@opentelemetry/api') const log = require('../log') -const { createMetricsClient } = require('./client') +const { createMetricsClient, subscribeToIdentityRefresh } = require('./client') const METER_NAME = 'datadog.runtime_metrics' @@ -53,6 +53,7 @@ const registeredBatchCallbacks = [] // equivalent; keep a DogStatsD client so OTLP-path customers don't lose them. let client = null let flushInterval = null +let unsubscribeIdentityRefresh = null module.exports = { /** @@ -62,6 +63,7 @@ module.exports = { this.stop() client = createMetricsClient(config) + unsubscribeIdentityRefresh = subscribeToIdentityRefresh(client, config, resetSamplerBaselines) flushInterval = setInterval(() => { client.flush() }, config.DD_RUNTIME_METRICS_FLUSH_INTERVAL ?? 10_000) @@ -221,6 +223,8 @@ module.exports = { clearInterval(flushInterval) flushInterval = null } + unsubscribeIdentityRefresh?.() + unsubscribeIdentityRefresh = null client = null }, @@ -247,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 diff --git a/packages/dd-trace/src/runtime_metrics/runtime_metrics.js b/packages/dd-trace/src/runtime_metrics/runtime_metrics.js index 74e522b047d..8e192289daa 100644 --- a/packages/dd-trace/src/runtime_metrics/runtime_metrics.js +++ b/packages/dd-trace/src/runtime_metrics/runtime_metrics.js @@ -8,7 +8,7 @@ const process = require('process') const { performance, PerformanceObserver, monitorEventLoopDelay } = require('perf_hooks') const log = require('../log') const { NODE_MAJOR, NODE_MINOR } = require('../../../../version') -const { createMetricsClient } = require('./client') +const { createMetricsClient, subscribeToIdentityRefresh } = require('./client') const eventLoopDelayResolution = 4 const EVENT_LOOP_SAMPLE_PER_ITERATION_AVAILABLE = NODE_MAJOR > 26 || @@ -27,6 +27,7 @@ let client = null let lastTime = 0 let lastCpuUsage = null let eventLoopDelayObserver = null +let unsubscribeIdentityRefresh = null // !!!!!!!!!!! // IMPORTANT @@ -48,6 +49,7 @@ module.exports = { const trackGc = config.runtimeMetrics.gc !== false client = createMetricsClient(config) + unsubscribeIdentityRefresh = subscribeToIdentityRefresh(client, config, resetSamplerBaselines) if (trackGc) { startGCObserver() @@ -113,6 +115,8 @@ module.exports = { clearInterval(interval) interval = null + unsubscribeIdentityRefresh?.() + unsubscribeIdentityRefresh = null client = null lastCpuUsage = null @@ -160,6 +164,27 @@ module.exports = { }, } +/** + * Rebases the CPU/event-loop/ELU sampler baselines to now. Without this, a MicroVM clone's first + * capture after resume would report a delta spanning the snapshot pause instead of just the time + * since resume. + * @returns {void} + */ +function resetSamplerBaselines () { + lastTime = performance.now() + lastElu = performance.eventLoopUtilization() + + if (lastCpuUsage !== null) { + lastCpuUsage = process.cpuUsage() + } + + if (eventLoopDelayObserver) { + eventLoopDelayObserver.disable() + eventLoopDelayObserver.reset() + eventLoopDelayObserver.enable() + } +} + function captureCpuUsage () { const currentCpuUsage = process.cpuUsage() const elapsedUsageUser = currentCpuUsage.user - lastCpuUsage.user diff --git a/packages/dd-trace/src/serverless.js b/packages/dd-trace/src/serverless.js index 23e424b4baa..78ac2896c98 100644 --- a/packages/dd-trace/src/serverless.js +++ b/packages/dd-trace/src/serverless.js @@ -84,5 +84,9 @@ module.exports = { getIsAzureFunction, enableGCPPubSubPushSubscription, getIsFlexConsumptionAzureFunction, + IS_AWS_LAMBDA_MICROVM: getEnvironmentVariable('AWS_LAMBDA_MICROVM_IMAGE_ARN') !== undefined, + // true only for a Node that bundles its own OpenSSL, whose CSPRNG keeps the snapshot's DRBG + // state across a MicroVM clone resume + NODE_BUNDLES_OPENSSL: process.config.variables.node_shared_openssl === false, IS_SERVERLESS: isInServerlessEnvironment(), } diff --git a/packages/dd-trace/src/span_stats.js b/packages/dd-trace/src/span_stats.js index 82d9d5e96ef..073a9800e43 100644 --- a/packages/dd-trace/src/span_stats.js +++ b/packages/dd-trace/src/span_stats.js @@ -1,8 +1,10 @@ 'use strict' const os = require('node:os') -const pkg = require('../../../package.json') +const { channel } = require('dc-polyfill') + +const pkg = require('../../../package.json') const { LogCollapsingLowestDenseDDSketch } = require('../../../vendor/dist/@datadog/sketches-js') const { MEASURED, @@ -26,6 +28,12 @@ const { DEFAULT_SERVICE_NAME, } = require('./encode/tags-processors') +const identityRefreshChannel = channel('datadog:identity:refresh') + +// Only one SpanStatsProcessor 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 unsubscribeBucketReset = null + class SpanAggStats { constructor (aggKey) { this.aggKey = aggKey @@ -205,6 +213,14 @@ class SpanStatsProcessor { this.timer = setInterval(this.onInterval.bind(this), intervalMs) this.timer.unref?.() } + + // A clone resume shouldn't export buckets accumulated before the snapshot under its own identity. + unsubscribeBucketReset?.() + const onIdentityRefresh = () => { + this.buckets = new TimeBuckets() + } + identityRefreshChannel.subscribe(onIdentityRefresh) + unsubscribeBucketReset = () => identityRefreshChannel.unsubscribe(onIdentityRefresh) } onInterval () { diff --git a/packages/dd-trace/src/telemetry/session-propagation.js b/packages/dd-trace/src/telemetry/session-propagation.js index 528f36dec49..3c6f7d9f421 100644 --- a/packages/dd-trace/src/telemetry/session-propagation.js +++ b/packages/dd-trace/src/telemetry/session-propagation.js @@ -4,13 +4,14 @@ const dc = /** @type {typeof import('diagnostics_channel')} */ (require('dc-poly const childProcessChannel = dc.tracingChannel('datadog:child_process:execution') let subscribed = false -let runtimeId +let savedConfig function isOptionsObject (value) { return value != null && typeof value === 'object' && !Array.isArray(value) && value } function getEnvWithRuntimeId (env) { + const runtimeId = savedConfig.DD_ROOT_JS_SESSION_ID || savedConfig.tags['runtime-id'] // eslint-disable-next-line eslint-rules/eslint-process-env return { ...(env ?? process.env), DD_ROOT_JS_SESSION_ID: runtimeId } } @@ -43,7 +44,7 @@ function start (config) { if (!config.telemetry.DD_INSTRUMENTATION_TELEMETRY_ENABLED || subscribed) return subscribed = true - runtimeId = config.DD_ROOT_JS_SESSION_ID || config.tags['runtime-id'] + savedConfig = config childProcessChannel.subscribe( /** @type {import('diagnostics_channel').TracingChannelSubscribers} */ ({ start: onChildProcessStart }) diff --git a/packages/dd-trace/src/tracer.js b/packages/dd-trace/src/tracer.js index 8aae2b44170..ad37768adef 100644 --- a/packages/dd-trace/src/tracer.js +++ b/packages/dd-trace/src/tracer.js @@ -12,7 +12,7 @@ const Scope = require('./scope') const { isError } = require('./util') const { setStartupLogConfig } = require('./startup-log') const { DataStreamsCheckpointer, DataStreamsManager, DataStreamsProcessor } = require('./datastreams') -const { IS_SERVERLESS } = require('./serverless') +const { IS_AWS_LAMBDA_MICROVM, IS_SERVERLESS } = require('./serverless') const log = require('./log') // Always-on writer (console.warn), not the channel-gated `log`: these surface regardless of // DD_TRACE_DEBUG. @@ -40,15 +40,11 @@ class DatadogTracer extends Tracer { flushLoadOrderWarnings(logDiagnostic) } - if (!IS_SERVERLESS) { - const storeConfig = require('./tracer_metadata') - // Keep a reference to the handle, to keep the memfd alive in memory. - // It is read by the service discovery feature. - const metadata = storeConfig(config) + if (!IS_SERVERLESS && !IS_AWS_LAMBDA_MICROVM) { + const metadata = require('./tracer_metadata')(config) if (metadata === undefined) { log.warn('Could not store tracer configuration for service discovery') } - this._inmem_cfg = metadata } } diff --git a/packages/dd-trace/src/tracer_metadata.js b/packages/dd-trace/src/tracer_metadata.js index ff2532ff9ba..36dd0b15d4a 100644 --- a/packages/dd-trace/src/tracer_metadata.js +++ b/packages/dd-trace/src/tracer_metadata.js @@ -2,6 +2,12 @@ const tracerVersion = require('../../../version').VERSION +// Keep the memfd alive for service discovery for the process lifetime. +let metadataHandle + +/** + * @param {import('./config/config-base')} config + */ function storeConfig (config) { try { // Load binding first to not import other modules if it throws @@ -29,7 +35,8 @@ function storeConfig (config) { containerId || null ) - return processDiscovery.storeMetadata(metadata) + metadataHandle = processDiscovery.storeMetadata(metadata) + return metadataHandle } catch { // Either libdatadog or process-discovery is unavailable. } diff --git a/packages/dd-trace/test/ci-visibility/exporters/agentless/writer.spec.js b/packages/dd-trace/test/ci-visibility/exporters/agentless/writer.spec.js index 6e9bf292158..e97b904f80e 100644 --- a/packages/dd-trace/test/ci-visibility/exporters/agentless/writer.spec.js +++ b/packages/dd-trace/test/ci-visibility/exporters/agentless/writer.spec.js @@ -1,5 +1,7 @@ 'use strict' +const assert = require('node:assert/strict') + const { describe, it, beforeEach } = require('mocha') const sinon = require('sinon') const proxyquire = require('proxyquire') @@ -11,9 +13,11 @@ let writer let span let request let encoder +let encoderArgs let coverageEncoder let url let log +let tags describe('CI Visibility Writer', () => { beforeEach(() => { @@ -36,7 +40,8 @@ describe('CI Visibility Writer', () => { error: sinon.spy(), } - const AgentlessCiVisibilityEncoder = function () { + const AgentlessCiVisibilityEncoder = function (...args) { + encoderArgs = args return encoder } @@ -56,7 +61,14 @@ describe('CI Visibility Writer', () => { '../../../encode/coverage-ci-visibility': { CoverageCIVisibilityEncoder }, '../../../log': log, }) - writer = new Writer({ url, tags: { 'runtime-id': 'runtime-id' }, coverageUrl: url }) + tags = { 'runtime-id': 'runtime-id' } + writer = new Writer({ url, tags, coverageUrl: url }) + }) + + describe('constructor', () => { + it('should pass the live tags object (not a copied runtime-id) to the encoder', () => { + assert.strictEqual(encoderArgs[1].tags, tags) + }) }) describe('append', () => { diff --git a/packages/dd-trace/test/config/index.spec.js b/packages/dd-trace/test/config/index.spec.js index 1e27935f1ed..7e0bb882415 100644 --- a/packages/dd-trace/test/config/index.spec.js +++ b/packages/dd-trace/test/config/index.spec.js @@ -12,6 +12,7 @@ const sinon = require('sinon') const { it, describe, beforeEach, afterEach } = require('mocha') const context = describe const proxyquire = require('proxyquire') +const { channel } = require('dc-polyfill') require('../setup/core') const exporters = require('../../../../ext/exporters') @@ -5361,4 +5362,93 @@ rules: assert.strictEqual(config.service, 'node') }) }) + + describe('refreshRuntimeId', () => { + const loadConfigModule = (overrides = {}) => { + const uuid = overrides.uuid || require('../../../../vendor/dist/crypto-randomuuid') + const parsers = proxyquire.noPreserveCache()('../../src/config/parsers', {}) + const supportedConfigurations = proxyquire.noPreserveCache()('../../src/config/supported-configurations.json', {}) + const configDefaults = proxyquire.noPreserveCache()('../../src/config/defaults', { + './supported-configurations.json': supportedConfigurations, + '../log': log, + './parsers': parsers, + '../../../../version': { DD_MAJOR }, + }) + const configHelper = proxyquire.noPreserveCache()('../../src/config/helper', { + './supported-configurations.json': supportedConfigurations, + }) + const serverless = proxyquire.noPreserveCache()('../../src/serverless', {}) + return proxyquire.noPreserveCache()('../../src/config', { + './defaults': configDefaults, + '../log': log, + '../telemetry': { updateConfig }, + '../serverless': serverless, + 'node:fs': fs, + './helper': configHelper, + '../pkg': pkg, + '../../../../version': { DD_MAJOR }, + '../../../../vendor/dist/crypto-randomuuid': uuid, + }) + } + + beforeEach(() => { + log = proxyquire('../../src/log', {}) + sinon.spy(log, 'info') + sinon.spy(log, 'warn') + sinon.spy(log, 'error') + }) + + it('should not generate a runtime id until a Config is constructed', () => { + const uuid = sinon.stub().returns('11111111-2222-4333-8444-555555555555') + const configModule = loadConfigModule({ uuid }) + + sinon.assert.notCalled(uuid) + + configModule() + + sinon.assert.calledOnce(uuid) + }) + + it('should update config.tags[runtime-id] to a new UUID', () => { + const configModule = loadConfigModule() + const config = configModule() + const originalId = config.tags['runtime-id'] + + channel('datadog:identity:update').publish(config) + + assert.ok(config.tags['runtime-id']) + assert.strictEqual(typeof config.tags['runtime-id'], 'string') + // runtime-id should have been set + assert.notStrictEqual(config.tags['runtime-id'], originalId) + }) + + it('should call uuid again to regenerate the runtime id', () => { + const uuid = sinon.stub().returns('11111111-2222-4333-8444-555555555555') + const configModule = loadConfigModule({ uuid }) + const config = configModule() + + channel('datadog:identity:update').publish(config) + + // once at module load for the initial runtimeId, once on refresh + sinon.assert.calledTwice(uuid) + // the buffered pool is drained by the publisher, so the refresh must not opt out of it + assert.deepStrictEqual(uuid.secondCall.args, []) + }) + + it('should store new value that differs from original runtimeId', () => { + const uuid = sinon.stub() + uuid.onFirstCall().returns('00000000-0000-4000-8000-000000000001') + uuid.onSecondCall().returns('00000000-0000-4000-8000-000000000002') + const configModule = loadConfigModule({ uuid }) + const config = configModule() + + channel('datadog:identity:update').publish(config) + const firstRefresh = config.tags['runtime-id'] + + channel('datadog:identity:update').publish(config) + const secondRefresh = config.tags['runtime-id'] + + assert.notStrictEqual(firstRefresh, secondRefresh) + }) + }) }) diff --git a/packages/dd-trace/test/crashtracking/crashtracker.spec.js b/packages/dd-trace/test/crashtracking/crashtracker.spec.js index e149b888b6e..f857c3076d3 100644 --- a/packages/dd-trace/test/crashtracking/crashtracker.spec.js +++ b/packages/dd-trace/test/crashtracking/crashtracker.spec.js @@ -14,6 +14,7 @@ describeNotWindows('crashtracker', () => { let crashtracker let binding let config + let identityRefreshChannel let libdatadog let log @@ -36,6 +37,9 @@ describeNotWindows('crashtracker', () => { log = { error: sinon.stub(), } + identityRefreshChannel = { + subscribe: sinon.stub(), + } sinon.stub(binding, 'init') sinon.stub(binding, 'updateConfig') @@ -43,6 +47,7 @@ describeNotWindows('crashtracker', () => { sinon.stub(binding, 'reportUncaughtExceptionMonitor') crashtracker = proxyquire('../../src/crashtracking/crashtracker', { + 'dc-polyfill': { channel: sinon.stub().returns(identityRefreshChannel) }, '../log': log, }) }) @@ -134,6 +139,27 @@ describeNotWindows('crashtracker', () => { }) }) + describe('identity refresh', () => { + it('should reconfigure the binding with refreshed tags when the identity-refresh channel fires', () => { + crashtracker.start(config) + + const refreshedConfig = { ...config, tags: { foo: 'baz' } } + identityRefreshChannel.subscribe.firstCall.args[0](refreshedConfig) + + sinon.assert.called(binding.updateMetadata) + const metadata = binding.updateMetadata.lastCall.args[0] + assert.ok(metadata.tags.includes('foo:baz'), `Expected tags to include foo:baz, got ${inspect(metadata.tags)}`) + }) + + it('should subscribe only after successful initialization', () => { + binding.init.throws(new Error('init failed')) + + crashtracker.start(config) + + sinon.assert.notCalled(identityRefreshChannel.subscribe) + }) + }) + describe('uncaughtExceptionMonitor', () => { it('should register a listener on start', () => { assert.strictEqual(process.listenerCount('uncaughtExceptionMonitor'), 0) diff --git a/packages/dd-trace/test/dogstatsd.spec.js b/packages/dd-trace/test/dogstatsd.spec.js index 2544531201b..24f8932ac72 100644 --- a/packages/dd-trace/test/dogstatsd.spec.js +++ b/packages/dd-trace/test/dogstatsd.spec.js @@ -8,11 +8,14 @@ const os = require('node:os') const { describe, it, beforeEach, afterEach } = require('mocha') const sinon = require('sinon') const proxyquire = require('proxyquire') +const { channel } = require('dc-polyfill') const datadogCore = require('../../datadog-core') require('./setup/core') +const identityRefreshChannel = channel('datadog:identity:refresh') + describe('dogstatsd', () => { let client let DogStatsDClient @@ -724,6 +727,105 @@ describe('dogstatsd', () => { sinon.assert.called(udp4.send) assert.strictEqual(udp4.send.firstCall.args[0].toString(), 'test.avg:10|g|#foo:bar|c:ci-1234\n') }) + + it('should refresh its tags when the identity-refresh channel fires', () => { + const config = { + dogstatsd: { + hostname: '127.0.0.1', + port: 8125, + }, + lookup: dns.lookup, + runtimeMetricsRuntimeId: true, + tags: { 'runtime-id': 'initial-id' }, + } + + client = new CustomMetrics(config) + client.distribution('test.stale', 1) + + config.tags['runtime-id'] = 'refreshed-id' + identityRefreshChannel.publish(config) + + client.gauge('test.avg', 10) + client.flush() + + assert.strictEqual(udp4.send.firstCall.args[0].toString(), 'test.avg:10|g|#runtime-id:refreshed-id\n') + + udp4.send.resetHistory() + config.tags = {} + identityRefreshChannel.publish(config) + + client.gauge('test.avg', 20) + client.flush() + + assert.strictEqual(udp4.send.firstCall.args[0].toString(), 'test.avg:20|g\n') + }) + + it('should preserve buffered metrics when an identity refresh does not change its tags', () => { + const config = { + dogstatsd: { + hostname: '127.0.0.1', + port: 8125, + }, + lookup: dns.lookup, + runtimeMetricsRuntimeId: true, + tags: { 'runtime-id': 'initial-id' }, + } + + client = new CustomMetrics(config) + client.distribution('test.buffered', 1) + + identityRefreshChannel.publish(config) + client.flush() + + assert.strictEqual(udp4.send.firstCall.args[0].toString(), 'test.buffered:1|d|#runtime-id:initial-id\n') + }) + + it('should drop pending aggregated counters/gauges/histograms when an identity refresh changes its tags', + () => { + const config = { + dogstatsd: { + hostname: '127.0.0.1', + port: 8125, + }, + lookup: dns.lookup, + runtimeMetricsRuntimeId: true, + tags: { 'runtime-id': 'initial-id' }, + } + + client = new CustomMetrics(config) + client.increment('test.count', 10) + client.gauge('test.avg', 5) + client.histogram('test.hist', 1) + + config.tags['runtime-id'] = 'refreshed-id' + identityRefreshChannel.publish(config) + + client.flush() + + sinon.assert.notCalled(udp4.send) + }) + + it('should preserve pending aggregated counters/gauges/histograms when an identity refresh does not change ' + + 'its tags', () => { + const config = { + dogstatsd: { + hostname: '127.0.0.1', + port: 8125, + }, + lookup: dns.lookup, + runtimeMetricsRuntimeId: true, + tags: { 'runtime-id': 'initial-id' }, + } + + client = new CustomMetrics(config) + client.increment('test.count', 10) + + identityRefreshChannel.publish(config) + client.flush() + + sinon.assert.called(udp4.send) + assert.strictEqual(udp4.send.firstCall.args[0].toString(), 'test.count:10|c|#runtime-id:initial-id\n') + }) }) describe('MetricsAggregationClient', () => { diff --git a/packages/dd-trace/test/encode/agentless-ci-visibility.spec.js b/packages/dd-trace/test/encode/agentless-ci-visibility.spec.js index 95317e0f3a1..828c42256d8 100644 --- a/packages/dd-trace/test/encode/agentless-ci-visibility.spec.js +++ b/packages/dd-trace/test/encode/agentless-ci-visibility.spec.js @@ -29,16 +29,17 @@ describe('agentless-ci-visibility-encode', () => { let writer let logger let trace + let AgentlessCiVisibilityEncoder beforeEach(() => { logger = { debug: sinon.stub(), } - const { AgentlessCiVisibilityEncoder } = proxyquire('../../src/encode/agentless-ci-visibility', { + AgentlessCiVisibilityEncoder = proxyquire('../../src/encode/agentless-ci-visibility', { '../log': logger, - }) + }).AgentlessCiVisibilityEncoder writer = { flush: sinon.spy() } - encoder = new AgentlessCiVisibilityEncoder(writer, {}) + encoder = new AgentlessCiVisibilityEncoder(writer, { tags: {} }) trace = [{ trace_id: id('1234abcd1234abcd'), @@ -112,6 +113,36 @@ describe('agentless-ci-visibility-encode', () => { assert.strictEqual(spanEvent.content.metrics.negative, -123456712345) }) + it('should encode runtime-id from tags, reflecting a mutation made after construction', () => { + const tags = { 'runtime-id': 'initial-id' } + const localEncoder = new AgentlessCiVisibilityEncoder(writer, { tags }) + + localEncoder.encode(trace) + const firstDecoded = msgpack.decode(localEncoder.makePayload(), { useBigInt64: true }) + assert.strictEqual(firstDecoded.metadata['*']['runtime-id'], 'initial-id') + + tags['runtime-id'] = 'refreshed-id' + + localEncoder.encode(trace) + const secondDecoded = msgpack.decode(localEncoder.makePayload(), { useBigInt64: true }) + assert.strictEqual(secondDecoded.metadata['*']['runtime-id'], 'refreshed-id') + }) + + it('should encode env from tags, reflecting a mutation made after construction', () => { + const tags = { env: 'initial-env' } + const localEncoder = new AgentlessCiVisibilityEncoder(writer, { tags }) + + localEncoder.encode(trace) + const firstDecoded = msgpack.decode(localEncoder.makePayload(), { useBigInt64: true }) + assert.strictEqual(firstDecoded.metadata['*'].env, 'initial-env') + + tags.env = 'refreshed-env' + + localEncoder.encode(trace) + const secondDecoded = msgpack.decode(localEncoder.makePayload(), { useBigInt64: true }) + assert.strictEqual(secondDecoded.metadata['*'].env, 'refreshed-env') + }) + it('should report its count', () => { assert.strictEqual(encoder.count(), 0) diff --git a/packages/dd-trace/test/exporters/agentless/exporter.spec.js b/packages/dd-trace/test/exporters/agentless/exporter.spec.js index 247c1f8ba7c..ac5e9561bfa 100644 --- a/packages/dd-trace/test/exporters/agentless/exporter.spec.js +++ b/packages/dd-trace/test/exporters/agentless/exporter.spec.js @@ -7,11 +7,14 @@ const { inspect } = require('node:util') const { describe, it, beforeEach, afterEach } = require('mocha') const sinon = require('sinon') const proxyquire = require('proxyquire') +const { channel } = require('dc-polyfill') const { assertObjectContains } = require('../../../../../integration-tests/helpers') require('../../setup/core') +const identityRefreshChannel = channel('datadog:identity:refresh') + describe('AgentlessExporter', () => { let Exporter let exporter @@ -26,6 +29,7 @@ describe('AgentlessExporter', () => { append: sinon.stub(), flush: sinon.stub().callsFake((cb) => cb && cb()), setUrl: sinon.stub(), + resetPendingBatch: sinon.stub(), } const Writer = function () { @@ -120,6 +124,54 @@ describe('AgentlessExporter', () => { languageName: 'nodejs', }) }) + + it('should reflect a runtime id updated on config after construction', () => { + const writerOptions = {} + const Writer = function (opts) { + Object.assign(writerOptions, opts) + return writer + } + + Exporter = proxyquire('../../../src/exporters/agentless', { + './writer': Writer, + }) + + const config = { + site: 'datadoghq.com', + env: 'production', + tags: { 'runtime-id': 'test-uuid' }, + } + + exporter = new Exporter(config) + + config.tags['runtime-id'] = 'new-uuid' + + assert.strictEqual(writerOptions.metadata.runtimeID, 'new-uuid') + }) + + it('should reflect an env updated on config after construction', () => { + const writerOptions = {} + const Writer = function (opts) { + Object.assign(writerOptions, opts) + return writer + } + + Exporter = proxyquire('../../../src/exporters/agentless', { + './writer': Writer, + }) + + const config = { + site: 'datadoghq.com', + env: 'production', + tags: { 'runtime-id': 'test-uuid' }, + } + + exporter = new Exporter(config) + + config.env = 'staging' + + assert.strictEqual(writerOptions.metadata.env, 'staging') + }) }) describe('export', () => { @@ -209,6 +261,27 @@ describe('AgentlessExporter', () => { }) }) + describe('identity refresh', () => { + it('drops the pending trace batch when the identity-refresh channel fires', () => { + exporter = new Exporter({}) + + identityRefreshChannel.publish({ tags: {} }) + + sinon.assert.calledOnce(writer.resetPendingBatch) + }) + + it('stops reacting once a newer exporter takes over', () => { + exporter = new Exporter({}) + new Exporter({}) // eslint-disable-line no-new + writer.resetPendingBatch.resetHistory() + + identityRefreshChannel.publish({ tags: {} }) + + // Only one reset, not two - the first exporter's subscription was replaced, not stacked on. + sinon.assert.calledOnce(writer.resetPendingBatch) + }) + }) + describe('setUrl', () => { let log diff --git a/packages/dd-trace/test/exporters/common/writer.spec.js b/packages/dd-trace/test/exporters/common/writer.spec.js index 17a5ce79daf..0e615a9ee70 100644 --- a/packages/dd-trace/test/exporters/common/writer.spec.js +++ b/packages/dd-trace/test/exporters/common/writer.spec.js @@ -73,4 +73,10 @@ describe('common Writer', () => { sinon.assert.notCalled(encoder.reset) sinon.assert.calledOnceWithExactly(writer._sendPayload, payload, 2, done) }) + + it('resetPendingBatch discards the pending encoded batch', () => { + writer.resetPendingBatch() + + sinon.assert.calledOnce(encoder.reset) + }) }) diff --git a/packages/dd-trace/test/id.spec.js b/packages/dd-trace/test/id.spec.js index 0cb78b23cfb..71102ab7b3b 100644 --- a/packages/dd-trace/test/id.spec.js +++ b/packages/dd-trace/test/id.spec.js @@ -5,6 +5,7 @@ const assert = require('node:assert/strict') const { describe, it, beforeEach, afterEach } = require('mocha') const sinon = require('sinon') const proxyquire = require('proxyquire') +const { channel } = require('dc-polyfill') require('./setup/core') @@ -155,4 +156,75 @@ describe('id', () => { assert.strictEqual(spanId.toString(16), '000000000000abcd') assert.strictEqual(spanId.toString(10), '43981') }) + + describe('reseed()', () => { + let freshId + let randomFillSyncStub + + beforeEach(() => { + // Fill with a value that increments per call, so IDs drawn from different + // randomFillSync() fills are distinguishable instead of all looking alike. + let fillByte = 0 + randomFillSyncStub = sinon.stub().callsFake(buf => { + fillByte++ + buf.fill(fillByte) + }) + + freshId = proxyquire('../src/id', { + crypto: { randomFillSync: randomFillSyncStub }, + }) + }) + + it('should generate a different id after reseed', () => { + const before = freshId().toString() + + channel('datadog:identity:update').publish({ tags: {} }) + const after = freshId().toString() + + assert.notStrictEqual(after, before) + }) + + it('should reset the batch cursor to 0', () => { + // Call id() several times to advance the batch counter + freshId() + freshId() + freshId() + randomFillSyncStub.resetHistory() + + channel('datadog:identity:update').publish({ tags: {} }) + // After reseed, batch = 0, so the next call must refill from randomFillSync + freshId() + + sinon.assert.called(randomFillSyncStub) + }) + + it('should force a fresh randomFillSync() call on the very next id() after reseed', () => { + channel('datadog:identity:update').publish({ tags: {} }) + randomFillSyncStub.resetHistory() + + freshId() + + sinon.assert.calledOnce(randomFillSyncStub) + }) + + it('should be safe to call repeatedly', () => { + channel('datadog:identity:update').publish({ tags: {} }) + channel('datadog:identity:update').publish({ tags: {} }) + randomFillSyncStub.resetHistory() + + freshId() + + sinon.assert.calledOnce(randomFillSyncStub) + }) + + it('should reseed when datadog:identity:update is published', () => { + freshId() + randomFillSyncStub.resetHistory() + + channel('datadog:identity:update').publish({ tags: {} }) + freshId() + + sinon.assert.calledOnce(randomFillSyncStub) + }) + }) }) diff --git a/packages/dd-trace/test/opentelemetry/logs.spec.js b/packages/dd-trace/test/opentelemetry/logs.spec.js index 3928372c6c9..552c5bc2d3f 100644 --- a/packages/dd-trace/test/opentelemetry/logs.spec.js +++ b/packages/dd-trace/test/opentelemetry/logs.spec.js @@ -9,6 +9,7 @@ const sinon = require('sinon') const proxyquire = require('proxyquire') const { logs } = require('@opentelemetry/api-logs') const { trace, context } = require('@opentelemetry/api') +const { channel } = require('dc-polyfill') const { timeInputToHrTime } = require('../../../../vendor/dist/@opentelemetry/core') require('../setup/core') @@ -16,6 +17,8 @@ const { protoLogsService } = require('../../src/opentelemetry/otlp/protobuf_load const { getConfigFresh } = require('../helpers/config') const { assertObjectContains } = require('../../../../integration-tests/helpers') +const identityRefreshChannel = channel('datadog:identity:refresh') + /** * @param {object} type protobufjs Type instance for the OTLP service message * @param {object} message Decoded protobufjs Message @@ -799,4 +802,41 @@ describe('OpenTelemetry Logs', () => { assert(telemetryMetrics.manager.namespace().count().inc.calledWith(1)) }) }) + + describe('Identity refresh', () => { + it('exports resource attributes rebuilt after identity refresh', () => { + const validator = mockOtlpExport((decoded) => { + const runtimeId = decoded.resourceLogs[0].resource.attributes.find( + attribute => attribute.key === 'runtime-id' + ) + assert.strictEqual(runtimeId.value.stringValue, 'refreshed-id') + }) + const { config, logs } = setupLogs() + + config.tags['runtime-id'] = 'refreshed-id' + identityRefreshChannel.publish(config) + logs.getLogger('test-logger').emit({ body: 'test' }) + + validator() + }) + + it('drops log records queued before an identity refresh', () => { + const validator = mockOtlpExport((decoded) => { + const records = decoded.resourceLogs[0].scopeLogs[0].logRecords + assert.strictEqual(records.length, 1) + assert.strictEqual(records[0].body.stringValue, 'after-refresh') + }) + // A batch size larger than 1 so the pre-refresh record queues instead of auto-exporting. + const { config, logs, loggerProvider } = setupLogs(true, '10') + + logs.getLogger('test-logger').emit({ body: 'before-refresh' }) + + identityRefreshChannel.publish(config) + + logs.getLogger('test-logger').emit({ body: 'after-refresh' }) + loggerProvider.forceFlush() + + validator() + }) + }) }) diff --git a/packages/dd-trace/test/opentelemetry/metrics.spec.js b/packages/dd-trace/test/opentelemetry/metrics.spec.js index 0ad70fbdc51..5c661735795 100644 --- a/packages/dd-trace/test/opentelemetry/metrics.spec.js +++ b/packages/dd-trace/test/opentelemetry/metrics.spec.js @@ -8,12 +8,15 @@ const { describe, it, beforeEach, afterEach } = require('mocha') const sinon = require('sinon') const proxyquire = require('proxyquire') const { metrics } = require('@opentelemetry/api') +const { channel } = require('dc-polyfill') require('../setup/core') const { protoMetricsService } = require('../../src/opentelemetry/otlp/protobuf_loader').getProtobufTypes() const { getConfigFresh } = require('../helpers/config') const { DEFAULT_MAX_MEASUREMENT_QUEUE_SIZE } = require('../../src/opentelemetry/metrics/constants') +const identityRefreshChannel = channel('datadog:identity:refresh') + /** * @param {object} type protobufjs Type instance for the OTLP service message * @param {Buffer} originalPayload Raw wire bytes captured from the exporter @@ -1237,4 +1240,63 @@ describe('OpenTelemetry Meter Provider', () => { } }) }) + + describe('Identity refresh', () => { + it('exports refreshed resources without resetting the ObservableCounter delta baseline', () => { + const clock = sinon.useFakeTimers() + const exportedMetrics = [] + mockOtlpExport((decoded) => { + const resourceAttributes = decoded.resourceMetrics[0].resource.attributes + const runtimeId = resourceAttributes.find(attribute => attribute.key === 'runtime-id') + const counter = decoded.resourceMetrics[0].scopeMetrics[0].metrics[0] + exportedMetrics.push({ + runtimeId: runtimeId.value.stringValue, + value: counter.sum.dataPoints[0].asInt, + }) + }) + + const { config } = setupMetrics() + const initialRuntimeId = config.tags['runtime-id'] + const meter = metrics.getMeter('app') + let value = 20 + meter.createObservableCounter('obs').addCallback((result) => result.observe(value)) + + clock.tick(100) + + // Refresh happens after the first export already established a baseline of 20. + config.tags['runtime-id'] = 'refreshed-id' + identityRefreshChannel.publish(config) + value = 25 + + clock.tick(100) + + assert.deepStrictEqual(exportedMetrics, [ + { runtimeId: initialRuntimeId, value: 20 }, + { runtimeId: 'refreshed-id', value: 5 }, + ]) + }) + + it('drops sync Counter measurements recorded before an identity refresh', () => { + const clock = sinon.useFakeTimers() + const exportedValues = [] + mockOtlpExport((decoded) => { + const metric = decoded.resourceMetrics[0].scopeMetrics[0].metrics[0] + exportedValues.push(metric.sum.dataPoints[0].asInt) + }) + + const { config } = setupMetrics() + const meter = metrics.getMeter('app') + const counter = meter.createCounter('requests') + + // Recorded before the refresh: dropped, so it never reaches an export under the refreshed + // identity, and it does not contribute to the cumulative total reported afterwards. + counter.add(5) + identityRefreshChannel.publish(config) + + counter.add(7) + clock.tick(100) + + assert.deepStrictEqual(exportedValues, [7]) + }) + }) }) diff --git a/packages/dd-trace/test/opentelemetry/metrics/otlp_span_stats_exporter.spec.js b/packages/dd-trace/test/opentelemetry/metrics/otlp_span_stats_exporter.spec.js index c1fae321295..a831b991272 100644 --- a/packages/dd-trace/test/opentelemetry/metrics/otlp_span_stats_exporter.spec.js +++ b/packages/dd-trace/test/opentelemetry/metrics/otlp_span_stats_exporter.spec.js @@ -4,14 +4,17 @@ const assert = require('node:assert/strict') const http = require('node:http') const { describe, it, beforeEach, afterEach } = require('mocha') const sinon = require('sinon') +const { channel } = require('dc-polyfill') require('../../setup/core') const { OtlpStatsExporter } = require('../../../src/opentelemetry/metrics/otlp_span_stats_exporter') -const { buildResourceAttributes, createOtlpSpanStatsExporter } = require('../../../src/opentelemetry/metrics') +const { createOtlpSpanStatsExporter } = require('../../../src/opentelemetry/metrics') const { SpanBuckets } = require('../../../src/span_stats') const { HTTP_STATUS_CODE } = require('../../../../../ext/tags') +const identityRefreshChannel = channel('datadog:identity:refresh') + const RESOURCE_ATTRS = { 'service.name': 'svc' } const BUCKET_SIZE_NS = 10 * 1e9 @@ -38,29 +41,6 @@ function makeDrained (spans) { return [{ timeNs: 12340000000000, bucket }] } -describe('buildResourceAttributes', () => { - it('includes sdk identity and maps service/env/version to OTel attributes', () => { - const attrs = buildResourceAttributes({}, { service: 'my-svc', env: 'prod', serviceVersion: '1.0.0' }) - - assert.strictEqual(attrs['telemetry.sdk.name'], 'datadog') - assert.strictEqual(attrs['telemetry.sdk.language'], 'nodejs') - assert.strictEqual(typeof attrs['telemetry.sdk.version'], 'string') - assert.strictEqual(attrs['service.name'], 'my-svc') - assert.strictEqual(attrs['deployment.environment.name'], 'prod') - assert.strictEqual(attrs['service.version'], '1.0.0') - }) - - it('includes datadog.runtime_id from tags when otelSemanticsEnabled is false', () => { - const attrs = buildResourceAttributes({ 'runtime-id': 'abc-123' }, { otelSemanticsEnabled: false }) - assert.strictEqual(attrs['datadog.runtime_id'], 'abc-123') - }) - - it('omits dd.* attributes when otelSemanticsEnabled is true', () => { - const attrs = buildResourceAttributes({ 'runtime-id': 'abc-123' }, { otelSemanticsEnabled: true }) - assert.ok(!Object.keys(attrs).some(k => k.startsWith('datadog.'))) - }) -}) - describe('createOtlpSpanStatsExporter', () => { let httpStub @@ -79,6 +59,50 @@ describe('createOtlpSpanStatsExporter', () => { }) assert.ok(exporter instanceof OtlpStatsExporter) }) + + it('exports resource attributes rebuilt after identity refresh', () => { + const config = { + OTEL_EXPORTER_OTLP_METRICS_ENDPOINT: 'http://localhost:4318/v1/metrics', + service: 'svc', + version: '1.0.0', + env: 'prod', + tags: { 'runtime-id': 'initial-id' }, + } + const exporter = createOtlpSpanStatsExporter(config) + + config.tags['runtime-id'] = 'refreshed-id' + identityRefreshChannel.publish(config) + exporter.export(makeDrained([makeSpan()]), BUCKET_SIZE_NS) + + const request = httpStub.firstCall.returnValue + const payload = JSON.parse(request.write.firstCall.args[0].toString()) + const resourceAttributes = Object.fromEntries( + payload.resourceMetrics[0].resource.attributes.map(attribute => [attribute.key, attribute.value.stringValue]) + ) + assert.strictEqual(resourceAttributes['telemetry.sdk.name'], 'datadog') + assert.strictEqual(resourceAttributes['telemetry.sdk.language'], 'nodejs') + assert.strictEqual(typeof resourceAttributes['telemetry.sdk.version'], 'string') + assert.strictEqual(resourceAttributes['service.name'], 'svc') + assert.strictEqual(resourceAttributes['service.version'], '1.0.0') + assert.strictEqual(resourceAttributes['deployment.environment.name'], 'prod') + assert.strictEqual(resourceAttributes['datadog.runtime_id'], 'refreshed-id') + }) + + it('omits Datadog resource attributes with OpenTelemetry semantics', () => { + const exporter = createOtlpSpanStatsExporter({ + OTEL_EXPORTER_OTLP_METRICS_ENDPOINT: 'http://localhost:4318/v1/metrics', + DD_TRACE_OTEL_SEMANTICS_ENABLED: true, + service: 'svc', + tags: { 'runtime-id': 'runtime-id' }, + }) + + exporter.export(makeDrained([makeSpan()]), BUCKET_SIZE_NS) + + const request = httpStub.firstCall.returnValue + const payload = JSON.parse(request.write.firstCall.args[0].toString()) + const resourceAttributes = payload.resourceMetrics[0].resource.attributes + assert.ok(!resourceAttributes.some(attribute => attribute.key.startsWith('datadog.'))) + }) }) describe('OtlpStatsExporter', () => { diff --git a/packages/dd-trace/test/opentelemetry/traces.spec.js b/packages/dd-trace/test/opentelemetry/traces.spec.js index b6078dd8c3f..987c30a23e0 100644 --- a/packages/dd-trace/test/opentelemetry/traces.spec.js +++ b/packages/dd-trace/test/opentelemetry/traces.spec.js @@ -7,12 +7,18 @@ const https = require('node:https') const { describe, it, beforeEach, afterEach } = require('mocha') const sinon = require('sinon') const proxyquire = require('proxyquire') +const { channel } = require('dc-polyfill') require('../setup/core') +const { HTTP_STATUS_CODE } = require('../../../../ext/tags') const { getConfigFresh } = require('../helpers/config') const id = require('../../src/id') +const { createOtlpSpanStatsExporter } = require('../../src/opentelemetry/metrics') const OtlpHttpTraceExporter = require('../../src/opentelemetry/trace/otlp_http_trace_exporter') const { createOtlpTraceExporter } = require('../../src/opentelemetry/trace') +const { SpanBuckets } = require('../../src/span_stats') + +const identityRefreshChannel = channel('datadog:identity:refresh') const OTEL_ENV_KEYS = [ 'OTEL_TRACES_EXPORTER', @@ -59,6 +65,23 @@ describe('OpenTelemetry Traces', () => { } } + function createSpanStatsDrain () { + const span = { + startTime: 12345 * 1e9, + duration: 1000, + error: 0, + name: 'op', + service: 'svc', + resource: 'res', + type: 'web', + meta: { [HTTP_STATUS_CODE]: 200 }, + metrics: {}, + } + const bucket = new SpanBuckets() + bucket.forSpan(span).record(span) + return [{ timeNs: 12340000000000, bucket }] + } + function mockOtlpExport (validator) { let capturedPayload, capturedHeaders let validatorCalled = false @@ -935,4 +958,93 @@ describe('OpenTelemetry Traces', () => { assert.ok(httpsStub.calledOnce, 'https.request should have been called after switching to https') }) }) + + describe('Identity refresh', () => { + it('exports resource attributes rebuilt after identity refresh', () => { + const validator = mockOtlpExport((decoded) => { + const runtimeId = decoded.resourceSpans[0].resource.attributes.find( + attribute => attribute.key === 'runtime-id' + ) + assert.strictEqual(runtimeId.value.stringValue, 'refreshed-id') + }) + const config = getConfigFresh() + const exporter = createOtlpTraceExporter(config) + + config.tags['runtime-id'] = 'refreshed-id' + identityRefreshChannel.publish(config) + exporter.export([createMockSpan()]) + + validator() + }) + + it('updates only the active exporter after identity refresh', () => { + const runtimeIds = [] + const validator = mockOtlpExport((decoded) => { + const runtimeId = decoded.resourceSpans[0].resource.attributes.find( + attribute => attribute.key === 'runtime-id' + ) + runtimeIds.push(runtimeId.value.stringValue) + }) + const firstConfig = getConfigFresh() + firstConfig.tags['runtime-id'] = 'first-initial-id' + const firstExporter = createOtlpTraceExporter(firstConfig) + + const secondConfig = getConfigFresh() + secondConfig.tags['runtime-id'] = 'second-initial-id' + const secondExporter = createOtlpTraceExporter(secondConfig) + + firstConfig.tags['runtime-id'] = 'stale-first-id' + secondConfig.tags['runtime-id'] = 'second-refreshed-id' + identityRefreshChannel.publish(secondConfig) + firstExporter.export([createMockSpan()]) + secondExporter.export([createMockSpan()]) + + assert.deepStrictEqual(runtimeIds, ['first-initial-id', 'second-refreshed-id']) + validator() + }) + + it('refreshes every active signal exporter', () => { + const payloads = new Map() + sinon.stub(http, 'request').callsFake((options, callback) => { + const response = { + statusCode: 200, + on: () => response, + once: () => response, + } + const request = { + write: data => payloads.set(options.path, JSON.parse(data.toString())), + end: () => {}, + on: () => request, + once: () => request, + } + callback(response) + return request + }) + + const traceConfig = getConfigFresh() + traceConfig.tags['runtime-id'] = 'initial-trace-id' + const traceExporter = createOtlpTraceExporter(traceConfig) + const statsConfig = { + OTEL_EXPORTER_OTLP_METRICS_ENDPOINT: 'http://localhost:4318/v1/metrics', + service: 'svc', + tags: { 'runtime-id': 'initial-stats-id' }, + } + const statsExporter = createOtlpSpanStatsExporter(statsConfig) + + traceConfig.tags['runtime-id'] = 'refreshed-trace-id' + statsConfig.tags['runtime-id'] = 'refreshed-stats-id' + identityRefreshChannel.publish(traceConfig) + traceExporter.export([createMockSpan()]) + statsExporter.export(createSpanStatsDrain(), 10 * 1e9) + + const traceRuntimeId = payloads.get('/v1/traces').resourceSpans[0].resource.attributes.find( + attribute => attribute.key === 'runtime-id' + ) + const statsRuntimeId = payloads.get('/v1/metrics').resourceMetrics[0].resource.attributes.find( + attribute => attribute.key === 'datadog.runtime_id' + ) + assert.strictEqual(traceRuntimeId.value.stringValue, 'refreshed-trace-id') + assert.strictEqual(statsRuntimeId.value.stringValue, 'refreshed-stats-id') + }) + }) }) diff --git a/packages/dd-trace/test/plugins/agent.js b/packages/dd-trace/test/plugins/agent.js index 182939f2c33..a1f2b1a8684 100644 --- a/packages/dd-trace/test/plugins/agent.js +++ b/packages/dd-trace/test/plugins/agent.js @@ -68,6 +68,8 @@ const TRACKED_NON_PREFIX_ENV_NAMES = new Set([ 'WEBSITE_SKU', // lambda RITM target path (computed once at module load) 'LAMBDA_TASK_ROOT', + // MicroVM clone-resume identity reseed hook registration + 'AWS_LAMBDA_MICROVM_IMAGE_ARN', // serverless service-name fallbacks (Config singleton) 'WEBSITE_SITE_NAME', // azure metadata payload (cached at first build) diff --git a/packages/dd-trace/test/profiling/profiler.spec.js b/packages/dd-trace/test/profiling/profiler.spec.js index 63312b9bf65..38dddcf3064 100644 --- a/packages/dd-trace/test/profiling/profiler.spec.js +++ b/packages/dd-trace/test/profiling/profiler.spec.js @@ -4,8 +4,8 @@ const assert = require('node:assert/strict') const { inspect } = require('node:util') const { describe, it, beforeEach, afterEach } = require('mocha') +const proxyquire = require('proxyquire').noCallThru() const sinon = require('sinon') -const proxyquire = require('proxyquire') require('../setup/core') @@ -47,6 +47,7 @@ describe('profiler', function () { systemInfoReport: { oomMonitoring: { enabled: false } }, } }, + getProfilingTags: (config) => ({ ...config.tags }), } async function waitForExport () { @@ -545,6 +546,26 @@ describe('profiler', function () { assert.strictEqual(infos.hasMissingSourceMaps, false) }) + + it('uses the current tags when exporting', async () => { + exporterPromise = new Promise(resolve => { + exporter.export = (exportSpec) => { + resolve(exportSpec) + return Promise.resolve() + } + }) + + const startOptions = makeStartOptions({ tags: { 'runtime-id': 'initial-id' } }) + await profiler.start(startOptions) + + startOptions.tags['runtime-id'] = 'refreshed-id' + + clock.tick(interval) + + const { tags } = await exporterPromise + + assert.strictEqual(tags['runtime-id'], 'refreshed-id') + }) }) describe('serverless', function () { diff --git a/packages/dd-trace/test/profiling/profilers/space.spec.js b/packages/dd-trace/test/profiling/profilers/space.spec.js index 2a8574bc6fa..88a290144bb 100644 --- a/packages/dd-trace/test/profiling/profilers/space.spec.js +++ b/packages/dd-trace/test/profiling/profilers/space.spec.js @@ -4,17 +4,19 @@ const assert = require('node:assert/strict') const path = require('node:path') const { pathToFileURL } = require('node:url') -const { describe, it, beforeEach } = require('mocha') -const proxyquire = require('proxyquire') +const { channel } = require('dc-polyfill') +const { describe, it, beforeEach, afterEach } = require('mocha') +const proxyquire = require('proxyquire').noCallThru() const sinon = require('sinon') require('../../setup/core') const { AgentExporter } = require('../../../src/profiling/exporters/agent') const { FileExporter } = require('../../../src/profiling/exporters/file') -// Test adapter: the space profiler reads the canonical DD_PROFILING_* names (allocation and OOM -// monitoring included) straight off the tracer config; only tags and exporters are passed through. -// Map the legacy flat option names onto a config-shaped object. +// Map the legacy flat test options onto the production constructor shape. +const identityRefreshChannel = channel('datadog:identity:refresh') +const activeProfilers = [] + function makeSpace (Cls, { allocationProfilingEnabled = false, heapSamplingInterval = 512 * 1024, @@ -25,24 +27,30 @@ function makeSpace (Cls, { tags = {}, exporters = [], } = {}) { - return new Cls({ + const profiler = new Cls({ DD_PROFILING_HEAP_SAMPLING_INTERVAL: heapSamplingInterval, DD_PROFILING_ALLOCATION_ENABLED: allocationProfilingEnabled, DD_PROFILING_EXPERIMENTAL_OOM_MONITORING_ENABLED: oomMonitoringEnabled, DD_PROFILING_EXPERIMENTAL_OOM_HEAP_LIMIT_EXTENSION_SIZE: heapLimitExtensionSize, DD_PROFILING_EXPERIMENTAL_OOM_MAX_HEAP_EXTENSION_COUNT: maxHeapExtensionCount, DD_PROFILING_EXPERIMENTAL_OOM_EXPORT_STRATEGIES: exportStrategies, - }, { tags, exporters }) + tags, + }, { tags: { ...tags }, exporters }) + activeProfilers.push(profiler) + return profiler } const exporterCliPath = path.join(__dirname, '../../../src/profiling', 'exporter_cli.js') describe('profilers/native/space', () => { let NativeSpaceProfiler + let logger let pprof let profile0 beforeEach(() => { + activeProfilers.length = 0 + logger = { error: sinon.stub() } profile0 = { encodeAsync: sinon.stub().returns(Promise.resolve('encoded')), } @@ -58,9 +66,16 @@ describe('profilers/native/space', () => { NativeSpaceProfiler = proxyquire('../../../src/profiling/profilers/space', { '@datadog/pprof': pprof, + '../../log': logger, }) }) + afterEach(() => { + for (const profiler of activeProfilers) { + profiler.stop() + } + }) + it('should start the internal space profiler', () => { const profiler = makeSpace(NativeSpaceProfiler, { allocationProfilingEnabled: false }) @@ -222,4 +237,49 @@ describe('profilers/native/space', () => { assert.strictEqual(logsEnabled, true) assert.deepStrictEqual(exportCommand, []) }) + + it('should re-register the OOM export command with refreshed tags', () => { + const url = new URL('http://127.0.0.1:8126/') + const tags = { 'runtime-id': 'initial-id' } + const profiler = makeSpace(NativeSpaceProfiler, { + oomMonitoringEnabled: true, + exportStrategies: ['process'], + tags, + exporters: [new AgentExporter({ url, DD_PROFILING_UPLOAD_TIMEOUT: 60_000 })], + }) + + profiler.start() + tags['runtime-id'] = 'refreshed-id' + identityRefreshChannel.publish() + + sinon.assert.calledTwice(pprof.heap.monitorOutOfMemory) + const exportCommand = pprof.heap.monitorOutOfMemory.secondCall.args[3] + assert.deepStrictEqual(exportCommand, [ + process.execPath, + exporterCliPath, + 'http://127.0.0.1:8126/', + 'runtime-id:refreshed-id,snapshot:on_oom', + 'space', + ]) + + profiler.stop() + tags['runtime-id'] = 'another-id' + identityRefreshChannel.publish() + + sinon.assert.calledTwice(pprof.heap.monitorOutOfMemory) + }) + + it('should contain OOM export refresh failures', () => { + const error = new Error('boom') + const profiler = makeSpace(NativeSpaceProfiler, { + oomMonitoringEnabled: true, + exportStrategies: ['process'], + }) + + profiler.start() + pprof.heap.monitorOutOfMemory.onSecondCall().throws(error) + + identityRefreshChannel.publish() + sinon.assert.calledOnceWithExactly(logger.error, error) + }) }) diff --git a/packages/dd-trace/test/proxy.spec.js b/packages/dd-trace/test/proxy.spec.js index 2e35f114a81..307509d6e6f 100644 --- a/packages/dd-trace/test/proxy.spec.js +++ b/packages/dd-trace/test/proxy.spec.js @@ -1,16 +1,22 @@ 'use strict' const assert = require('node:assert/strict') +const { once } = require('node:events') +const http = require('node:http') const { inspect } = require('node:util') const { describe, it, beforeEach, afterEach } = require('mocha') const sinon = require('sinon') const proxyquire = require('proxyquire') + +const { storage } = require('../../datadog-core') const featureRegistry = require('../src/feature-registry') const RemoteConfigCapabilities = require('../src/remote_config/capabilities') require('./setup/core') +const legacyStorage = storage('legacy') + describe('TracerProxy', () => { let ProxyClass let proxy @@ -136,6 +142,7 @@ describe('TracerProxy', () => { log = { error: sinon.spy(), + warn: sinon.spy(), } DatadogTracer = sinon.stub().returns(tracer) @@ -1116,8 +1123,352 @@ describe('TracerProxy', () => { }) }) }) + + describe('MicroVM identity reset', () => { + let channelMock + let diagnosticsChannelMock + let microProxy + let storeConfig + let uuidStub + let buildProxy + + beforeEach(() => { + uuidStub = sinon.stub().returns('00000000-0000-4000-8000-000000000000') + + channelMock = { + subscribe: sinon.stub(), + unsubscribe: sinon.stub(), + publish: sinon.stub(), + } + + diagnosticsChannelMock = { + channel: sinon.stub().returns(channelMock), + } + storeConfig = sinon.stub().returns({}) + + buildProxy = (nodeBundlesOpenssl = false) => new (proxyquire('../src/proxy', { + './tracer': DatadogTracer, + './noop/proxy': NoopProxy, + './config': Config, + './plugin_manager': PluginManager, + './runtime_metrics': runtimeMetrics, + './log': log, + './profiler': profiler, + './tracer_metadata': storeConfig, + './serverless': { + IS_AWS_LAMBDA_MICROVM: true, + IS_SERVERLESS: false, + NODE_BUNDLES_OPENSSL: nodeBundlesOpenssl, + }, + './appsec': appsec, + './appsec/iast': iast, + './telemetry': telemetry, + './remote_config': RemoteConfig, + './aiguard/sdk': AIGuardSdk, + './appsec/sdk': AppsecSdk, + './dogstatsd': dogStatsD, + './noop/dogstatsd': NoopDogStatsDClient, + './flare': flare, + './openfeature': openfeature, + './openfeature/flagging_provider': OpenFeatureProvider, + 'dc-polyfill': diagnosticsChannelMock, + '../../../vendor/dist/crypto-randomuuid': uuidStub, + }))() + + microProxy = buildProxy() + }) + + it('should register the MicroVM hook when env var is set', () => { + microProxy.init() + + sinon.assert.calledWith(diagnosticsChannelMock.channel, 'http.server.request.start') + sinon.assert.calledOnce(channelMock.subscribe) + }) + + it('should keep the MicroVM identity refresh when tracer initialization fails', () => { + const error = new Error('tracer initialization failed') + DatadogTracer.throws(error) + + microProxy.init() + + const subscriber = channelMock.subscribe.firstCall.args[0] + subscriber({ request: { method: 'POST', url: '/aws/lambda-microvms/runtime/v1/run' } }) + + sinon.assert.calledTwice(channelMock.publish) + sinon.assert.alwaysCalledWithExactly(channelMock.publish, config) + sinon.assert.notCalled(storeConfig) + sinon.assert.calledOnceWithExactly(log.error, 'Error initializing tracer', error) + }) + + it('should not store tracer metadata when tracing is disabled', () => { + config.DD_TRACE_ENABLED = false + microProxy.init() + + const subscriber = channelMock.subscribe.firstCall.args[0] + subscriber({ request: { method: 'POST', url: '/aws/lambda-microvms/runtime/v1/run' } }) + + sinon.assert.notCalled(storeConfig) + sinon.assert.calledTwice(channelMock.publish) + }) + + it('should publish datadog:identity:update with the tracer config on POST .../run', () => { + microProxy.init() + + const subscriber = channelMock.subscribe.firstCall.args[0] + sinon.assert.notCalled(channelMock.publish) + + subscriber({ request: { method: 'POST', url: '/aws/lambda-microvms/runtime/v1/run' } }) + + sinon.assert.calledWith(diagnosticsChannelMock.channel, 'datadog:identity:update') + sinon.assert.calledWith(diagnosticsChannelMock.channel, 'datadog:identity:refresh') + sinon.assert.calledTwice(channelMock.publish) + sinon.assert.alwaysCalledWithExactly(channelMock.publish, config) + }) + + it('should update identity, store metadata, and then refresh consumers', () => { + // Core identity producers (id/config/remote_config) self-subscribe to identity:update and + // must finish reseeding before identity:refresh notifies downstream cache-holders + // (dogstatsd, otel metrics, debugger); otherwise those subsystems would refresh from the + // pre-reseed identity. + storeConfig.returns(undefined) + microProxy.init() + + const subscriber = channelMock.subscribe.firstCall.args[0] + subscriber({ request: { method: 'POST', url: '/aws/lambda-microvms/runtime/v1/run' } }) + + const channelNames = diagnosticsChannelMock.channel.getCalls().map(call => call.args[0]) + const updateIndex = channelNames.indexOf('datadog:identity:update') + const refreshIndex = channelNames.indexOf('datadog:identity:refresh') + + assert.notStrictEqual(updateIndex, -1) + assert.notStrictEqual(refreshIndex, -1) + + const updateCall = diagnosticsChannelMock.channel.getCall(updateIndex) + const refreshCall = diagnosticsChannelMock.channel.getCall(refreshIndex) + assert.ok(updateCall.callId < storeConfig.firstCall.callId) + assert.ok(storeConfig.firstCall.callId < refreshCall.callId) + sinon.assert.calledOnceWithExactly(log.warn, 'Could not store tracer configuration for service discovery') + }) + + it('should NOT fire refreshIdentity on GET /aws/lambda-microvms/runtime/v1/run', () => { + microProxy.init() + + const subscriber = channelMock.subscribe.firstCall.args[0] + subscriber({ request: { method: 'GET', url: '/aws/lambda-microvms/runtime/v1/run' } }) + + sinon.assert.notCalled(channelMock.publish) + }) + + it('should NOT fire refreshIdentity on POST /other', () => { + microProxy.init() + + const subscriber = channelMock.subscribe.firstCall.args[0] + subscriber({ request: { method: 'POST', url: '/other' } }) + + sinon.assert.notCalled(channelMock.publish) + }) + + it('should warn at registration when Node bundles its own OpenSSL', () => { + buildProxy(true).init() + + sinon.assert.calledOnce(log.warn) + assert.match(log.warn.firstCall.args[0], /bundles its own OpenSSL/) + }) + + it('should not warn when Node links a shared OpenSSL', () => { + microProxy.init() + + sinon.assert.notCalled(log.warn) + }) + + it('should drain a full UUID batch before publishing datadog:identity:update', () => { + microProxy.init() + + const subscriber = channelMock.subscribe.firstCall.args[0] + sinon.assert.notCalled(uuidStub) + + subscriber({ request: { method: 'POST', url: '/aws/lambda-microvms/runtime/v1/run' } }) + + // a full batch, so the pool's cursor wraps and refills wherever the snapshot froze it + sinon.assert.callCount(uuidStub, 128) + + const updateIndex = diagnosticsChannelMock.channel.getCalls() + .findIndex(call => call.args[0] === 'datadog:identity:update') + const updateCall = diagnosticsChannelMock.channel.getCall(updateIndex) + + assert.ok(uuidStub.lastCall.callId < updateCall.callId) + }) + + it('should not drain the UUID pool when the request is not the run hook', () => { + microProxy.init() + + const subscriber = channelMock.subscribe.firstCall.args[0] + subscriber({ request: { method: 'GET', url: '/aws/lambda-microvms/runtime/v1/run' } }) + subscriber({ request: { method: 'POST', url: '/other' } }) + + sinon.assert.notCalled(uuidStub) + }) + + it('should unsubscribe HTTP channel after first fire', () => { + microProxy.init() + + const subscriber = channelMock.subscribe.firstCall.args[0] + subscriber({ request: { method: 'POST', url: '/aws/lambda-microvms/runtime/v1/run' } }) + + sinon.assert.calledOnceWithExactly(channelMock.unsubscribe, subscriber) + }) + }) + + describe('MicroVM identity refresh (real modules)', () => { + let RealConfig + let RealRemoteConfig + let udp4Send + let MicroVmProxy + let microProxy + let capturedConfig + let storedRuntimeId + let storeConfig + let server + + beforeEach(async () => { + // Real (not proxied) config/remote_config/dogstatsd modules, so this test proves the + // actual production wiring — not that mocks were called correctly. + RealConfig = proxyquire.noPreserveCache()('../src/config', {}) + RealRemoteConfig = proxyquire.noPreserveCache()('../src/remote_config', {}) + + udp4Send = sinon.spy() + const udp4 = { send: udp4Send, on: sinon.stub(), unref: sinon.stub() } + udp4.on.returns(udp4) + udp4.unref.returns(udp4) + const dgram = { createSocket: sinon.stub().returns(udp4) } + const dns = { lookup: sinon.stub().callsFake((hostname, callback) => callback(null, hostname, 4)) } + const RealCustomMetrics = proxyquire.noPreserveCache()('../src/dogstatsd', { dgram }).CustomMetrics + + storeConfig = sinon.stub().callsFake(metadataConfig => { + storedRuntimeId = metadataConfig.tags['runtime-id'] + return {} + }) + + capturedConfig = null + const CapturingConfig = (...args) => { + capturedConfig = RealConfig(...args) + capturedConfig.lookup = dns.lookup + capturedConfig.runtimeMetricsRuntimeId = true + // Force the UDP send path so this test can observe the outgoing packet directly, + // instead of going through the HTTP-proxy-to-agent path config.url otherwise selects. + capturedConfig.url = undefined + return capturedConfig + } + + MicroVmProxy = proxyquire('../src/proxy', { + './tracer': DatadogTracer, + './noop/proxy': NoopProxy, + './config': CapturingConfig, + './plugin_manager': PluginManager, + './runtime_metrics': runtimeMetrics, + './log': log, + './profiler': profiler, + './tracer_metadata': storeConfig, + './serverless': { + IS_AWS_LAMBDA_MICROVM: true, + IS_SERVERLESS: false, + }, + './appsec': appsec, + './appsec/iast': iast, + './telemetry': telemetry, + './remote_config': RemoteConfig, + './aiguard/sdk': AIGuardSdk, + './appsec/sdk': AppsecSdk, + './dogstatsd': { CustomMetrics: RealCustomMetrics }, + './noop/dogstatsd': NoopDogStatsDClient, + './flare': flare, + './openfeature': openfeature, + './openfeature/flagging_provider': OpenFeatureProvider, + // dc-polyfill intentionally NOT mocked — this test exercises the real shared channel. + }) + + microProxy = new MicroVmProxy() + + server = http.createServer((request, response) => { + assert.strictEqual(request.method, 'POST') + assert.strictEqual(request.url, '/aws/lambda-microvms/runtime/v1/run') + response.end() + }) + server.listen(0) + await once(server, 'listening') + }) + + afterEach(async () => { + const closed = once(server, 'close') + server.close() + await closed + }) + + it('refreshes config, remote config, and dogstatsd tags together on a real /run request', async () => { + microProxy.init() + + // Constructed independently from the same real remote_config module proxy.js uses + // internally — refreshClientId() mutates module-scoped state shared by every instance. + const rc = new RealRemoteConfig(capturedConfig) + const originalRuntimeId = capturedConfig.tags['runtime-id'] + const originalClientId = rc.state.client.id + + const customMetrics = microProxy.dogstatsd + + await triggerMicroVmRun(server) + + assert.notStrictEqual(capturedConfig.tags['runtime-id'], originalRuntimeId) + assert.notStrictEqual(rc.state.client.id, originalClientId) + + customMetrics.gauge('test.metric', 1) + customMetrics.flush() + + sinon.assert.called(udp4Send) + const payload = udp4Send.firstCall.args[0].toString() + assert.ok( + payload.includes(`runtime-id:${capturedConfig.tags['runtime-id']}`), + `expected refreshed runtime-id in payload, got: ${payload}` + ) + }) + + it('stores process metadata once with the refreshed /run identity', async () => { + microProxy.init() + + const originalRuntimeId = capturedConfig.tags['runtime-id'] + sinon.assert.notCalled(storeConfig) + + await triggerMicroVmRun(server) + await triggerMicroVmRun(server) + + sinon.assert.calledOnceWithExactly(storeConfig, capturedConfig) + assert.notStrictEqual(storedRuntimeId, originalRuntimeId) + assert.strictEqual(storedRuntimeId, capturedConfig.tags['runtime-id']) + }) + }) }) +/** + * @param {import('node:http').Server} server + */ +async function triggerMicroVmRun (server) { + await legacyStorage.run({ noop: true }, async () => { + const { address, port } = server.address() + const request = http.request({ + hostname: address === '::' ? '::1' : address, + method: 'POST', + path: '/aws/lambda-microvms/runtime/v1/run', + port, + }) + const responsePromise = once(request, 'response') + request.end() + + const [response] = await responsePromise + const end = once(response, 'end') + response.resume() + await end + }) +} + // Helper function to create APM_TRACING batch transaction objects function createApmTracingTransaction (configId, libConfig, action = 'apply') { const item = { diff --git a/packages/dd-trace/test/remote_config/index.spec.js b/packages/dd-trace/test/remote_config/index.spec.js index 1d43030cdd3..03b9ab34371 100644 --- a/packages/dd-trace/test/remote_config/index.spec.js +++ b/packages/dd-trace/test/remote_config/index.spec.js @@ -6,6 +6,7 @@ const { inspect } = require('node:util') const { describe, it, beforeEach } = require('mocha') const sinon = require('sinon') const proxyquire = require('proxyquire') +const { channel } = require('dc-polyfill') require('../setup/core') const Capabilities = require('../../src/remote_config/capabilities') @@ -820,6 +821,177 @@ describe('RemoteConfig', () => { assert.strictEqual(rc.appliedConfigs.size, 0) }) }) + + describe('identity state refresh', () => { + it('should replace state.client.id when identity is updated', () => { + const originalId = rc.state.client.id + + uuid.returns('refreshed-client-id') + channel('datadog:identity:update').publish(config) + + assert.strictEqual(rc.state.client.id, 'refreshed-client-id') + assert.notStrictEqual(rc.state.client.id, originalId) + }) + + it('should replace client_tracer.runtime_id when identity is updated', () => { + assert.strictEqual(rc.state.client.client_tracer.runtime_id, 'runtimeId') + + config.tags['runtime-id'] = 'refreshed-runtime-id' + channel('datadog:identity:update').publish(config) + + assert.strictEqual(rc.state.client.client_tracer.runtime_id, 'refreshed-runtime-id') + + config.tags['runtime-id'] = 'runtimeId' // restore for other tests + }) + + it('should include refreshed runtime_id and id in the JSON payload', () => { + uuid.returns('refreshed-client-id') + config.tags['runtime-id'] = 'live-runtime-id' + channel('datadog:identity:update').publish(config) + const payload = JSON.parse(rc.getPayload()) + + assert.strictEqual(payload.client.client_tracer.runtime_id, 'live-runtime-id') + assert.strictEqual(payload.client.id, 'refreshed-client-id') + + config.tags['runtime-id'] = 'runtimeId' + }) + + it('should cache client_tracer.tags and only refresh it on datadog:identity:update', () => { + const originalTags = rc.state.client.client_tracer.tags + + config.tags['new-tag'] = 'new-value' + + // unlike runtime_id, tags is a cached string, so a direct config mutation isn't picked up + assert.strictEqual(rc.state.client.client_tracer.tags, originalTags) + + channel('datadog:identity:update').publish(config) + + assert.notStrictEqual(rc.state.client.client_tracer.tags, originalTags) + assert.ok(rc.state.client.client_tracer.tags.includes('new-tag:new-value')) + + delete config.tags['new-tag'] + }) + }) + + describe('refreshClientId', () => { + let refreshIdentity + let uuidStub + let RemoteConfigWithId + + beforeEach(() => { + uuidStub = sinon.stub() + // first call is the module-load-time `let clientId = uuid()`, second is the refresh + uuidStub.onFirstCall().returns('1234-5678') + uuidStub.onSecondCall().returns('new-client-id-uuid') + + RemoteConfigWithId = proxyquire('../../src/remote_config', { + 'dc-polyfill': { + channel: sinon.stub().returns({ + subscribe: (listener) => { refreshIdentity = listener }, + }), + }, + '../../../../vendor/dist/crypto-randomuuid': uuidStub, + './scheduler': Scheduler, + '../../../../package.json': { version: '3.0.0' }, + '../exporters/common/request': request, + '../log': log, + '../tagger': tagger, + '../git_metadata': getGitMetadata, + '../service-naming/extra-services': { + getExtraServices: () => extraServices, + }, + }) + }) + + it('should update state.client.id on the existing instance after refresh', () => { + const rcInstance = new RemoteConfigWithId(config) + assert.strictEqual(rcInstance.state.client.id, '1234-5678') + + refreshIdentity(config) + + assert.strictEqual(rcInstance.state.client.id, 'new-client-id-uuid') + }) + + it('should rebuild client_tracer.tags to reflect the refreshed _dd.rc.client_id', () => { + const rcConfig = { + url: new URL('http://127.0.0.1:1337'), + tags: { 'runtime-id': 'runtimeId', '_dd.rc.client_id': 'old-client-id' }, + service: 'serviceName', + env: 'serviceEnv', + version: 'appVersion', + remoteConfig: { pollInterval: 5 }, + } + const rcInstance = new RemoteConfigWithId(rcConfig) + assert.deepStrictEqual(rcInstance.state.client.client_tracer.tags, [ + 'runtime-id:runtimeId', + '_dd.rc.client_id:old-client-id', + ]) + + refreshIdentity(rcConfig) + + const refreshedTags = rcInstance.state.client.client_tracer.tags + assert.deepStrictEqual(refreshedTags, [ + 'runtime-id:runtimeId', + '_dd.rc.client_id:new-client-id-uuid', + ]) + }) + + it('should set clientId to the value returned by uuid', () => { + const rcConfig = { + url: new URL('http://127.0.0.1:1337'), + tags: { 'runtime-id': 'runtimeId', '_dd.rc.client_id': 'old' }, + service: 'serviceName', + env: 'serviceEnv', + version: 'appVersion', + remoteConfig: { pollInterval: 5 }, + } + new RemoteConfigWithId(rcConfig) // eslint-disable-line no-new + refreshIdentity(rcConfig) + + assert.strictEqual(rcConfig.tags['_dd.rc.client_id'], 'new-client-id-uuid') + }) + + it('should update config.tags[_dd.rc.client_id] when it exists', () => { + const rcConfig = { + url: new URL('http://127.0.0.1:1337'), + tags: { + 'runtime-id': 'runtimeId', + '_dd.rc.client_id': 'old-client-id', + }, + service: 'serviceName', + env: 'serviceEnv', + version: 'appVersion', + remoteConfig: { pollInterval: 5 }, + } + new RemoteConfigWithId(rcConfig) // eslint-disable-line no-new + refreshIdentity(rcConfig) + + assert.strictEqual(rcConfig.tags['_dd.rc.client_id'], 'new-client-id-uuid') + }) + + it('should not update config.tags[_dd.rc.client_id] when tag is absent', () => { + const rcConfig = { + url: new URL('http://127.0.0.1:1337'), + tags: { 'runtime-id': 'runtimeId' }, + service: 'serviceName', + env: 'serviceEnv', + version: 'appVersion', + remoteConfig: { pollInterval: 5 }, + } + refreshIdentity(rcConfig) + + assert.strictEqual(rcConfig.tags['_dd.rc.client_id'], undefined) + }) + + it('should call uuid again to generate the new ID', () => { + refreshIdentity(config) + + // once at module load for the initial clientId, once on refresh + sinon.assert.calledTwice(uuidStub) + // the buffered pool is drained by the publisher, so the refresh must not opt out of it + assert.deepStrictEqual(uuidStub.secondCall.args, []) + }) + }) }) function toBase64 (data) { diff --git a/packages/dd-trace/test/runtime_metrics.spec.js b/packages/dd-trace/test/runtime_metrics.spec.js index 94309d1d4e5..3b0cfe25382 100644 --- a/packages/dd-trace/test/runtime_metrics.spec.js +++ b/packages/dd-trace/test/runtime_metrics.spec.js @@ -10,6 +10,7 @@ const { describe, it, beforeEach, afterEach } = require('mocha') const proxyquire = require('proxyquire') const sinon = require('sinon') const { metrics } = require('@opentelemetry/api') +const { channel } = require('dc-polyfill') require('./setup/core') const { NODE_MAJOR, NODE_MINOR } = require('../../../version') @@ -30,6 +31,8 @@ const PeriodicMetricReader = require('../src/opentelemetry/metrics/periodic_metr const OtlpTransformer = require('../src/opentelemetry/metrics/otlp_transformer') const otlpRuntimeMetrics = require('../src/runtime_metrics/otlp_runtime_metrics') +const identityRefreshChannel = channel('datadog:identity:refresh') + function createGarbage (count = 50) { let last = {} const obj = last @@ -177,6 +180,7 @@ NATIVE_METRICS_VARIANTS.forEach((nativeMetrics) => { increment: wrapSpy(client, client.increment), histogram: wrapSpy(client, client.histogram), flush: client.flush.bind(client), + updateTags: client.updateTags, } }) @@ -187,6 +191,7 @@ NATIVE_METRICS_VARIANTS.forEach((nativeMetrics) => { increment: sinon.spy(), histogram: sinon.spy(), flush: sinon.spy(), + updateTags: sinon.spy(), } const proxiedObject = { @@ -299,6 +304,56 @@ NATIVE_METRICS_VARIANTS.forEach((nativeMetrics) => { assert.ok(!tags.some(tag => tag.startsWith('entrypoint.')), 'expected no entrypoint tags') }) + it('should refresh the DogStatsD client tags when the identity-refresh channel fires', () => { + config.tags['runtime-id'] = 'initial-id' + config.runtimeMetricsRuntimeId = true + runtimeMetrics.stop() + runtimeMetrics.start(config) + client.updateTags.resetHistory() + + // Simulates `proxy.js#refreshIdentity` mutating `config.tags['runtime-id']` in place + // and then publishing to the shared identity-refresh channel (MicroVM clone resume). + config.tags['runtime-id'] = 'refreshed-id' + identityRefreshChannel.publish(config) + + sinon.assert.calledOnce(client.updateTags) + const tags = client.updateTags.lastCall.args[0] + assert.ok(tags.includes('runtime-id:refreshed-id'), `expected tags to include refreshed-id: ${tags}`) + }) + + it('should stop reacting to identity refresh after stop', () => { + runtimeMetrics.stop() + client.updateTags.resetHistory() + + identityRefreshChannel.publish(config) + + sinon.assert.notCalled(client.updateTags) + }) + + it('should reset the event-loop-delay observer baseline on identity refresh', function () { + if (nativeMetrics) this.skip() + + const fakeHistogram = makeFakeEventLoopDelayHistogram({ count: 5 }) + const localRuntimeMetrics = proxyquire('../src/runtime_metrics/runtime_metrics', { + perf_hooks: { ...require('perf_hooks'), monitorEventLoopDelay: () => fakeHistogram }, + '@datadog/native-metrics': { + start () { + throw new Error('Native metrics are not supported in this environment') + }, + }, + './client': proxyquire('../src/runtime_metrics/client', { + '../dogstatsd': { DogStatsDClient: Client }, + }), + }) + + localRuntimeMetrics.start(config) + identityRefreshChannel.publish(config) + + assert.strictEqual(fakeHistogram.getResetCallCount(), 1) + + localRuntimeMetrics.stop() + }) + it('should start collecting runtimeMetrics every 10 seconds', async () => { runtimeMetrics.stop() runtimeMetrics.start(config) @@ -1049,6 +1104,7 @@ FakePerformanceObserverForOtlp.instances = [] * batchCallbacks: Array<{ cb: Function, observables: object[] }>, * fireBatchCallbacks: () => Map>, * fakeMetricsClient: object, + * identityRefreshCalls: Array<{ client: object, config: object, onRefresh: Function, unsubscribe: Function }>, * }} */ function loadOtlpRuntimeMetricsTestModule (overrides = {}) { @@ -1057,6 +1113,7 @@ function loadOtlpRuntimeMetricsTestModule (overrides = {}) { const records = {} const batchCallbacks = [] const statsdCalls = [] + const identityRefreshCalls = [] FakePerformanceObserverForOtlp.instances = [] function makeFactory (type) { @@ -1124,6 +1181,11 @@ function loadOtlpRuntimeMetricsTestModule (overrides = {}) { }, './client': { createMetricsClient: () => fakeMetricsClient, + subscribeToIdentityRefresh: (client, config, onRefresh) => { + const unsubscribe = sinon.spy() + identityRefreshCalls.push({ client, config, onRefresh, unsubscribe }) + return unsubscribe + }, }, }) @@ -1151,6 +1213,7 @@ function loadOtlpRuntimeMetricsTestModule (overrides = {}) { fireBatchCallbacks, fakeMetricsClient, statsdCalls, + identityRefreshCalls, } } @@ -1426,6 +1489,7 @@ describe('otlp_runtime_metrics', () => { '../log': { debug () {}, error: errorLog }, './client': { createMetricsClient: () => ({ flush () {} }), + subscribeToIdentityRefresh: () => () => {}, }, }) const dispatcher = proxyquire.noCallThru()('../src/runtime_metrics', { @@ -1481,6 +1545,37 @@ describe('otlp_runtime_metrics', () => { assert.strictEqual(Object.keys(createdInstruments).length, Object.keys(SPEC).length, 'should register every metric again after stop') }) + + it('subscribes the DogStatsD client to identity refresh on start and unsubscribes on stop', () => { + const ctx = loadOtlpRuntimeMetricsTestModule() + const config = { runtimeMetrics: { eventLoop: true } } + + ctx.otlpMetrics.start(config) + + assert.strictEqual(ctx.identityRefreshCalls.length, 1) + assert.strictEqual(ctx.identityRefreshCalls[0].client, ctx.fakeMetricsClient) + assert.strictEqual(ctx.identityRefreshCalls[0].config, config) + sinon.assert.notCalled(ctx.identityRefreshCalls[0].unsubscribe) + + ctx.otlpMetrics.stop() + + sinon.assert.calledOnce(ctx.identityRefreshCalls[0].unsubscribe) + }) + + it('resets the event-loop-delay histogram baseline on identity refresh', () => { + const fakeH = makeFakeEventLoopDelayHistogram({ count: 5 }) + const ctx = loadOtlpRuntimeMetricsTestModule({ + monitorEventLoopDelay: () => fakeH, + }) + ctx.otlpMetrics.start({ runtimeMetrics: { eventLoop: true } }) + + assert.strictEqual(ctx.identityRefreshCalls.length, 1) + ctx.identityRefreshCalls[0].onRefresh() + + assert.strictEqual(fakeH.getResetCallCount(), 1) + + ctx.otlpMetrics.stop() + }) }) // End-to-end through a real MeterProvider + PeriodicMetricReader + OtlpTransformer, diff --git a/packages/dd-trace/test/span_stats.spec.js b/packages/dd-trace/test/span_stats.spec.js index 515d5e564fc..107b1fc5254 100644 --- a/packages/dd-trace/test/span_stats.spec.js +++ b/packages/dd-trace/test/span_stats.spec.js @@ -6,8 +6,12 @@ const { hostname } = require('os') const { describe, it } = require('mocha') const sinon = require('sinon') const proxyquire = require('proxyquire') +const { channel } = require('dc-polyfill') require('./setup/core') + +const identityRefreshChannel = channel('datadog:identity:refresh') + const { LogCollapsingLowestDenseDDSketch } = require('../../../vendor/dist/@datadog/sketches-js') const { version } = require('../src/pkg') const pkg = require('../../../package.json') @@ -586,4 +590,35 @@ describe('SpanStatsProcessor', () => { p.onSpanFinished(topLevelSpan) assert.strictEqual(p.buckets.size, 1) }) + + it('should clear pending buckets when the identity-refresh channel fires', () => { + const p = new SpanStatsProcessor(config) + clearTimeout(p.timer) + + p.onSpanFinished(topLevelSpan) + assert.strictEqual(p.buckets.size, 1) + + const previousBuckets = p.buckets + identityRefreshChannel.publish(config) + + assert.notStrictEqual(p.buckets, previousBuckets) + assert.strictEqual(p.buckets.size, 0) + }) + + it('should stop reacting to identity refresh once a newer instance takes over', () => { + const first = new SpanStatsProcessor(config) + clearTimeout(first.timer) + const firstBuckets = first.buckets + + const second = new SpanStatsProcessor(config) + clearTimeout(second.timer) + const secondBuckets = second.buckets + + identityRefreshChannel.publish(config) + + // Only the second (newest) instance should react - the first's subscription was replaced, + // not stacked on top of. + assert.strictEqual(first.buckets, firstBuckets) + assert.notStrictEqual(second.buckets, secondBuckets) + }) }) diff --git a/packages/dd-trace/test/telemetry/session-propagation.spec.js b/packages/dd-trace/test/telemetry/session-propagation.spec.js index 7fe44a4c8c0..5dbd4cff21f 100644 --- a/packages/dd-trace/test/telemetry/session-propagation.spec.js +++ b/packages/dd-trace/test/telemetry/session-propagation.spec.js @@ -228,6 +228,21 @@ describe('session-propagation', () => { ]) }) + it('reflects a runtime id updated on config after start (e.g. MicroVM identity refresh)', () => { + const config = createConfig() + sessionPropagation.start(config) + + config.tags['runtime-id'] = 'refreshed-id' + + const context = publishStart({ callArgs: ['node', ['test.js'], {}], shell: false }) + + assert.deepStrictEqual(context.callArgs, [ + 'node', + ['test.js'], + { env: createExpectedEnv({ DD_ROOT_JS_SESSION_ID: 'refreshed-id' }) }, + ]) + }) + it('ignores execution contexts without call arguments', () => { sessionPropagation.start(createConfig()) diff --git a/packages/dd-trace/test/tracer.spec.js b/packages/dd-trace/test/tracer.spec.js index ee9c4d5ec05..6b5293dbaab 100644 --- a/packages/dd-trace/test/tracer.spec.js +++ b/packages/dd-trace/test/tracer.spec.js @@ -447,4 +447,22 @@ describe('Tracer', () => { 'expected service discovery warning to be emitted at log level warn') }) }) + + describe('MicroVM service discovery metadata', () => { + it('should defer storing metadata while a MicroVM image is being built', () => { + const storeConfig = sinon.stub() + const PatchedTracer = proxyquire('../src/tracer', { + './serverless': { + IS_AWS_LAMBDA_MICROVM: true, + IS_SERVERLESS: false, + }, + './tracer_metadata': storeConfig, + }) + + // eslint-disable-next-line no-new + new PatchedTracer(getConfig({ service: 'service' })) + + sinon.assert.notCalled(storeConfig) + }) + }) })