Skip to content

Commit 438bcb4

Browse files
MichaelDoyleMichaelDoyle
committed
fix(tracing): upgrade to otel 2.0
1 parent 9950eb8 commit 438bcb4

16 files changed

Lines changed: 2117 additions & 618 deletions

File tree

‎js/core/package.json‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -26,12 +26,12 @@
2626
"author": "genkit",
2727
"license": "Apache-2.0",
2828
"dependencies": {
29-
"@opentelemetry/api": "^1.9.0",
30-
"@opentelemetry/context-async-hooks": "~1.25.0",
31-
"@opentelemetry/core": "~1.25.0",
32-
"@opentelemetry/sdk-metrics": "~1.25.0",
33-
"@opentelemetry/sdk-node": "^0.52.0",
34-
"@opentelemetry/sdk-trace-base": "~1.25.0",
29+
"@opentelemetry/api": "^1.9.1",
30+
"@opentelemetry/context-async-hooks": "^2.6.1",
31+
"@opentelemetry/core": "^2.6.1",
32+
"@opentelemetry/sdk-metrics": "^2.6.1",
33+
"@opentelemetry/sdk-node": "^0.214.0",
34+
"@opentelemetry/sdk-trace-base": "^2.6.1",
3535
"@types/json-schema": "^7.0.15",
3636
"ajv": "^8.12.0",
3737
"ajv-formats": "^3.0.1",

‎js/core/src/tracing.ts‎

Lines changed: 4 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -100,9 +100,10 @@ export async function enableTelemetry(
100100
if (isOTelInitializationDisabled()) {
101101
return;
102102
}
103-
global[instrumentationKey] =
104-
telemetryConfig instanceof Promise ? telemetryConfig : Promise.resolve();
105-
return getTelemetryProvider().enableTelemetry(telemetryConfig);
103+
const instrumentationPromise =
104+
getTelemetryProvider().enableTelemetry(telemetryConfig);
105+
global[instrumentationKey] = instrumentationPromise;
106+
return instrumentationPromise;
106107
}
107108

108109
/**

‎js/core/src/tracing/exporter.ts‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -73,7 +73,7 @@ export class TraceServerExporter implements SpanExporter {
7373
displayName: span.name,
7474
links: span.links,
7575
spanKind: SpanKind[span.kind],
76-
parentSpanId: span.parentSpanId,
76+
parentSpanId: span.parentSpanContext?.spanId,
7777
sameProcessAsParentSpan: { value: !span.spanContext().isRemote },
7878
status: span.status,
7979
timeEvents: {
@@ -86,17 +86,17 @@ export class TraceServerExporter implements SpanExporter {
8686
})),
8787
},
8888
};
89-
if (span.instrumentationLibrary !== undefined) {
89+
if (span.instrumentationScope !== undefined) {
9090
spanData.instrumentationLibrary = {
91-
name: span.instrumentationLibrary.name,
91+
name: span.instrumentationScope.name,
9292
};
93-
if (span.instrumentationLibrary.schemaUrl !== undefined) {
93+
if (span.instrumentationScope.schemaUrl !== undefined) {
9494
spanData.instrumentationLibrary.schemaUrl =
95-
span.instrumentationLibrary.schemaUrl;
95+
span.instrumentationScope.schemaUrl;
9696
}
97-
if (span.instrumentationLibrary.version !== undefined) {
97+
if (span.instrumentationScope.version !== undefined) {
9898
spanData.instrumentationLibrary.version =
99-
span.instrumentationLibrary.version;
99+
span.instrumentationScope.version;
100100
}
101101
}
102102
deleteUndefinedProps(spanData);

‎js/core/src/tracing/node-telemetry-provider.ts‎

Lines changed: 93 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -14,10 +14,13 @@
1414
* limitations under the License.
1515
*/
1616

17+
import { Context } from '@opentelemetry/api';
1718
import { NodeSDK } from '@opentelemetry/sdk-node';
1819
import {
1920
BatchSpanProcessor,
2021
SimpleSpanProcessor,
22+
type ReadableSpan,
23+
type Span,
2124
type SpanProcessor,
2225
} from '@opentelemetry/sdk-trace-base';
2326
import { logger } from '../logging.js';
@@ -29,6 +32,7 @@ import { RealtimeSpanProcessor } from './realtime-span-processor.js';
2932

3033
let telemetrySDK: NodeSDK | null = null;
3134
let nodeOtelConfig: TelemetryConfig | null = null;
35+
let isSigtermHandlerRegistered = false;
3236

3337
export function initNodeTelemetryProvider() {
3438
setTelemetryProvider({
@@ -37,6 +41,64 @@ export function initNodeTelemetryProvider() {
3741
});
3842
}
3943

44+
/**
45+
* MultiSpanProcessor is a multiplexer that allows Genkit to register multiple
46+
* span processors reliably.
47+
*
48+
* We provide it as the sole entry in the `spanProcessors` array to the
49+
* OpenTelemetry NodeSDK because the SDK's internal logic for merging
50+
* `traceExporter`, `spanProcessor` (singular), and `spanProcessors` (plural)
51+
* varies across versions and can lead to exporters being silently overwritten
52+
* or ignored.
53+
*
54+
* By wrapping all processors into this single delegate, we guarantee that
55+
* Genkit's internal telemetry and any user-provided exporters both receive
56+
* every start and end event.
57+
*/
58+
export class MultiSpanProcessor implements SpanProcessor {
59+
constructor(private processors: SpanProcessor[]) {}
60+
onStart(span: Span, parentContext: Context) {
61+
this.processors.forEach((p) => {
62+
try {
63+
p.onStart?.(span, parentContext);
64+
} catch (e) {
65+
logger.error(`Error in span processor onStart: ${e}`);
66+
}
67+
});
68+
}
69+
onEnd(span: ReadableSpan) {
70+
this.processors.forEach((p) => {
71+
try {
72+
p.onEnd(span);
73+
} catch (e) {
74+
logger.error(`Error in span processor onEnd: ${e}`);
75+
}
76+
});
77+
}
78+
async forceFlush() {
79+
await Promise.all(
80+
this.processors.map(async (p) => {
81+
try {
82+
await p.forceFlush();
83+
} catch (e) {
84+
logger.error(`Error in span processor forceFlush: ${e}`);
85+
}
86+
})
87+
);
88+
}
89+
async shutdown() {
90+
await Promise.all(
91+
this.processors.map(async (p) => {
92+
try {
93+
await p.shutdown();
94+
} catch (e) {
95+
logger.error(`Error in span processor shutdown: ${e}`);
96+
}
97+
})
98+
);
99+
}
100+
}
101+
40102
/**
41103
* Enables tracing and metrics open telemetry configuration.
42104
*/
@@ -52,23 +114,48 @@ async function enableTelemetry(
52114
? await telemetryConfig
53115
: telemetryConfig;
54116

55-
nodeOtelConfig = telemetryConfig || {};
117+
// If already initialized and new config is empty,
118+
// skip to avoid unnecessary restarts
119+
if (
120+
telemetrySDK &&
121+
(!telemetryConfig ||
122+
(!telemetryConfig.spanProcessors &&
123+
!telemetryConfig.spanProcessor &&
124+
!telemetryConfig.traceExporter &&
125+
!telemetryConfig.metricReader))
126+
) {
127+
return;
128+
}
129+
130+
nodeOtelConfig = { ...(telemetryConfig || {}) };
56131

57132
const processors: SpanProcessor[] = [createTelemetryServerProcessor()];
58-
if (nodeOtelConfig.traceExporter) {
59-
throw new Error('Please specify spanProcessors instead.');
60-
}
61133
if (nodeOtelConfig.spanProcessors) {
62134
processors.push(...nodeOtelConfig.spanProcessors);
63135
}
64136
if (nodeOtelConfig.spanProcessor) {
65137
processors.push(nodeOtelConfig.spanProcessor);
66138
delete nodeOtelConfig.spanProcessor;
67139
}
68-
nodeOtelConfig.spanProcessors = processors;
140+
141+
if (processors.length > 1) {
142+
nodeOtelConfig.spanProcessors = [new MultiSpanProcessor(processors)];
143+
} else {
144+
nodeOtelConfig.spanProcessors = processors;
145+
}
146+
delete nodeOtelConfig.spanProcessor;
147+
148+
if (telemetrySDK) {
149+
await cleanUpTracing();
150+
}
151+
69152
telemetrySDK = new NodeSDK(nodeOtelConfig);
70153
telemetrySDK.start();
71-
process.on('SIGTERM', async () => await cleanUpTracing());
154+
155+
if (!isSigtermHandlerRegistered) {
156+
process.on('SIGTERM', async () => await cleanUpTracing());
157+
isSigtermHandlerRegistered = true;
158+
}
72159
}
73160

74161
async function cleanUpTracing(): Promise<void> {

‎js/core/tests/action_test.ts‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@ import { z } from 'zod';
2121
import { action, defineAction } from '../src/action.js';
2222
import { initNodeFeatures } from '../src/node.js';
2323
import { Registry } from '../src/registry.js';
24-
import { enableTelemetry } from '../src/tracing.js';
24+
import { enableTelemetry, flushTracing } from '../src/tracing.js';
2525
import { TestSpanExporter } from './utils.js';
2626

2727
initNodeFeatures();
@@ -33,8 +33,9 @@ enableTelemetry({
3333

3434
describe('action', () => {
3535
var registry: Registry;
36-
beforeEach(() => {
36+
beforeEach(async () => {
3737
registry = new Registry();
38+
await flushTracing();
3839
spanExporter.exportedSpans = [];
3940
});
4041

@@ -289,6 +290,7 @@ describe('action', () => {
289290
act.__action.key = 'some-custom-key';
290291

291292
await act();
293+
await flushTracing();
292294

293295
assert.strictEqual(spanExporter.exportedSpans.length, 1);
294296
assert.strictEqual(

‎js/core/tests/flow_test.ts‎

Lines changed: 10 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@ import { defineFlow, run } from '../src/flow.js';
2121
import { defineAction, getContext, z } from '../src/index.js';
2222
import { initNodeFeatures } from '../src/node.js';
2323
import { Registry } from '../src/registry.js';
24-
import { enableTelemetry } from '../src/tracing.js';
24+
import { enableTelemetry, flushTracing } from '../src/tracing.js';
2525
import { TestSpanExporter } from './utils.js';
2626

2727
initNodeFeatures();
@@ -48,10 +48,12 @@ function createTestFlow(registry: Registry) {
4848
describe('flow', () => {
4949
let registry: Registry;
5050

51-
beforeEach(() => {
51+
beforeEach(async () => {
5252
// Skips starting reflection server.
5353
delete process.env.GENKIT_ENV;
5454
registry = new Registry();
55+
await flushTracing();
56+
spanExporter.exportedSpans = [];
5557
});
5658

5759
describe('runFlow', () => {
@@ -370,6 +372,8 @@ describe('flow', () => {
370372
telemetryLabels: { custom: 'label' },
371373
});
372374

375+
await flushTracing();
376+
373377
assert.equal(result, 'bar foo');
374378
assert.strictEqual(spanExporter.exportedSpans.length, 1);
375379
assert.strictEqual(spanExporter.exportedSpans[0].displayName, 'testFlow');
@@ -394,6 +398,8 @@ describe('flow', () => {
394398
});
395399
const result = await output;
396400

401+
await flushTracing();
402+
397403
assert.equal(result, 'bar foo');
398404
assert.strictEqual(spanExporter.exportedSpans.length, 1);
399405
assert.strictEqual(spanExporter.exportedSpans[0].displayName, 'testFlow');
@@ -445,6 +451,8 @@ describe('flow', () => {
445451
context: { user: 'pavel' },
446452
});
447453

454+
await flushTracing();
455+
448456
assert.equal(result, 'foo bar');
449457
assert.strictEqual(spanExporter.exportedSpans.length, 3);
450458

0 commit comments

Comments
 (0)