feat(openfeature): add exposure route fallback

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",

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')
 
 /**

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:
  *

packages/dd-trace/src/openfeature/writers/base.js

@@ -11,14 +11,52 @@ 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 === 'EAI_AGAIN' || error?.code === 'ECONNREFUSED' ||
+    error?.code === 'ENOENT' || error?.code === 'ENOTFOUND' ||
+    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 delivery result is ambiguous
+ */
+function isAmbiguousNetworkFailure (error) {
+  return error?.code === 'ECONNRESET' || error?.code === 'EPIPE' || error?.code === 'ETIMEDOUT'
+}
+
+/**
+ * Base writer for Feature Flagging and Experimentation event delivery.
  * @class BaseFFEWriter
  */
 class BaseFFEWriter {
@@ -40,13 +78,15 @@ class BaseFFEWriter {
     this._payloadSizeLimit = payloadSizeLimit
     this._eventSizeLimit = eventSizeLimit
     this._headers = headers || {}
+    this._fallbackRoute = undefined
 
     this._requestOptions = {
       headers: {
         ...this._headers,
         'Content-Type': 'application/json',
       },
       method: 'POST',
+      retry: true,
       timeout: this._timeout,
       url: this._baseUrl,
       path: this._endpoint,
@@ -115,15 +155,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 +194,115 @@ 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',
+        retry: true,
+        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 through direct intake after a local route failure.
+   *
+   * @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 (fallbackRoute && isAmbiguousNetworkFailure(error)) {
+        log.debug(
+          '%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) {
+        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

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)
   }
 
   /**

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,
     }))
   })