Skip to content

Commit d97074c

Browse files
petyosipetyosi
committed
Address Cloudflare instrumentation review fixes
1 parent 48d78c7 commit d97074c

9 files changed

Lines changed: 224 additions & 70 deletions

File tree

‎packages/otel-cf-workers/src/instrumentation/analytics-engine.ts‎

Lines changed: 10 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -7,15 +7,18 @@ type ExtraAttributeFn = (argArray: any[], result: any) => Attributes
77

88
const dbSystem = 'Cloudflare Analytics Engine'
99

10-
const AEAttributes: Record<string | symbol, ExtraAttributeFn> = {
10+
export const AEAttributes: Record<string | symbol, ExtraAttributeFn> = {
1111
writeDataPoint(argArray) {
1212
const attrs: Attributes = {}
13-
const opts = argArray[0]
14-
if (typeof opts === 'object') {
15-
attrs['db.cf.ae.indexes'] = opts.indexes.length
16-
attrs['db.cf.ae.index'] = (opts.indexes[0] as ArrayBuffer | string).toString()
17-
attrs['db.cf.ae.doubles'] = opts.doubles.length
18-
attrs['db.cf.ae.blobs'] = opts.blobs.length
13+
const opts = argArray[0] as AnalyticsEngineDataPoint | undefined
14+
if (typeof opts === 'object' && opts !== null) {
15+
const firstIndex = opts.indexes?.[0]
16+
attrs['db.cf.ae.indexes'] = opts.indexes?.length ?? 0
17+
if (firstIndex !== undefined && firstIndex !== null) {
18+
attrs['db.cf.ae.index'] = firstIndex.toString()
19+
}
20+
attrs['db.cf.ae.doubles'] = opts.doubles?.length ?? 0
21+
attrs['db.cf.ae.blobs'] = opts.blobs?.length ?? 0
1922
}
2023
return attrs
2124
},

‎packages/otel-cf-workers/src/instrumentation/do.ts‎

Lines changed: 16 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -144,26 +144,28 @@ export async function executeDOAlarm(alarmFn: NonNullable<AlarmFn>, id: DurableO
144144
return promise
145145
}
146146

147-
function instrumentFetchFn(fetchFn: FetchFn, initialiser: Initialiser, env: Env, id: DurableObjectId): FetchFn {
147+
function instrumentFetchFn(fetchFn: FetchFn, initialiser: Initialiser, env: Env, state: DurableObjectState): FetchFn {
148148
const fetchHandler: ProxyHandler<FetchFn> = {
149149
async apply(target, thisArg, argArray: Parameters<FetchFn>) {
150150
const request = argArray[0]
151151
const config = initialiser(env, request)
152152
const context = setConfig(config)
153153
try {
154154
const bound = target.bind(unwrap(thisArg))
155-
return await api_context.with(context, executeDOFetch, undefined, bound, request, id)
155+
return await api_context.with(context, executeDOFetch, undefined, bound, request, state.id)
156156
} finally {
157-
exportSpans().catch((error: unknown) => {
158-
console.error('Error exporting Durable Object fetch spans:', error)
159-
})
157+
state.waitUntil(
158+
exportSpans().catch((error: unknown) => {
159+
console.error('Error exporting Durable Object fetch spans:', error)
160+
})
161+
)
160162
}
161163
},
162164
}
163165
return wrap(fetchFn, fetchHandler)
164166
}
165167

166-
function instrumentAlarmFn(alarmFn: AlarmFn, initialiser: Initialiser, env: Env, id: DurableObjectId) {
168+
function instrumentAlarmFn(alarmFn: AlarmFn, initialiser: Initialiser, env: Env, state: DurableObjectState) {
167169
if (!alarmFn) {
168170
return undefined
169171
}
@@ -174,11 +176,13 @@ function instrumentAlarmFn(alarmFn: AlarmFn, initialiser: Initialiser, env: Env,
174176
const context = setConfig(config)
175177
try {
176178
const bound = target.bind(unwrap(thisArg))
177-
await api_context.with(context, executeDOAlarm, undefined, bound, id)
179+
await api_context.with(context, executeDOAlarm, undefined, bound, state.id)
178180
} finally {
179-
exportSpans().catch((error: unknown) => {
180-
console.error('Error exporting Durable Object alarm spans:', error)
181-
})
181+
state.waitUntil(
182+
exportSpans().catch((error: unknown) => {
183+
console.error('Error exporting Durable Object alarm spans:', error)
184+
})
185+
)
182186
}
183187
},
184188
}
@@ -190,10 +194,10 @@ function instrumentDurableObject(doObj: DurableObject, initialiser: Initialiser,
190194
get(target, prop) {
191195
if (prop === 'fetch') {
192196
const fetchFn = Reflect.get(target, prop)
193-
return instrumentFetchFn(fetchFn, initialiser, env, state.id)
197+
return instrumentFetchFn(fetchFn, initialiser, env, state)
194198
} else if (prop === 'alarm') {
195199
const alarmFn = Reflect.get(target, prop)
196-
return instrumentAlarmFn(alarmFn, initialiser, env, state.id)
200+
return instrumentAlarmFn(alarmFn, initialiser, env, state)
197201
} else {
198202
const result = Reflect.get(target, prop)
199203
if (typeof result === 'function') {

‎packages/otel-cf-workers/src/instrumentation/kv.ts‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,7 @@ type ExtraAttributeFn = (argArray: any[], result: any) => Attributes
77

88
const dbSystem = 'Cloudflare KV'
99

10-
const KVAttributes: Record<string | symbol, ExtraAttributeFn> = {
10+
export const KVAttributes: Record<string | symbol, ExtraAttributeFn> = {
1111
delete(_argArray) {
1212
return {}
1313
},
@@ -45,10 +45,11 @@ const KVAttributes: Record<string | symbol, ExtraAttributeFn> = {
4545
const { cursor, limit } = opts
4646
attrs['db.cf.kv.list_request_cursor'] = cursor || undefined
4747
attrs['db.cf.kv.list_limit'] = limit || undefined
48-
const { list_complete, cacheStatus } = result as KVNamespaceListResult<any, any>
48+
const listResult = result as KVNamespaceListResult<any, any>
49+
const { list_complete, cacheStatus } = listResult
4950
attrs['db.cf.kv.list_complete'] = list_complete || undefined
5051
if (!list_complete) {
51-
attrs['db.cf.kv.list_response_cursor'] = cursor || undefined
52+
attrs['db.cf.kv.list_response_cursor'] = listResult.cursor || undefined
5253
}
5354
if (typeof cacheStatus === 'string') {
5455
attrs['db.cf.kv.cache_status'] = cacheStatus

‎packages/otel-cf-workers/src/multiexporter.ts‎

Lines changed: 52 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -2,18 +2,62 @@ import type { ReadableSpan, SpanExporter } from '@opentelemetry/sdk-trace-base'
22
import type { ExportResult } from '@opentelemetry/core'
33
import { ExportResultCode } from '@opentelemetry/core'
44

5-
// First implementation, completely synchronous, more tested.
5+
type ExportResultCallback = (result: ExportResult) => void
6+
7+
function resultFromError(error: unknown): ExportResult {
8+
return { code: ExportResultCode.FAILED, error: error instanceof Error ? error : new Error(String(error)) }
9+
}
10+
11+
function combinedExportResult(results: ExportResult[]): ExportResult {
12+
const failed = results.find((result) => result.code === ExportResultCode.FAILED)
13+
if (failed) {
14+
return failed.error === undefined ? { code: ExportResultCode.FAILED } : { code: ExportResultCode.FAILED, error: failed.error }
15+
}
16+
return { code: ExportResultCode.SUCCESS }
17+
}
18+
19+
function exportToAll(exporters: SpanExporter[], items: ReadableSpan[], resultCallback: ExportResultCallback): void {
20+
if (exporters.length === 0) {
21+
resultCallback({ code: ExportResultCode.SUCCESS })
22+
return
23+
}
24+
25+
const results: ExportResult[] = []
26+
let pending = exporters.length
27+
28+
const recordResult = (result: ExportResult): void => {
29+
results.push(result)
30+
pending -= 1
31+
if (pending === 0) {
32+
resultCallback(combinedExportResult(results))
33+
}
34+
}
35+
36+
for (const exporter of exporters) {
37+
let settled = false
38+
const once = (result: ExportResult): void => {
39+
if (!settled) {
40+
settled = true
41+
recordResult(result)
42+
}
43+
}
44+
45+
try {
46+
exporter.export(items, once)
47+
} catch (error) {
48+
once(resultFromError(error))
49+
}
50+
}
51+
}
652

753
export class MultiSpanExporter implements SpanExporter {
854
private readonly exporters: SpanExporter[]
955
constructor(exporters: SpanExporter[]) {
1056
this.exporters = exporters
1157
}
1258

13-
export(items: ReadableSpan[], resultCallback: (result: ExportResult) => void): void {
14-
for (const exporter of this.exporters) {
15-
exporter.export(items, resultCallback)
16-
}
59+
export(items: ReadableSpan[], resultCallback: ExportResultCallback): void {
60+
exportToAll(this.exporters, items, resultCallback)
1761
}
1862

1963
async shutdown(): Promise<void> {
@@ -29,7 +73,7 @@ export class MultiSpanExporterAsync implements SpanExporter {
2973
this.exporters = exporters
3074
}
3175

32-
export(items: ReadableSpan[], resultCallback: (result: ExportResult) => void): void {
76+
export(items: ReadableSpan[], resultCallback: ExportResultCallback): void {
3377
const promises = this.exporters.map(
3478
async (exporter) =>
3579
new Promise<ExportResult>((resolve) => {
@@ -39,17 +83,10 @@ export class MultiSpanExporterAsync implements SpanExporter {
3983

4084
Promise.all(promises).then(
4185
(results) => {
42-
const failed = results.filter((result) => result.code === ExportResultCode.FAILED)
43-
if (failed.length > 0) {
44-
// not ideal, but just return the first error
45-
const error = failed[0]?.error
46-
resultCallback(error === undefined ? { code: ExportResultCode.FAILED } : { code: ExportResultCode.FAILED, error })
47-
} else {
48-
resultCallback({ code: ExportResultCode.SUCCESS })
49-
}
86+
resultCallback(combinedExportResult(results))
5087
},
5188
(error: unknown) => {
52-
resultCallback({ code: ExportResultCode.FAILED, error: error instanceof Error ? error : new Error(String(error)) })
89+
resultCallback(resultFromError(error))
5390
}
5491
)
5592
}

‎packages/otel-cf-workers/src/tracer.ts‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -13,7 +13,7 @@ let withNextSpanAttributes: Attributes
1313

1414
export class WorkerTracer implements Tracer {
1515
private readonly _spanProcessors: SpanProcessor[]
16-
private readonly resource: Resource
16+
private resource: Resource
1717
private readonly scope: InstrumentationScope
1818
private readonly idGenerator: IdGenerator
1919
constructor(spanProcessors: SpanProcessor[], resource: Resource, scope: InstrumentationScope, idGenerator: IdGenerator) {
@@ -28,7 +28,7 @@ export class WorkerTracer implements Tracer {
2828
}
2929

3030
addToResource(extra: Resource): void {
31-
this.resource.merge(extra)
31+
this.resource = this.resource.merge(extra)
3232
}
3333

3434
startSpan(name: string, options: SpanOptions = {}, context: Context = api_context.active()): Span {

‎packages/otel-cf-workers/test/instrumentation/analytics-engine.test.ts‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
import { describe, expect, it } from 'vitest'
2+
import { AEAttributes } from '../../src/instrumentation/analytics-engine'
3+
4+
describe('Analytics Engine attributes', () => {
5+
it('handles partial data points without optional arrays', () => {
6+
expect(AEAttributes.writeDataPoint([{}], undefined)).toEqual({
7+
'db.cf.ae.indexes': 0,
8+
'db.cf.ae.doubles': 0,
9+
'db.cf.ae.blobs': 0,
10+
})
11+
})
12+
13+
it('records provided data point array counts', () => {
14+
expect(AEAttributes.writeDataPoint([{ indexes: ['index'], doubles: [1, 2], blobs: ['blob'] }], undefined)).toEqual({
15+
'db.cf.ae.indexes': 1,
16+
'db.cf.ae.index': 'index',
17+
'db.cf.ae.doubles': 2,
18+
'db.cf.ae.blobs': 1,
19+
})
20+
})
21+
})

‎packages/otel-cf-workers/test/instrumentation/do.test.ts‎

Lines changed: 73 additions & 31 deletions
Original file line numberDiff line numberDiff line change
@@ -1,6 +1,6 @@
11
/// <reference types="@cloudflare/workers-types" />
22

3-
import { describe, expect, it } from 'vitest'
3+
import { describe, expect, it, vitest } from 'vitest'
44
import type { ResolvedTraceConfig } from '../../src/types'
55
import { instrumentDOClass } from '../../src/instrumentation/do'
66

@@ -12,35 +12,48 @@ const durableObjectId = {
1212

1313
const noop = () => undefined
1414

15-
const durableObjectState = {
16-
id: durableObjectId,
17-
storage: {},
18-
waitUntil: noop,
19-
async blockConcurrencyWhile<T>(callback: () => Promise<T>): Promise<T> {
20-
return callback()
21-
},
22-
acceptWebSocket: noop,
23-
getWebSockets() {
24-
return []
25-
},
26-
setWebSocketAutoResponse: noop,
27-
getWebSocketAutoResponse() {
28-
return null
29-
},
30-
getWebSocketAutoResponseTimestamp() {
31-
return null
32-
},
33-
setHibernatableWebSocketEventTimeout: noop,
34-
getHibernatableWebSocketEventTimeout() {
35-
return null
36-
},
37-
getTags() {
38-
return []
39-
},
40-
abort: noop,
41-
props: {},
42-
facets: {},
43-
} as unknown as DurableObjectState
15+
interface TestDurableObjectState extends DurableObjectState {
16+
waitUntilPromises: Promise<unknown>[]
17+
}
18+
19+
function createDurableObjectState(): TestDurableObjectState {
20+
const waitUntilPromises: Promise<unknown>[] = []
21+
22+
return {
23+
id: durableObjectId,
24+
storage: {},
25+
waitUntil(promise: Promise<unknown>) {
26+
waitUntilPromises.push(promise)
27+
},
28+
async blockConcurrencyWhile<T>(callback: () => Promise<T>): Promise<T> {
29+
return callback()
30+
},
31+
acceptWebSocket: noop,
32+
getWebSockets() {
33+
return []
34+
},
35+
setWebSocketAutoResponse: noop,
36+
getWebSocketAutoResponse() {
37+
return null
38+
},
39+
getWebSocketAutoResponseTimestamp() {
40+
return null
41+
},
42+
setHibernatableWebSocketEventTimeout: noop,
43+
getHibernatableWebSocketEventTimeout() {
44+
return null
45+
},
46+
getTags() {
47+
return []
48+
},
49+
abort: noop,
50+
props: {},
51+
facets: {},
52+
waitUntilPromises,
53+
} as unknown as TestDurableObjectState
54+
}
55+
56+
const resolvedConfig = {} as ResolvedTraceConfig
4457

4558
class CustomMethodDurableObject implements DurableObject {
4659
count = 0
@@ -55,18 +68,47 @@ class CustomMethodDurableObject implements DurableObject {
5568
}
5669
}
5770

71+
class LifecycleDurableObject implements DurableObject {
72+
fetch(): Response {
73+
return new Response()
74+
}
75+
76+
alarm(): void {
77+
return undefined
78+
}
79+
}
80+
5881
interface CustomMethodDurableObjectInstance extends DurableObject {
5982
increment(delta: number): number
6083
}
6184

6285
describe('instrumentDOClass', () => {
6386
it('binds custom Durable Object methods to the original object', () => {
64-
const InstrumentedDurableObject = instrumentDOClass(CustomMethodDurableObject, () => ({}) as ResolvedTraceConfig)
87+
const durableObjectState = createDurableObjectState()
88+
const InstrumentedDurableObject = instrumentDOClass(CustomMethodDurableObject, () => resolvedConfig)
6589
const durableObject = new InstrumentedDurableObject(durableObjectState, {}) as unknown as CustomMethodDurableObjectInstance
6690

6791
const increment = Reflect.get(durableObject, 'increment') as (delta: number) => number
6892

6993
expect(increment(2)).toBe(2)
7094
expect(durableObject.increment(3)).toBe(5)
7195
})
96+
97+
it('schedules fetch and alarm span export with Durable Object state waitUntil', async () => {
98+
const consoleError = vitest.spyOn(console, 'error').mockImplementation(() => undefined)
99+
const durableObjectState = createDurableObjectState()
100+
const InstrumentedDurableObject = instrumentDOClass(LifecycleDurableObject, () => resolvedConfig)
101+
const durableObject = new InstrumentedDurableObject(durableObjectState, {})
102+
103+
try {
104+
await durableObject.fetch(new Request('https://example.com'))
105+
const alarm = Reflect.get(durableObject, 'alarm') as () => Promise<void>
106+
await alarm()
107+
108+
expect(durableObjectState.waitUntilPromises).toHaveLength(2)
109+
await Promise.allSettled(durableObjectState.waitUntilPromises)
110+
} finally {
111+
consoleError.mockRestore()
112+
}
113+
})
72114
})

0 commit comments

Comments
 (0)