From 77e55eb616baee5703082f5dccddf8e30b536410 Mon Sep 17 00:00:00 2001 From: Leo Romanovsky Date: Sun, 9 Aug 2026 13:45:41 -0400 Subject: [PATCH 1/5] feat(openfeature): add exposure route fallback --- .../dd-trace/src/openfeature/writers/base.js | 137 ++++++++++++++++-- .../src/openfeature/writers/exposures.js | 56 +++++-- .../openfeature/writers/exposures.spec.js | 120 +++++++++++++++ 3 files changed, 286 insertions(+), 27 deletions(-) diff --git a/packages/dd-trace/src/openfeature/writers/base.js b/packages/dd-trace/src/openfeature/writers/base.js index cbb60d29d40..c0007c2963b 100644 --- a/packages/dd-trace/src/openfeature/writers/base.js +++ b/packages/dd-trace/src/openfeature/writers/base.js @@ -11,14 +11,40 @@ const log = require('../../log') * @property {number} [timeout] - Request timeout in milliseconds * @property {object} config - Tracer configuration object * @property {string} endpoint - API endpoint path - * @property {URL} [agentUrl] - Base URL for the agent + * @property {URL} [agentUrl] - Initial delivery URL * @property {number} [payloadSizeLimit] - Maximum payload size in bytes * @property {number} [eventSizeLimit] - Maximum individual event size in bytes * @property {object} [headers] - Additional HTTP headers */ /** - * BaseFFEWriter is the base class for sending Feature Flagging & Exposure Events payloads to the Datadog Agent. + * @typedef {object} WriterRoute + * @property {URL} url - Route base URL + * @property {string} endpoint - Route endpoint + * @property {object} headers - Route-specific headers + * @property {import('node:https').Agent} [agent] - Optional HTTPS proxy agent + */ + +/** + * @typedef {object} ActiveWriterRoute + * @property {URL} url - Route base URL + * @property {string} endpoint - Route endpoint + * @property {object} requestOptions - HTTP request options + */ + +/** + * Tests whether a local route definitively rejected an event batch. + * + * @param {Error | null} error - Request error + * @param {number | undefined} statusCode - HTTP response status + * @returns {boolean} Whether direct retry is safe + */ +function isDefinitiveRejection (error, statusCode) { + return error?.code === 'ECONNREFUSED' || statusCode === 403 || statusCode === 404 || statusCode === 405 +} + +/** + * Base writer for Feature Flagging and Experimentation event delivery. * @class BaseFFEWriter */ class BaseFFEWriter { @@ -40,6 +66,7 @@ class BaseFFEWriter { this._payloadSizeLimit = payloadSizeLimit this._eventSizeLimit = eventSizeLimit this._headers = headers || {} + this._fallbackRoute = undefined this._requestOptions = { headers: { @@ -115,15 +142,8 @@ class BaseFFEWriter { // eslint-disable-next-line eslint-rules/eslint-log-printf-style log.debug(() => `${this.constructor.name} flushing payload: ${safeJSONStringify(payload)}`) - request(payload, this._requestOptions, (err, resp, code) => { - if (err) { - log.error('Failed to send events to %s%s', this._baseUrl.href, this._endpoint, err) - } else if (code >= 200 && code < 300) { - log.debug('Successfully sent %d events', events.length) - } else { - log.warn('Events request returned status %d', code) - } - }) + const route = this.#createActiveRoute() + this.#sendRequest(payload, events.length, route, this._fallbackRoute) } /** @@ -161,6 +181,101 @@ class BaseFFEWriter { _encode (payload) { return JSON.stringify(payload) } + + /** + * Applies the active route and an optional direct fallback route. + * + * @param {WriterRoute} route - Active route + * @param {WriterRoute} [fallbackRoute] - Direct fallback route + * @returns {void} + */ + _setRoutes (route, fallbackRoute) { + this.#activateRoute(this.#createRoute(route)) + this._fallbackRoute = fallbackRoute ? this.#createRoute(fallbackRoute) : undefined + } + + /** + * Creates request state for a configured writer route. + * + * @param {WriterRoute} route - Configured route + * @returns {ActiveWriterRoute} Active route state + */ + #createRoute (route) { + return { + url: route.url, + endpoint: route.endpoint, + requestOptions: { + ...(route.agent && { agent: route.agent }), + headers: { + ...route.headers, + 'Content-Type': 'application/json', + }, + method: 'POST', + timeout: this._timeout, + url: route.url, + path: route.endpoint, + }, + } + } + + /** + * Captures the current route for one event batch. + * + * @returns {ActiveWriterRoute} Active route state + */ + #createActiveRoute () { + return { + url: this._baseUrl, + endpoint: this._endpoint, + requestOptions: this._requestOptions, + } + } + + /** + * Makes a route active for future event batches. + * + * @param {ActiveWriterRoute} route - Route state + * @returns {void} + */ + #activateRoute (route) { + this._baseUrl = route.url + this._endpoint = route.endpoint + this._requestOptions = route.requestOptions + } + + /** + * Sends an encoded batch and retries it directly only after definitive rejection. + * + * @param {string} payload - Encoded event batch + * @param {number} eventCount - Event count + * @param {ActiveWriterRoute} route - Selected route + * @param {ActiveWriterRoute} [fallbackRoute] - Direct fallback route + * @returns {void} + */ + #sendRequest (payload, eventCount, route, fallbackRoute) { + request(payload, route.requestOptions, (error, response, statusCode) => { + if (fallbackRoute && isDefinitiveRejection(error, statusCode)) { + log.debug( + '%s switching from %s%s to direct intake after definitive rejection', + this.constructor.name, + route.url.href, + route.endpoint + ) + this.#activateRoute(fallbackRoute) + this._fallbackRoute = undefined + this.#sendRequest(payload, eventCount, fallbackRoute) + return + } + + if (error) { + log.error('Failed to send events to %s%s: %s', route.url.href, route.endpoint, error.message) + } else if (statusCode >= 200 && statusCode < 300) { + log.debug('Successfully sent %d events', eventCount) + } else { + log.warn('Events request returned status %d', statusCode) + } + }) + } } module.exports = BaseFFEWriter diff --git a/packages/dd-trace/src/openfeature/writers/exposures.js b/packages/dd-trace/src/openfeature/writers/exposures.js index 370dad21733..c41d344d71f 100644 --- a/packages/dd-trace/src/openfeature/writers/exposures.js +++ b/packages/dd-trace/src/openfeature/writers/exposures.js @@ -6,9 +6,9 @@ const { EVP_EVENT_SIZE_LIMIT, } = require('../constants/constants') const { + EVP_EVENT_PLATFORM_SUBDOMAIN, EVP_PROXY_PATH_V2, EVP_SUBDOMAIN_HEADER_NAME, - EVP_EVENT_PLATFORM_SUBDOMAIN, } = require('../../evp_proxy/constants') const { joinEVPProxyPath } = require('../../evp_proxy/path') const log = require('../../log') @@ -19,6 +19,15 @@ const BaseFFEWriter = require('./base') // drop emits a warning and `droppedEventCount` accumulates the cumulative loss. const PENDING_MAX_EVENTS = 1000 +/** + * @typedef {object} ExposureRoute + * @property {URL} url - Route base URL + * @property {string} basePath - EVP base path + * @property {object} [headers] - Route-specific headers + * @property {import('node:https').Agent} [agent] - Optional HTTPS proxy agent + * @property {ExposureRoute} [fallback] - Optional direct fallback route + */ + /** * @typedef {object} ExposureEvent * @property {number} timestamp - Unix timestamp in milliseconds @@ -48,10 +57,10 @@ const PENDING_MAX_EVENTS = 1000 */ /** - * ExposuresWriter is responsible for sending exposure events to the Datadog Agent. + * Sends exposure events through the selected local EVP proxy or direct intake route. */ class ExposuresWriter extends BaseFFEWriter { - // Disabled until the agent strategy probe resolves. + // Disabled until route selection resolves. #enabled = false /** @type {ExposureEvent[]} */ @@ -64,23 +73,27 @@ class ExposuresWriter extends BaseFFEWriter { /** * @param {import('../../config/config-base')} config - Tracer configuration object - * @param {{url: URL, basePath: string}} [route] - Caller-supplied local EVP route + * @param {ExposureRoute} [route] - Caller-supplied route */ constructor (config, route) { route ??= { url: config.url, basePath: EVP_PROXY_PATH_V2 } - const fullEndpoint = joinEVPProxyPath(route.basePath, EXPOSURES_ENDPOINT) + const headers = route.headers ?? { + [EVP_SUBDOMAIN_HEADER_NAME]: EVP_EVENT_PLATFORM_SUBDOMAIN, + } super({ config, agentUrl: route.url, - endpoint: fullEndpoint, + endpoint: joinEVPProxyPath(route.basePath, EXPOSURES_ENDPOINT), payloadSizeLimit: EVP_PAYLOAD_SIZE_LIMIT, eventSizeLimit: EVP_EVENT_SIZE_LIMIT, - headers: { - [EVP_SUBDOMAIN_HEADER_NAME]: EVP_EVENT_PLATFORM_SUBDOMAIN, - }, + headers, }) + if (route.agent || route.fallback) { + this.#setRoute({ ...route, headers }) + } + /** @type {ExposureContext} */ const context = { service: config.service, @@ -99,7 +112,8 @@ class ExposuresWriter extends BaseFFEWriter { /** * @param {boolean} enabled - Whether to enable the writer - * @param {{url: URL, basePath: string}} [route] - Discovered local EVP route + * @param {ExposureRoute} [route] - Selected EVP route + * @returns {void} */ setEnabled (enabled, route) { if (route) { @@ -118,16 +132,26 @@ class ExposuresWriter extends BaseFFEWriter { /** * Applies caller-supplied route data without performing discovery. * - * @param {{url: URL, basePath: string}} route - Local EVP route + * @param {ExposureRoute} route - Selected EVP route * @returns {void} */ #setRoute (route) { - const endpoint = joinEVPProxyPath(route.basePath, EXPOSURES_ENDPOINT) + const fallbackRoute = route.fallback && { + url: route.fallback.url, + endpoint: joinEVPProxyPath(route.fallback.basePath, EXPOSURES_ENDPOINT), + headers: route.fallback.headers ?? {}, + agent: route.fallback.agent, + } + const headers = route.headers ?? { + [EVP_SUBDOMAIN_HEADER_NAME]: EVP_EVENT_PLATFORM_SUBDOMAIN, + } - this._baseUrl = route.url - this._endpoint = endpoint - this._requestOptions.url = route.url - this._requestOptions.path = endpoint + this._setRoutes({ + url: route.url, + endpoint: joinEVPProxyPath(route.basePath, EXPOSURES_ENDPOINT), + headers, + agent: route.agent, + }, fallbackRoute) } /** diff --git a/packages/dd-trace/test/openfeature/writers/exposures.spec.js b/packages/dd-trace/test/openfeature/writers/exposures.spec.js index 977c3f3c791..127ad77f58e 100644 --- a/packages/dd-trace/test/openfeature/writers/exposures.spec.js +++ b/packages/dd-trace/test/openfeature/writers/exposures.spec.js @@ -364,6 +364,25 @@ describe('OpenFeature Exposures Writer', () => { assert.strictEqual(parsedPayload.context.service, 'test-service') }) + it('should flush events through the selected EVP v4 proxy path', () => { + const url = new URL('http://serverless-init:9126') + writer.setEnabled(true, { + url, + basePath: '/evp_proxy/v4/', + headers: { + 'X-Datadog-EVP-Subdomain': 'event-platform-intake', + }, + }) + writer.append(exposureEvent) + + writer.flush() + + const [, options] = request.getCall(0).args + assert.strictEqual(options.url, url) + assert.strictEqual(options.path, '/evp_proxy/v4/api/v2/exposures') + assert.strictEqual(options.headers['X-Datadog-EVP-Subdomain'], 'event-platform-intake') + }) + it('should use a caller-supplied route without performing discovery', () => { const url = new URL('http://custom-agent:9126') writer.setEnabled(true, { @@ -379,6 +398,107 @@ describe('OpenFeature Exposures Writer', () => { assert.strictEqual(options.path, '/evp_proxy/v2/api/v2/exposures') }) + it('should flush events directly to HTTPS intake without the local EVP prefix', () => { + const url = new URL('https://event-platform-intake.datadoghq.com') + const agent = {} + writer.setEnabled(true, { + url, + basePath: '', + agent, + headers: { + 'DD-API-KEY': 'test-api-key', + }, + }) + writer.append(exposureEvent) + + writer.flush() + + const [, options] = request.getCall(0).args + assert.strictEqual(options.url, url) + assert.strictEqual(options.path, '/api/v2/exposures') + assert.strictEqual(options.agent, agent) + assert.strictEqual(options.headers['DD-API-KEY'], 'test-api-key') + assert.strictEqual(options.headers['X-Datadog-EVP-Subdomain'], undefined) + }) + + for (const [name, error, statusCode] of [ + ['connection refusal', Object.assign(new Error('connect ECONNREFUSED'), { code: 'ECONNREFUSED' })], + ['HTTP 403', Object.assign(new Error('Forbidden'), { status: 403 }), 403], + ['HTTP 404', Object.assign(new Error('Not Found'), { status: 404 }), 404], + ['HTTP 405', Object.assign(new Error('Method Not Allowed'), { status: 405 }), 405], + ]) { + it(`should switch to direct intake after definitive local ${name}`, async () => { + const localUrl = new URL('http://serverless-init:8126') + const directUrl = new URL('https://event-platform-intake.datadoghq.com') + const directAgent = {} + request.onFirstCall().yieldsAsync(error, null, statusCode) + writer.setEnabled(true, { + url: localUrl, + basePath: '/evp_proxy/v4', + headers: { + 'X-Datadog-EVP-Subdomain': 'event-platform-intake', + }, + fallback: { + url: directUrl, + basePath: '', + agent: directAgent, + headers: { + 'DD-API-KEY': 'test-api-key', + }, + }, + }) + writer.append(exposureEvent) + + writer.flush() + await clock.tickAsync(0) + + sinon.assert.calledTwice(request) + assert.strictEqual(request.firstCall.args[1].url, localUrl) + assert.strictEqual(request.firstCall.args[1].path, '/evp_proxy/v4/api/v2/exposures') + assert.strictEqual(request.secondCall.args[1].url, directUrl) + assert.strictEqual(request.secondCall.args[1].path, '/api/v2/exposures') + assert.strictEqual(request.secondCall.args[1].agent, directAgent) + assert.strictEqual(request.secondCall.args[1].headers['DD-API-KEY'], 'test-api-key') + + writer.append(exposureEvent) + writer.flush() + + sinon.assert.calledThrice(request) + assert.strictEqual(request.thirdCall.args[1].url, directUrl) + }) + } + + for (const [name, error, statusCode] of [ + ['connection reset', Object.assign(new Error('socket hang up'), { code: 'ECONNRESET' })], + ['timeout', Object.assign(new Error('request timed out'), { code: 'ETIMEDOUT' })], + ['HTTP 429', Object.assign(new Error('Too Many Requests'), { status: 429 }), 429], + ['HTTP 500', Object.assign(new Error('Internal Server Error'), { status: 500 }), 500], + ]) { + it(`should not retry ambiguous local ${name} through direct intake`, async () => { + request.yieldsAsync(error, null, statusCode) + writer.setEnabled(true, { + url: new URL('http://serverless-init:8126'), + basePath: '/evp_proxy/v4', + headers: { + 'X-Datadog-EVP-Subdomain': 'event-platform-intake', + }, + fallback: { + url: new URL('https://event-platform-intake.datadoghq.com'), + basePath: '', + headers: { + 'DD-API-KEY': 'test-api-key', + }, + }, + }) + writer.append(exposureEvent) + + writer.flush() + await clock.tickAsync(0) + + sinon.assert.calledOnce(request) + }) + } + it('should empty buffer after flushing', () => { writer.append(exposureEvent) assert.strictEqual(writer._buffer?.length, 1) From ed2bf61379e4ae3967bb3164ea7003498fcda727 Mon Sep 17 00:00:00 2001 From: Leo Romanovsky Date: Tue, 18 Aug 2026 10:42:04 -0700 Subject: [PATCH 2/5] fix(openfeature): harden exposure route fallback --- .../dd-trace/src/openfeature/writers/base.js | 26 ++++++++- .../openfeature/writers/exposures.spec.js | 55 ++++++++++++++++++- 2 files changed, 77 insertions(+), 4 deletions(-) diff --git a/packages/dd-trace/src/openfeature/writers/base.js b/packages/dd-trace/src/openfeature/writers/base.js index c0007c2963b..2331860638c 100644 --- a/packages/dd-trace/src/openfeature/writers/base.js +++ b/packages/dd-trace/src/openfeature/writers/base.js @@ -40,7 +40,18 @@ const log = require('../../log') * @returns {boolean} Whether direct retry is safe */ function isDefinitiveRejection (error, statusCode) { - return error?.code === 'ECONNREFUSED' || statusCode === 403 || statusCode === 404 || statusCode === 405 + return error?.code === 'ECONNREFUSED' || error?.code === 'ENOENT' || + statusCode === 403 || statusCode === 404 || statusCode === 405 +} + +/** + * Tests whether a local route can have accepted an event batch before failing. + * + * @param {Error | null} error - Request error + * @returns {boolean} Whether the current batch must not be replayed + */ +function isAmbiguousNetworkFailure (error) { + return error?.code === 'ECONNRESET' || error?.code === 'ETIMEDOUT' } /** @@ -74,6 +85,7 @@ class BaseFFEWriter { 'Content-Type': 'application/json', }, method: 'POST', + retry: false, timeout: this._timeout, url: this._baseUrl, path: this._endpoint, @@ -211,6 +223,7 @@ class BaseFFEWriter { 'Content-Type': 'application/json', }, method: 'POST', + retry: false, timeout: this._timeout, url: route.url, path: route.endpoint, @@ -267,6 +280,17 @@ class BaseFFEWriter { return } + if (fallbackRoute && isAmbiguousNetworkFailure(error)) { + log.debug( + '%s switching future batches from %s%s to direct intake after ambiguous failure', + this.constructor.name, + route.url.href, + route.endpoint + ) + this.#activateRoute(fallbackRoute) + this._fallbackRoute = undefined + } + if (error) { log.error('Failed to send events to %s%s: %s', route.url.href, route.endpoint, error.message) } else if (statusCode >= 200 && statusCode < 300) { diff --git a/packages/dd-trace/test/openfeature/writers/exposures.spec.js b/packages/dd-trace/test/openfeature/writers/exposures.spec.js index 127ad77f58e..67a6ea82b8a 100644 --- a/packages/dd-trace/test/openfeature/writers/exposures.spec.js +++ b/packages/dd-trace/test/openfeature/writers/exposures.spec.js @@ -348,6 +348,7 @@ describe('OpenFeature Exposures Writer', () => { const [payload, options] = request.getCall(0).args assert.strictEqual(options.method, 'POST') + assert.strictEqual(options.retry, false) assert.match(options.path, /\/evp_proxy\/v2\//) assert.strictEqual(options.headers['Content-Type'], 'application/json') assert.strictEqual(options.headers['X-Datadog-EVP-Subdomain'], 'event-platform-intake') @@ -416,6 +417,7 @@ describe('OpenFeature Exposures Writer', () => { const [, options] = request.getCall(0).args assert.strictEqual(options.url, url) assert.strictEqual(options.path, '/api/v2/exposures') + assert.strictEqual(options.retry, false) assert.strictEqual(options.agent, agent) assert.strictEqual(options.headers['DD-API-KEY'], 'test-api-key') assert.strictEqual(options.headers['X-Datadog-EVP-Subdomain'], undefined) @@ -423,6 +425,10 @@ describe('OpenFeature Exposures Writer', () => { for (const [name, error, statusCode] of [ ['connection refusal', Object.assign(new Error('connect ECONNREFUSED'), { code: 'ECONNREFUSED' })], + ['missing Unix socket', Object.assign( + new Error('connect ENOENT /var/run/datadog/apm.socket'), + { code: 'ENOENT' } + )], ['HTTP 403', Object.assign(new Error('Forbidden'), { status: 403 }), 403], ['HTTP 404', Object.assign(new Error('Not Found'), { status: 404 }), 404], ['HTTP 405', Object.assign(new Error('Method Not Allowed'), { status: 405 }), 405], @@ -471,13 +477,50 @@ describe('OpenFeature Exposures Writer', () => { for (const [name, error, statusCode] of [ ['connection reset', Object.assign(new Error('socket hang up'), { code: 'ECONNRESET' })], ['timeout', Object.assign(new Error('request timed out'), { code: 'ETIMEDOUT' })], + ]) { + it(`should switch future batches after ambiguous local ${name} without replaying the current batch`, async () => { + const localUrl = new URL('http://serverless-init:8126') + const directUrl = new URL('https://event-platform-intake.datadoghq.com') + request.onFirstCall().yieldsAsync(error, null, statusCode) + writer.setEnabled(true, { + url: localUrl, + basePath: '/evp_proxy/v4', + headers: { + 'X-Datadog-EVP-Subdomain': 'event-platform-intake', + }, + fallback: { + url: directUrl, + basePath: '', + headers: { + 'DD-API-KEY': 'test-api-key', + }, + }, + }) + writer.append(exposureEvent) + + writer.flush() + await clock.tickAsync(0) + + sinon.assert.calledOnce(request) + assert.strictEqual(request.firstCall.args[1].url, localUrl) + + writer.append(exposureEvent) + writer.flush() + + sinon.assert.calledTwice(request) + assert.strictEqual(request.secondCall.args[1].url, directUrl) + }) + } + + for (const [name, error, statusCode] of [ ['HTTP 429', Object.assign(new Error('Too Many Requests'), { status: 429 }), 429], ['HTTP 500', Object.assign(new Error('Internal Server Error'), { status: 500 }), 500], ]) { - it(`should not retry ambiguous local ${name} through direct intake`, async () => { - request.yieldsAsync(error, null, statusCode) + it(`should not replay ${name} through direct intake or switch future batches`, async () => { + const localUrl = new URL('http://serverless-init:8126') + request.onFirstCall().yieldsAsync(error, null, statusCode) writer.setEnabled(true, { - url: new URL('http://serverless-init:8126'), + url: localUrl, basePath: '/evp_proxy/v4', headers: { 'X-Datadog-EVP-Subdomain': 'event-platform-intake', @@ -496,6 +539,12 @@ describe('OpenFeature Exposures Writer', () => { await clock.tickAsync(0) sinon.assert.calledOnce(request) + + writer.append(exposureEvent) + writer.flush() + + sinon.assert.calledTwice(request) + assert.strictEqual(request.secondCall.args[1].url, localUrl) }) } From 3294487f377efb7f4478942825fce6c9aa6c2e69 Mon Sep 17 00:00:00 2001 From: Leo Romanovsky Date: Tue, 18 Aug 2026 11:52:49 -0700 Subject: [PATCH 3/5] fix(openfeature): preserve exposure delivery retries --- packages/dd-trace/src/openfeature/writers/base.js | 4 ++-- packages/dd-trace/test/openfeature/writers/exposures.spec.js | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/packages/dd-trace/src/openfeature/writers/base.js b/packages/dd-trace/src/openfeature/writers/base.js index 2331860638c..8d06b424381 100644 --- a/packages/dd-trace/src/openfeature/writers/base.js +++ b/packages/dd-trace/src/openfeature/writers/base.js @@ -85,7 +85,7 @@ class BaseFFEWriter { 'Content-Type': 'application/json', }, method: 'POST', - retry: false, + retry: true, timeout: this._timeout, url: this._baseUrl, path: this._endpoint, @@ -223,7 +223,7 @@ class BaseFFEWriter { 'Content-Type': 'application/json', }, method: 'POST', - retry: false, + retry: true, timeout: this._timeout, url: route.url, path: route.endpoint, diff --git a/packages/dd-trace/test/openfeature/writers/exposures.spec.js b/packages/dd-trace/test/openfeature/writers/exposures.spec.js index 67a6ea82b8a..4a5f5cd7901 100644 --- a/packages/dd-trace/test/openfeature/writers/exposures.spec.js +++ b/packages/dd-trace/test/openfeature/writers/exposures.spec.js @@ -348,7 +348,7 @@ describe('OpenFeature Exposures Writer', () => { const [payload, options] = request.getCall(0).args assert.strictEqual(options.method, 'POST') - assert.strictEqual(options.retry, false) + assert.strictEqual(options.retry, true) assert.match(options.path, /\/evp_proxy\/v2\//) assert.strictEqual(options.headers['Content-Type'], 'application/json') assert.strictEqual(options.headers['X-Datadog-EVP-Subdomain'], 'event-platform-intake') @@ -417,7 +417,7 @@ describe('OpenFeature Exposures Writer', () => { const [, options] = request.getCall(0).args assert.strictEqual(options.url, url) assert.strictEqual(options.path, '/api/v2/exposures') - assert.strictEqual(options.retry, false) + assert.strictEqual(options.retry, true) assert.strictEqual(options.agent, agent) assert.strictEqual(options.headers['DD-API-KEY'], 'test-api-key') assert.strictEqual(options.headers['X-Datadog-EVP-Subdomain'], undefined) From 65047b4ec6b644d6cafcf55431de1d619eba671e Mon Sep 17 00:00:00 2001 From: Leo Romanovsky Date: Tue, 18 Aug 2026 14:09:32 -0700 Subject: [PATCH 4/5] perf(evp_proxy): vendor direct intake proxy dependencies --- package.json | 4 +- packages/dd-trace/src/evp_proxy/direct.js | 4 +- .../dd-trace/test/evp_proxy/direct.spec.js | 4 +- vendor/package-lock.json | 56 +++++++++++++++++++ vendor/package.json | 2 + yarn.lock | 2 +- 6 files changed, 64 insertions(+), 8 deletions(-) diff --git a/package.json b/package.json index b221e71fb0f..61a7c6ce51f 100644 --- a/package.json +++ b/package.json @@ -179,10 +179,8 @@ ], "dependencies": { "dc-polyfill": "^0.1.11", - "https-proxy-agent": "^7.0.6", "import-in-the-middle": "^3.3.2", - "opentracing": ">=0.14.7", - "proxy-from-env": "^2.1.0" + "opentracing": ">=0.14.7" }, "optionalDependencies": { "@datadog/libdatadog": "0.12.1", diff --git a/packages/dd-trace/src/evp_proxy/direct.js b/packages/dd-trace/src/evp_proxy/direct.js index 77186ade847..b4dbf83c090 100644 --- a/packages/dd-trace/src/evp_proxy/direct.js +++ b/packages/dd-trace/src/evp_proxy/direct.js @@ -2,8 +2,8 @@ const { format } = require('node:url') -const { HttpsProxyAgent } = require('https-proxy-agent') -const { getProxyForUrl } = require('proxy-from-env') +const { HttpsProxyAgent } = require('../../../../vendor/dist/https-proxy-agent') +const { getProxyForUrl } = require('../../../../vendor/dist/proxy-from-env') const log = require('../log') /** diff --git a/packages/dd-trace/test/evp_proxy/direct.spec.js b/packages/dd-trace/test/evp_proxy/direct.spec.js index 3dde3900357..396a61284d1 100644 --- a/packages/dd-trace/test/evp_proxy/direct.spec.js +++ b/packages/dd-trace/test/evp_proxy/direct.spec.js @@ -18,8 +18,8 @@ describe('direct EVP route', () => { log = { debug: sinon.spy() } ;({ createDirectEVPRoute } = proxyquire('../../src/evp_proxy/direct', { - 'https-proxy-agent': { HttpsProxyAgent }, - 'proxy-from-env': { getProxyForUrl }, + '../../../../vendor/dist/https-proxy-agent': { HttpsProxyAgent }, + '../../../../vendor/dist/proxy-from-env': { getProxyForUrl }, '../log': log, })) }) diff --git a/vendor/package-lock.json b/vendor/package-lock.json index 72edc90abba..19c50d0a6ef 100644 --- a/vendor/package-lock.json +++ b/vendor/package-lock.json @@ -17,6 +17,7 @@ "crypto-randomuuid": "^1.0.0", "escape-string-regexp": "^5.0.0", "esquery": "^1.7.0", + "https-proxy-agent": "^7.0.6", "istanbul-lib-coverage": "^3.2.2", "jest-docblock": "^29.7.0", "jsonpath-plus": "^10.4.0", @@ -28,6 +29,7 @@ "mutexify": "^1.4.0", "pprof-format": "^2.3.0", "protobufjs": "^8.7.1", + "proxy-from-env": "^2.1.0", "retry": "^0.13.1", "rfdc": "^1.4.1", "semifies": "^1.0.0", @@ -511,6 +513,15 @@ "integrity": "sha512-dWHzHa2WqEXI/O1E9OjrocMTKJl2mSrEolh1Iomrv6U+JuNwaHXsXx9bLu5gG7BUWFIN0skIQJQ/L1rIex4X6w==", "license": "MIT" }, + "node_modules/agent-base": { + "version": "7.1.4", + "resolved": "https://registry.npmjs.org/agent-base/-/agent-base-7.1.4.tgz", + "integrity": "sha512-MnA+YT8fwfJPgBx3m60MNqakm30XOkyIoH1y6huTQvC0PwZG7ki8NacLBcrPbNoo8vEZy7Jpuk7+jMO+CUovTQ==", + "license": "MIT", + "engines": { + "node": ">= 14" + } + }, "node_modules/astring": { "version": "1.9.0", "resolved": "https://registry.npmjs.org/astring/-/astring-1.9.0.tgz", @@ -526,6 +537,23 @@ "integrity": "sha512-/RC5F4l1SCqD/jazwUF6+t34Cd8zTSAGZ7rvvZu1whZUhD2a5MOGKjSGowoGcpj/cbVZk1ZODIooJEQQq3nNAA==", "license": "MIT" }, + "node_modules/debug": { + "version": "4.4.3", + "resolved": "https://registry.npmjs.org/debug/-/debug-4.4.3.tgz", + "integrity": "sha512-RGwwWnwQvkVfavKVt22FGLw+xYSdzARwm0ru6DhTVA3umU5hZc28V3kO4stgYryrTlLpuvgI9GiijltAjNbcqA==", + "license": "MIT", + "dependencies": { + "ms": "^2.1.3" + }, + "engines": { + "node": ">=6.0" + }, + "peerDependenciesMeta": { + "supports-color": { + "optional": true + } + } + }, "node_modules/detect-newline": { "version": "3.1.0", "resolved": "https://registry.npmjs.org/detect-newline/-/detect-newline-3.1.0.tgz", @@ -574,6 +602,19 @@ "integrity": "sha512-/d9sfos4yxzpwkDkuN7k2SqFKtYNmCTzgfEpz82x34IM9/zc8KGxQoXg1liNC/izpRM/MBdt44Nmx41ZWqk+FQ==", "license": "MIT" }, + "node_modules/https-proxy-agent": { + "version": "7.0.6", + "resolved": "https://registry.npmjs.org/https-proxy-agent/-/https-proxy-agent-7.0.6.tgz", + "integrity": "sha512-vK9P5/iUfdl95AI+JVyUuIcVtd4ofvtrOr3HNtM2yxC9bnMbEdp3x01OhQNnjb8IJYi38VlTE3mBXwcfvywuSw==", + "license": "MIT", + "dependencies": { + "agent-base": "^7.1.2", + "debug": "4" + }, + "engines": { + "node": ">= 14" + } + }, "node_modules/istanbul-lib-coverage": { "version": "3.2.2", "resolved": "https://registry.npmjs.org/istanbul-lib-coverage/-/istanbul-lib-coverage-3.2.2.tgz", @@ -678,6 +719,12 @@ "integrity": "sha512-EGWKgxALGMgzvxYF1UyGTy0HXX/2vHLkw6+NvDKW2jypWbHpjQuj4UMcqQWXHERJhVGKikolT06G3bcKe4fi7w==", "license": "MIT" }, + "node_modules/ms": { + "version": "2.1.3", + "resolved": "https://registry.npmjs.org/ms/-/ms-2.1.3.tgz", + "integrity": "sha512-6FlzubTLZG3J2a/NVCAleEhjzq5oxgHyaCU9yYXvcLsvoVaHJq/s5xXI6/XXP6tz7R9xAOtHnSO/tXtF3WRTlA==", + "license": "MIT" + }, "node_modules/mutexify": { "version": "1.4.0", "resolved": "https://registry.npmjs.org/mutexify/-/mutexify-1.4.0.tgz", @@ -705,6 +752,15 @@ "node": ">=12.0.0" } }, + "node_modules/proxy-from-env": { + "version": "2.1.0", + "resolved": "https://registry.npmjs.org/proxy-from-env/-/proxy-from-env-2.1.0.tgz", + "integrity": "sha512-cJ+oHTW1VAEa8cJslgmUZrc+sjRKgAKl3Zyse6+PV38hZe/V6Z14TbCuXcan9F9ghlz4QrFr2c92TNF82UkYHA==", + "license": "MIT", + "engines": { + "node": ">=10" + } + }, "node_modules/queue-tick": { "version": "1.0.1", "resolved": "https://registry.npmjs.org/queue-tick/-/queue-tick-1.0.1.tgz", diff --git a/vendor/package.json b/vendor/package.json index 6e1ed0e12f3..aa43c1e17c1 100644 --- a/vendor/package.json +++ b/vendor/package.json @@ -14,6 +14,7 @@ "crypto-randomuuid": "^1.0.0", "escape-string-regexp": "^5.0.0", "esquery": "^1.7.0", + "https-proxy-agent": "^7.0.6", "istanbul-lib-coverage": "^3.2.2", "jest-docblock": "^29.7.0", "jsonpath-plus": "^10.4.0", @@ -25,6 +26,7 @@ "mutexify": "^1.4.0", "pprof-format": "^2.3.0", "protobufjs": "^8.7.1", + "proxy-from-env": "^2.1.0", "retry": "^0.13.1", "rfdc": "^1.4.1", "semifies": "^1.0.0", diff --git a/yarn.lock b/yarn.lock index e72f4276d4a..79eb11e6600 100644 --- a/yarn.lock +++ b/yarn.lock @@ -2588,7 +2588,7 @@ https-proxy-agent@^5.0.1: agent-base "6" debug "4" -https-proxy-agent@^7.0.5, https-proxy-agent@^7.0.6: +https-proxy-agent@^7.0.5: version "7.0.6" resolved "https://registry.yarnpkg.com/https-proxy-agent/-/https-proxy-agent-7.0.6.tgz#da8dfeac7da130b05c2ba4b59c9b6cd66611a6b9" integrity sha512-vK9P5/iUfdl95AI+JVyUuIcVtd4ofvtrOr3HNtM2yxC9bnMbEdp3x01OhQNnjb8IJYi38VlTE3mBXwcfvywuSw== From 48a3c789cff0bdcc801957282cf38a6665c97f21 Mon Sep 17 00:00:00 2001 From: Leo Romanovsky Date: Tue, 18 Aug 2026 14:39:57 -0700 Subject: [PATCH 5/5] fix(openfeature): retry exposure fallback failures --- packages/dd-trace/src/evp_proxy/discovery.js | 7 ++++--- packages/dd-trace/src/openfeature/writers/base.js | 13 ++++++++----- .../test/openfeature/writers/exposures.spec.js | 12 ++++++++---- 3 files changed, 20 insertions(+), 12 deletions(-) diff --git a/packages/dd-trace/src/evp_proxy/discovery.js b/packages/dd-trace/src/evp_proxy/discovery.js index a32485841b4..86b6d51e1de 100644 --- a/packages/dd-trace/src/evp_proxy/discovery.js +++ b/packages/dd-trace/src/evp_proxy/discovery.js @@ -29,9 +29,10 @@ const TRAILING_SLASHES = /\/+$/ * `/info` endpoint returns an error through the shared request timeout and * retry policy. A valid response without a compatible path returns no route. * Discovery sends no events, so the caller can safely select direct intake - * after either result. The caller also owns later delivery failures. It can - * switch future batches after an ambiguous timeout or reset, but it must not - * replay the current batch because the first receiver might have accepted it. + * after either result. The caller also owns later delivery failures. Exposure + * delivery uses retries and can therefore produce duplicates. After local + * retries fail, the caller can retry through direct intake and use that route + * for future batches. * * Reference implementations: * diff --git a/packages/dd-trace/src/openfeature/writers/base.js b/packages/dd-trace/src/openfeature/writers/base.js index 8d06b424381..e05cd74cec4 100644 --- a/packages/dd-trace/src/openfeature/writers/base.js +++ b/packages/dd-trace/src/openfeature/writers/base.js @@ -40,7 +40,8 @@ const log = require('../../log') * @returns {boolean} Whether direct retry is safe */ function isDefinitiveRejection (error, statusCode) { - return error?.code === 'ECONNREFUSED' || error?.code === 'ENOENT' || + return error?.code === 'EAI_AGAIN' || error?.code === 'ECONNREFUSED' || + error?.code === 'ENOENT' || error?.code === 'ENOTFOUND' || statusCode === 403 || statusCode === 404 || statusCode === 405 } @@ -48,10 +49,10 @@ function isDefinitiveRejection (error, statusCode) { * Tests whether a local route can have accepted an event batch before failing. * * @param {Error | null} error - Request error - * @returns {boolean} Whether the current batch must not be replayed + * @returns {boolean} Whether the delivery result is ambiguous */ function isAmbiguousNetworkFailure (error) { - return error?.code === 'ECONNRESET' || error?.code === 'ETIMEDOUT' + return error?.code === 'ECONNRESET' || error?.code === 'EPIPE' || error?.code === 'ETIMEDOUT' } /** @@ -257,7 +258,7 @@ class BaseFFEWriter { } /** - * Sends an encoded batch and retries it directly only after definitive rejection. + * Sends an encoded batch and retries it through direct intake after a local route failure. * * @param {string} payload - Encoded event batch * @param {number} eventCount - Event count @@ -282,13 +283,15 @@ class BaseFFEWriter { if (fallbackRoute && isAmbiguousNetworkFailure(error)) { log.debug( - '%s switching future batches from %s%s to direct intake after ambiguous failure', + '%s retrying through direct intake and switching future batches from %s%s after ambiguous failure', this.constructor.name, route.url.href, route.endpoint ) this.#activateRoute(fallbackRoute) this._fallbackRoute = undefined + this.#sendRequest(payload, eventCount, fallbackRoute) + return } if (error) { diff --git a/packages/dd-trace/test/openfeature/writers/exposures.spec.js b/packages/dd-trace/test/openfeature/writers/exposures.spec.js index 4a5f5cd7901..922dc8b172d 100644 --- a/packages/dd-trace/test/openfeature/writers/exposures.spec.js +++ b/packages/dd-trace/test/openfeature/writers/exposures.spec.js @@ -424,11 +424,13 @@ describe('OpenFeature Exposures Writer', () => { }) for (const [name, error, statusCode] of [ + ['temporary DNS failure', Object.assign(new Error('getaddrinfo EAI_AGAIN'), { code: 'EAI_AGAIN' })], ['connection refusal', Object.assign(new Error('connect ECONNREFUSED'), { code: 'ECONNREFUSED' })], ['missing Unix socket', Object.assign( new Error('connect ENOENT /var/run/datadog/apm.socket'), { code: 'ENOENT' } )], + ['unresolvable hostname', Object.assign(new Error('getaddrinfo ENOTFOUND'), { code: 'ENOTFOUND' })], ['HTTP 403', Object.assign(new Error('Forbidden'), { status: 403 }), 403], ['HTTP 404', Object.assign(new Error('Not Found'), { status: 404 }), 404], ['HTTP 405', Object.assign(new Error('Method Not Allowed'), { status: 405 }), 405], @@ -476,9 +478,10 @@ describe('OpenFeature Exposures Writer', () => { for (const [name, error, statusCode] of [ ['connection reset', Object.assign(new Error('socket hang up'), { code: 'ECONNRESET' })], + ['broken pipe', Object.assign(new Error('write EPIPE'), { code: 'EPIPE' })], ['timeout', Object.assign(new Error('request timed out'), { code: 'ETIMEDOUT' })], ]) { - it(`should switch future batches after ambiguous local ${name} without replaying the current batch`, async () => { + it(`should retry ambiguous local ${name} through direct intake and switch future batches`, async () => { const localUrl = new URL('http://serverless-init:8126') const directUrl = new URL('https://event-platform-intake.datadoghq.com') request.onFirstCall().yieldsAsync(error, null, statusCode) @@ -501,14 +504,15 @@ describe('OpenFeature Exposures Writer', () => { writer.flush() await clock.tickAsync(0) - sinon.assert.calledOnce(request) + sinon.assert.calledTwice(request) assert.strictEqual(request.firstCall.args[1].url, localUrl) + assert.strictEqual(request.secondCall.args[1].url, directUrl) writer.append(exposureEvent) writer.flush() - sinon.assert.calledTwice(request) - assert.strictEqual(request.secondCall.args[1].url, directUrl) + sinon.assert.calledThrice(request) + assert.strictEqual(request.thirdCall.args[1].url, directUrl) }) }