Skip to content

Commit ec9d0a4

Browse files
leoromanovskyleoromanovsky
committed
feat(openfeature): improve exposure buffering
1 parent 2ea4035 commit ec9d0a4

3 files changed

Lines changed: 235 additions & 122 deletions

File tree

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

Lines changed: 95 additions & 30 deletions
Original file line numberDiff line numberDiff line change
@@ -58,7 +58,8 @@ class BaseFFEWriter {
5858

5959
this._buffer = []
6060
this._bufferLimit = 1000
61-
this._bufferSize = 0
61+
this._bufferStart = 0
62+
this._dropWarningLogged = false
6263

6364
this._config = config
6465
this._endpoint = endpoint
@@ -99,51 +100,87 @@ class BaseFFEWriter {
99100
const eventArray = Array.isArray(events) ? events : [events]
100101

101102
for (const event of eventArray) {
102-
if (this._buffer.length >= this._bufferLimit) {
103-
log.warn('%s event buffer full (limit is %d), dropping event', this.constructor.name, this._bufferLimit)
104-
this._droppedEvents++
105-
continue
106-
}
107-
108-
const eventSizeBytes = Buffer.byteLength(JSON.stringify(event))
109-
110-
// Check individual event size limit if configured
111-
if (this._eventSizeLimit && eventSizeBytes > this._eventSizeLimit) {
112-
log.warn('%s event size %d bytes exceeds limit %d, dropping event',
113-
this.constructor.name, eventSizeBytes, this._eventSizeLimit)
103+
if (this._buffer.length < this._bufferLimit) {
104+
this._buffer.push(event)
105+
} else {
106+
this._buffer[this._bufferStart] = event
107+
this._bufferStart = (this._bufferStart + 1) % this._bufferLimit
114108
this._droppedEvents++
115-
continue
116-
}
117109

118-
// Check if adding this event would exceed payload size limit if configured
119-
if (this._payloadSizeLimit && this._bufferSize + eventSizeBytes > this._payloadSizeLimit) {
120-
log.debug('%s buffer size would exceed %d bytes, flushing first', this.constructor.name, this._payloadSizeLimit)
121-
this.flush()
110+
if (!this._dropWarningLogged) {
111+
this._dropWarningLogged = true
112+
log.warn(
113+
'%s dropped exposure event(s) at cap %d. This may invalidate experiment results.',
114+
this.constructor.name,
115+
this._bufferLimit
116+
)
117+
}
122118
}
123-
124-
this._bufferSize += eventSizeBytes
125-
this._buffer.push(event)
126119
}
127120
}
128121

129122
/**
130-
* Flushes all buffered events to the agent
123+
* Sizes, batches, and flushes all buffered events.
131124
*/
132125
flush () {
133126
if (this._buffer.length === 0) {
134127
return
135128
}
136-
const events = this._buffer
129+
130+
const events = this._bufferStart === 0
131+
? this._buffer
132+
: [...this._buffer.slice(this._bufferStart), ...this._buffer.slice(0, this._bufferStart)]
137133
this._buffer = []
138-
this._bufferSize = 0
134+
this._bufferStart = 0
139135

140-
const payload = this._encode(this.makePayload(events))
136+
let batch = []
137+
let batchSize = 0
141138

142-
// eslint-disable-next-line eslint-rules/eslint-log-printf-style
143-
log.debug(() => `${this.constructor.name} flushing payload: ${safeJSONStringify(payload)}`)
139+
for (const event of events) {
140+
let eventSize
141+
try {
142+
eventSize = Buffer.byteLength(JSON.stringify(event))
143+
} catch (error) {
144+
log.warn('%s could not serialize an event, dropping event: %s', this.constructor.name, error.message)
145+
this._droppedEvents++
146+
continue
147+
}
144148

145-
const route = this.#createActiveRoute()
146-
this.#sendRequest(payload, events.length, route, this._fallbackRoute)
149+
if (this._eventSizeLimit && eventSize > this._eventSizeLimit) {
150+
log.warn(
151+
'%s event size %d bytes exceeds limit %d, dropping event',
152+
this.constructor.name,
153+
eventSize,
154+
this._eventSizeLimit
155+
)
156+
this._droppedEvents++
157+
continue
158+
}
159+
160+
if (this._payloadSizeLimit && eventSize > this._payloadSizeLimit) {
161+
log.warn(
162+
'%s event size %d bytes exceeds payload limit %d, dropping event',
163+
this.constructor.name,
164+
eventSize,
165+
this._payloadSizeLimit
166+
)
167+
this._droppedEvents++
168+
continue
169+
}
170+
171+
if (batch.length > 0 && this._payloadSizeLimit && batchSize + eventSize > this._payloadSizeLimit) {
172+
this.#send(batch)
173+
batch = []
174+
batchSize = 0
175+
}
176+
177+
batch.push(event)
178+
batchSize += eventSize
179+
}
180+
181+
if (batch.length > 0) {
182+
this.#send(batch)
183+
}
147184
}
148185

149186
/**
@@ -194,6 +231,34 @@ class BaseFFEWriter {
194231
this._fallbackRoute = fallbackRoute ? this.#createRoute(fallbackRoute) : undefined
195232
}
196233

234+
/**
235+
* Sends one event batch.
236+
*
237+
* @param {Array<object>} events - Events in the batch
238+
* @returns {void}
239+
*/
240+
#send (events) {
241+
let payload
242+
try {
243+
payload = this._encode(this.makePayload(events))
244+
} catch (error) {
245+
log.warn(
246+
'%s could not encode %d event(s), dropping batch: %s',
247+
this.constructor.name,
248+
events.length,
249+
error.message
250+
)
251+
this._droppedEvents += events.length
252+
return
253+
}
254+
255+
// eslint-disable-next-line eslint-rules/eslint-log-printf-style
256+
log.debug(() => `${this.constructor.name} flushing payload: ${safeJSONStringify(payload)}`)
257+
258+
const route = this.#createActiveRoute()
259+
this.#sendRequest(payload, events.length, route, this._fallbackRoute)
260+
}
261+
197262
/**
198263
* Creates request state for a configured writer route.
199264
*

‎packages/dd-trace/src/openfeature/writers/exposures.js‎

Lines changed: 47 additions & 42 deletions
Original file line numberDiff line numberDiff line change
@@ -11,14 +11,8 @@ const {
1111
EVP_SUBDOMAIN_HEADER_NAME,
1212
} = require('../../evp_proxy/constants')
1313
const { joinEVPProxyPath } = require('../../evp_proxy/path')
14-
const log = require('../../log')
1514
const BaseFFEWriter = require('./base')
1615

17-
// Disabled-state cap. Drops invalidate experiment results because the provider's
18-
// exposure dedupe cache keeps masking dropped events after recovery. The first
19-
// drop emits a warning and `droppedEventCount` accumulates the cumulative loss.
20-
const PENDING_MAX_EVENTS = 1000
21-
2216
/**
2317
* @typedef {object} ExposureRoute
2418
* @property {URL} url - Route base URL
@@ -63,14 +57,14 @@ class ExposuresWriter extends BaseFFEWriter {
6357
// Disabled until route selection resolves.
6458
#enabled = false
6559

66-
/** @type {ExposureEvent[]} */
67-
#pendingEvents = []
60+
#routeResolved = false
61+
62+
/** @type {ReturnType<typeof setImmediate> | undefined} */
63+
#startupFlushImmediate
6864

6965
/** @type {ExposureContext} */
7066
#context
7167

72-
#dropWarned = false
73-
7468
/**
7569
* @param {import('../../config/config-base')} config - Tracer configuration object
7670
* @param {ExposureRoute} [route] - Caller-supplied route
@@ -116,16 +110,20 @@ class ExposuresWriter extends BaseFFEWriter {
116110
* @returns {void}
117111
*/
118112
setEnabled (enabled, route) {
113+
this.#routeResolved = true
114+
119115
if (route) {
120116
this.#setRoute(route)
121117
}
122118

123119
this.#enabled = enabled
124120

125-
if (enabled && this.#pendingEvents.length > 0) {
126-
// Flush all pending events as a batch
127-
super.append(this.#pendingEvents)
128-
this.#pendingEvents = []
121+
if (enabled) {
122+
this.#scheduleStartupFlush()
123+
} else {
124+
this.#cancelStartupFlush()
125+
this._buffer = []
126+
this._bufferStart = 0
129127
}
130128
}
131129

@@ -155,50 +153,57 @@ class ExposuresWriter extends BaseFFEWriter {
155153
}
156154

157155
/**
158-
* Appends exposure event(s) to the buffer
156+
* Appends exposure event(s) to the buffer.
157+
*
159158
* @param {ExposureEvent|ExposureEvent[]} events - Exposure event(s) to append
159+
* @returns {void}
160160
*/
161161
append (events) {
162-
if (this.#enabled) {
163-
super.append(events)
164-
return
165-
}
162+
if (this.#routeResolved && !this.#enabled) return
166163

167-
const eventArray = Array.isArray(events) ? events : [events]
168-
this.#pendingEvents.push(...eventArray)
169-
if (this.#pendingEvents.length > PENDING_MAX_EVENTS) {
170-
const dropped = this.#pendingEvents.length - PENDING_MAX_EVENTS
171-
this.#pendingEvents.splice(0, dropped)
172-
this._droppedEvents += dropped
173-
if (!this.#dropWarned) {
174-
this.#dropWarned = true
175-
log.warn(
176-
'%s dropped exposure event(s) at cap %d. This may invalidate experiment results.',
177-
this.constructor.name, PENDING_MAX_EVENTS)
178-
}
179-
}
164+
super.append(events)
180165
}
181166

182167
/**
183-
* @returns {number} Cumulative number of exposure events dropped due to buffer overflow.
168+
* Flushes buffered exposure events through the selected route.
169+
*
170+
* @returns {void}
184171
*/
185-
get droppedEventCount () {
186-
return this._droppedEvents
172+
flush () {
173+
if (!this.#enabled) return
174+
175+
this.#cancelStartupFlush()
176+
super.flush()
187177
}
188178

189179
/**
190-
* Flushes buffered exposure events to the agent
180+
* Flushes startup events after route selection completes.
181+
*
182+
* @returns {void}
191183
*/
192-
flush () {
193-
if (!this.#enabled) {
194-
return
195-
}
196-
super.flush()
184+
#scheduleStartupFlush () {
185+
if (this.#startupFlushImmediate || this._buffer.length === 0) return
186+
187+
this.#startupFlushImmediate = setImmediate(() => {
188+
this.#startupFlushImmediate = undefined
189+
this.flush()
190+
})
191+
}
192+
193+
/**
194+
* Cancels a scheduled drain.
195+
*
196+
* @returns {void}
197+
*/
198+
#cancelStartupFlush () {
199+
if (!this.#startupFlushImmediate) return
200+
clearImmediate(this.#startupFlushImmediate)
201+
this.#startupFlushImmediate = undefined
197202
}
198203

199204
/**
200205
* Formats exposure events with service context metadata
201-
* @param {Array<ExposureEvent>} events - Array of exposure events
206+
* @param {Array<ExposureEvent>} events - Array of exposure events to format
202207
* @returns {ExposureEventPayload} Formatted payload with service context
203208
*/
204209
makePayload (events) {

0 commit comments

Comments
 (0)