mirror of
https://github.com/n8n-io/n8n.git
synced 2026-10-17 05:29:44 +02:00
fix(core): Keep OpenTelemetry export working after a restart when Sentry is enabled (#38480)
Co-authored-by: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
ocdejong
GitHub
parent
60bfb02db9
commit
dbefd920b5
|
|
@ -187,7 +187,6 @@
|
|||
"@opentelemetry/exporter-trace-otlp-grpc": "^0.221.0",
|
||||
"@opentelemetry/exporter-trace-otlp-proto": "^0.221.0",
|
||||
"@opentelemetry/resources": "^2.10.0",
|
||||
"@opentelemetry/sdk-node": "^0.221.0",
|
||||
"@opentelemetry/sdk-trace-base": "^2.10.0",
|
||||
"@opentelemetry/sdk-trace-node": "^2.10.0",
|
||||
"@opentelemetry/semantic-conventions": "^1.43.0",
|
||||
|
|
|
|||
|
|
@ -28,6 +28,7 @@ import { License } from '@/license';
|
|||
import { LoadNodesAndCredentials } from '@/load-nodes-and-credentials';
|
||||
import { CommunityPackagesConfig } from '@/modules/community-packages/community-packages.config';
|
||||
import { CommunityPackagesService } from '@/modules/community-packages/community-packages.service';
|
||||
import { OtelService } from '@/modules/otel/otel.service';
|
||||
import { NodeTypes } from '@/node-types';
|
||||
import { PostHogClient } from '@/posthog';
|
||||
import { PollJobProvider } from '@/scheduling/poll-trigger-node/poll-job-provider';
|
||||
|
|
@ -72,6 +73,7 @@ mockInstance(NodeTypes);
|
|||
const shutdownService = mockInstance(ShutdownService);
|
||||
shutdownService.validate.mockReturnValue(undefined);
|
||||
mockInstance(PostHogClient);
|
||||
mockInstance(OtelService);
|
||||
mockInstance(TelemetryEventRelay);
|
||||
mockInstance(ActivityEventRelay);
|
||||
mockInstance(WorkflowFailureNotificationEventRelay);
|
||||
|
|
@ -126,6 +128,7 @@ describe('Start - AuthRolesService initialization', () => {
|
|||
Container.set(MultiMainSetup, multiMainSetup);
|
||||
Container.set(AuthHandlerRegistry, authHandlerRegistry);
|
||||
Container.set(PostHogClient, mockInstance(PostHogClient));
|
||||
Container.set(OtelService, mockInstance(OtelService));
|
||||
Container.set(TelemetryEventRelay, mockInstance(TelemetryEventRelay));
|
||||
Container.set(ActivityEventRelay, mockInstance(ActivityEventRelay));
|
||||
Container.set(
|
||||
|
|
|
|||
|
|
@ -24,6 +24,7 @@ import { WorkflowFailureNotificationEventRelay } from '@/events/relays/workflow-
|
|||
import type { ExecutionPersistence } from '@/executions/execution-persistence';
|
||||
import { LoadNodesAndCredentials } from '@/load-nodes-and-credentials';
|
||||
import { CommunityPackagesConfig } from '@/modules/community-packages/community-packages.config';
|
||||
import { OtelService } from '@/modules/otel/otel.service';
|
||||
import { NodeTypes } from '@/node-types';
|
||||
import { PostHogClient } from '@/posthog';
|
||||
import { PubSubRegistry } from '@/scaling/pubsub/pubsub.registry';
|
||||
|
|
@ -61,6 +62,7 @@ mockInstance(NodeTypes);
|
|||
mockInstance(ShutdownService);
|
||||
mockInstance(MessageEventBus);
|
||||
mockInstance(PostHogClient);
|
||||
mockInstance(OtelService);
|
||||
mockInstance(TelemetryEventRelay);
|
||||
mockInstance(ActivityEventRelay);
|
||||
mockInstance(WorkflowFailureNotificationEventRelay);
|
||||
|
|
|
|||
|
|
@ -4,13 +4,22 @@
|
|||
import type { Logger } from '@n8n/backend-common';
|
||||
import { ExpressionEngineConfig, type GlobalConfig } from '@n8n/config';
|
||||
import { EXPRESSION_METRICS } from '@n8n/expression-runtime';
|
||||
import { trace } from '@opentelemetry/api';
|
||||
import type { Tracer } from '@opentelemetry/api';
|
||||
import promClient from 'prom-client';
|
||||
import { mock } from 'vitest-mock-extended';
|
||||
|
||||
import { ExpressionObservabilityProvider } from '../expression-observability.provider';
|
||||
|
||||
import type { OtelService } from '@/modules/otel/otel.service';
|
||||
|
||||
const scopedLogger = mock<Logger>();
|
||||
const otelService = mock<OtelService>();
|
||||
const startSpanMock = vi.fn().mockReturnValue({
|
||||
setStatus: vi.fn(),
|
||||
setAttribute: vi.fn(),
|
||||
recordException: vi.fn(),
|
||||
end: vi.fn(),
|
||||
});
|
||||
|
||||
function buildConfig(overrides: Partial<ExpressionEngineConfig> = {}): ExpressionEngineConfig {
|
||||
const config = new ExpressionEngineConfig();
|
||||
|
|
@ -32,6 +41,7 @@ function buildGlobalConfig(prefix = 'n8n_'): GlobalConfig {
|
|||
describe('ExpressionObservabilityProvider', () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
otelService.getTracer.mockReturnValue({ startSpan: startSpanMock } as unknown as Tracer);
|
||||
promClient.register.clear();
|
||||
});
|
||||
|
||||
|
|
@ -41,6 +51,7 @@ describe('ExpressionObservabilityProvider', () => {
|
|||
buildConfig({ observabilityEnabled: false }),
|
||||
buildLogger(),
|
||||
buildGlobalConfig(),
|
||||
otelService,
|
||||
);
|
||||
|
||||
expect(() => {
|
||||
|
|
@ -57,6 +68,7 @@ describe('ExpressionObservabilityProvider', () => {
|
|||
buildConfig({ engine: 'legacy' }),
|
||||
buildLogger(),
|
||||
buildGlobalConfig(),
|
||||
otelService,
|
||||
);
|
||||
|
||||
provider.metrics.counter(EXPRESSION_METRICS.poolAcquired.name, 1);
|
||||
|
|
@ -72,6 +84,7 @@ describe('ExpressionObservabilityProvider', () => {
|
|||
buildConfig(),
|
||||
buildLogger(),
|
||||
buildGlobalConfig(),
|
||||
otelService,
|
||||
);
|
||||
provider.metrics.counter(EXPRESSION_METRICS.poolAcquired.name, 2);
|
||||
|
||||
|
|
@ -86,6 +99,7 @@ describe('ExpressionObservabilityProvider', () => {
|
|||
buildConfig(),
|
||||
buildLogger(),
|
||||
buildGlobalConfig(),
|
||||
otelService,
|
||||
);
|
||||
provider.metrics.gauge(EXPRESSION_METRICS.codeCacheSize.name, 42);
|
||||
|
||||
|
|
@ -98,6 +112,7 @@ describe('ExpressionObservabilityProvider', () => {
|
|||
buildConfig(),
|
||||
buildLogger(),
|
||||
buildGlobalConfig(),
|
||||
otelService,
|
||||
);
|
||||
provider.metrics.histogram(EXPRESSION_METRICS.evaluationDuration.name, 3, {
|
||||
status: 'success',
|
||||
|
|
@ -116,6 +131,7 @@ describe('ExpressionObservabilityProvider', () => {
|
|||
buildConfig(),
|
||||
buildLogger(),
|
||||
buildGlobalConfig(),
|
||||
otelService,
|
||||
);
|
||||
|
||||
promClient.register.clear();
|
||||
|
|
@ -140,6 +156,7 @@ describe('ExpressionObservabilityProvider', () => {
|
|||
buildConfig(),
|
||||
buildLogger(),
|
||||
buildGlobalConfig(),
|
||||
otelService,
|
||||
);
|
||||
|
||||
promClient.register.clear();
|
||||
|
|
@ -153,25 +170,12 @@ describe('ExpressionObservabilityProvider', () => {
|
|||
});
|
||||
|
||||
describe('tail sampling', () => {
|
||||
const startSpanMock = vi.fn().mockReturnValue({
|
||||
setStatus: vi.fn(),
|
||||
setAttribute: vi.fn(),
|
||||
recordException: vi.fn(),
|
||||
end: vi.fn(),
|
||||
});
|
||||
|
||||
beforeEach(() => {
|
||||
startSpanMock.mockClear();
|
||||
vi.spyOn(trace, 'getTracer').mockReturnValue({
|
||||
startSpan: startSpanMock,
|
||||
} as unknown as ReturnType<typeof trace.getTracer>);
|
||||
});
|
||||
|
||||
it('drops healthy spans under the slow threshold', () => {
|
||||
const provider = new ExpressionObservabilityProvider(
|
||||
buildConfig({ slowEvaluationThresholdMs: 50, tracesSampleRate: 0 }),
|
||||
buildLogger(),
|
||||
buildGlobalConfig(),
|
||||
otelService,
|
||||
);
|
||||
provider.metrics.histogram(EXPRESSION_METRICS.evaluationDuration.name, 0.01, {
|
||||
status: 'success',
|
||||
|
|
@ -185,6 +189,7 @@ describe('ExpressionObservabilityProvider', () => {
|
|||
buildConfig({ slowEvaluationThresholdMs: 50 }),
|
||||
buildLogger(),
|
||||
buildGlobalConfig(),
|
||||
otelService,
|
||||
);
|
||||
provider.metrics.histogram(EXPRESSION_METRICS.evaluationDuration.name, 0.1, {
|
||||
status: 'success',
|
||||
|
|
@ -206,6 +211,7 @@ describe('ExpressionObservabilityProvider', () => {
|
|||
buildConfig({ slowEvaluationThresholdMs: 50 }),
|
||||
buildLogger(),
|
||||
buildGlobalConfig(),
|
||||
otelService,
|
||||
);
|
||||
provider.metrics.histogram(EXPRESSION_METRICS.evaluationDuration.name, 0.005, {
|
||||
status: 'error',
|
||||
|
|
@ -227,6 +233,7 @@ describe('ExpressionObservabilityProvider', () => {
|
|||
buildConfig({ tracesEnabled: false, slowEvaluationThresholdMs: 10 }),
|
||||
buildLogger(),
|
||||
buildGlobalConfig(),
|
||||
otelService,
|
||||
);
|
||||
provider.metrics.histogram(EXPRESSION_METRICS.evaluationDuration.name, 0.5, {
|
||||
status: 'error',
|
||||
|
|
@ -242,6 +249,7 @@ describe('ExpressionObservabilityProvider', () => {
|
|||
buildConfig(),
|
||||
buildLogger(),
|
||||
buildGlobalConfig(),
|
||||
otelService,
|
||||
);
|
||||
expect(() => {
|
||||
provider.metrics.counter('test.unknown', 1);
|
||||
|
|
@ -256,6 +264,7 @@ describe('ExpressionObservabilityProvider', () => {
|
|||
buildConfig(),
|
||||
buildLogger(),
|
||||
buildGlobalConfig(),
|
||||
otelService,
|
||||
);
|
||||
|
||||
provider.metrics.histogram(EXPRESSION_METRICS.evaluationDuration.name, 0.01, {
|
||||
|
|
@ -285,6 +294,7 @@ describe('ExpressionObservabilityProvider', () => {
|
|||
buildConfig(),
|
||||
buildLogger(),
|
||||
buildGlobalConfig(),
|
||||
otelService,
|
||||
);
|
||||
provider.logs.info('hello', { k: 'v' });
|
||||
provider.logs.warn('warn', { k: 'v' });
|
||||
|
|
|
|||
|
|
@ -10,8 +10,7 @@ import type {
|
|||
TracesAPI,
|
||||
} from '@n8n/expression-runtime';
|
||||
import { EXPRESSION_METRICS, NoOpProvider } from '@n8n/expression-runtime';
|
||||
import { SpanStatusCode, trace } from '@opentelemetry/api';
|
||||
import type { Tracer } from '@opentelemetry/api';
|
||||
import { SpanStatusCode } from '@opentelemetry/api';
|
||||
import { UnexpectedError } from 'n8n-workflow';
|
||||
import promClient, { type Counter, type Gauge, type Histogram } from 'prom-client';
|
||||
|
||||
|
|
@ -26,6 +25,8 @@ import {
|
|||
toPromName,
|
||||
} from './expression-observability.formatters';
|
||||
|
||||
import { OtelService } from '@/modules/otel/otel.service';
|
||||
|
||||
type TailSampleDecision = 'drop' | 'keep';
|
||||
|
||||
@Service()
|
||||
|
|
@ -40,8 +41,6 @@ export class ExpressionObservabilityProvider implements ObservabilityProvider {
|
|||
|
||||
private readonly prefix!: string;
|
||||
|
||||
private tracer?: Tracer;
|
||||
|
||||
private readonly metricDefs = new Map<string, MetricDef>(
|
||||
(Object.values(EXPRESSION_METRICS) as MetricDef[]).map((def) => [def.name, def]),
|
||||
);
|
||||
|
|
@ -50,6 +49,7 @@ export class ExpressionObservabilityProvider implements ObservabilityProvider {
|
|||
private readonly config: ExpressionEngineConfig,
|
||||
private readonly logger: Logger,
|
||||
globalConfig: GlobalConfig,
|
||||
private readonly otelService: OtelService,
|
||||
) {
|
||||
this.scopedLogger = this.logger.scoped('expression-engine');
|
||||
|
||||
|
|
@ -167,7 +167,7 @@ export class ExpressionObservabilityProvider implements ObservabilityProvider {
|
|||
? 'slow'
|
||||
: 'healthy';
|
||||
|
||||
const tracer = this.getTracer();
|
||||
const tracer = this.otelService.getTracer(TRACER_NAME);
|
||||
const errorType = tags?.type && tags.type !== 'none' ? tags.type : undefined;
|
||||
const span = tracer.startSpan('expression.evaluate', {
|
||||
attributes: {
|
||||
|
|
@ -194,7 +194,7 @@ export class ExpressionObservabilityProvider implements ObservabilityProvider {
|
|||
private startSpan(name: string, attributes?: Record<string, unknown>): Span {
|
||||
if (!this.config.tracesEnabled) return NoOpProvider.traces.startSpan(name, attributes);
|
||||
|
||||
const tracer = this.getTracer();
|
||||
const tracer = this.otelService.getTracer(TRACER_NAME);
|
||||
const otelSpan = tracer.startSpan(name, {
|
||||
attributes: normalizeAttributes(attributes),
|
||||
});
|
||||
|
|
@ -212,9 +212,4 @@ export class ExpressionObservabilityProvider implements ObservabilityProvider {
|
|||
end: () => otelSpan.end(),
|
||||
};
|
||||
}
|
||||
|
||||
private getTracer(): Tracer {
|
||||
this.tracer ??= trace.getTracer(TRACER_NAME);
|
||||
return this.tracer;
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,19 +1,22 @@
|
|||
import type { Logger } from '@n8n/backend-common';
|
||||
import type { OutboundHttp } from '@n8n/backend-network';
|
||||
import type { AgentsConfig } from '@n8n/config';
|
||||
import { context, type Tracer } from '@opentelemetry/api';
|
||||
import { mock } from 'vitest-mock-extended';
|
||||
import type { InstanceSettings } from 'n8n-core';
|
||||
|
||||
import { OtelTestProvider } from '@/modules/otel/__tests__/support/otel-test-provider';
|
||||
import { ExecutionLevelTracer } from '@/modules/otel/execution-level-tracer';
|
||||
import type { OtelSettingsService } from '@/modules/otel/otel-settings.service';
|
||||
import { OtelService } from '@/modules/otel/otel.service';
|
||||
|
||||
import { AgentRunTracingService } from '../agent-run-tracing.service';
|
||||
|
||||
// Deliberately does not mock '@opentelemetry/api': agent-run-tracing.service.test.ts
|
||||
// Deliberately uses a real tracer provider: agent-run-tracing.service.test.ts
|
||||
// already covers the built-metadata/branching logic against a stubbed tracer. This
|
||||
// file proves the real OTel path end-to-end — that a registered provider actually
|
||||
// file proves the real OTel path end-to-end — that the module's provider actually
|
||||
// records the span the service's tracer produces, and that a disabled OTel module
|
||||
// (no provider registered) or disabled agent tracing both emit nothing.
|
||||
// (no provider started) or disabled agent tracing both emit nothing.
|
||||
describe('AgentRunTracingService (real OTel provider)', () => {
|
||||
const baseMetadata = {
|
||||
agentId: 'agent-1',
|
||||
|
|
@ -26,7 +29,7 @@ describe('AgentRunTracingService (real OTel provider)', () => {
|
|||
const otel = OtelTestProvider.create();
|
||||
try {
|
||||
const agentsConfig = mock<AgentsConfig>({ tracingEnabled: true });
|
||||
const service = new AgentRunTracingService(agentsConfig);
|
||||
const service = new AgentRunTracingService(agentsConfig, otel.asOtelService());
|
||||
|
||||
const built = await service.build(baseMetadata);
|
||||
expect(built).toBeDefined();
|
||||
|
|
@ -57,6 +60,7 @@ describe('AgentRunTracingService (real OTel provider)', () => {
|
|||
const otel = OtelTestProvider.create({ withContextManager: true });
|
||||
try {
|
||||
const executionLevelTracer = new ExecutionLevelTracer(
|
||||
otel.asOtelService(),
|
||||
mock<OtelSettingsService>({ getSettings: () => ({ injectOutbound: false }) as never }),
|
||||
mock<Logger>(),
|
||||
);
|
||||
|
|
@ -70,7 +74,7 @@ describe('AgentRunTracingService (real OTel provider)', () => {
|
|||
executionLevelTracer.startNode({ executionId, node });
|
||||
|
||||
const agentsConfig = mock<AgentsConfig>({ tracingEnabled: true });
|
||||
const agentRunTracingService = new AgentRunTracingService(agentsConfig);
|
||||
const agentRunTracingService = new AgentRunTracingService(agentsConfig, otel.asOtelService());
|
||||
|
||||
const parentCtx = executionLevelTracer.getActiveContext(executionId, node.name);
|
||||
expect(parentCtx).toBeDefined();
|
||||
|
|
@ -118,15 +122,21 @@ describe('AgentRunTracingService (real OTel provider)', () => {
|
|||
}
|
||||
});
|
||||
|
||||
it('emits no spans when the OTel module has no provider registered', async () => {
|
||||
it('emits no spans when the OTel module has not started a provider', async () => {
|
||||
const agentsConfig = mock<AgentsConfig>({ tracingEnabled: true });
|
||||
const service = new AgentRunTracingService(agentsConfig);
|
||||
const otelService = new OtelService(
|
||||
mock<OtelSettingsService>(),
|
||||
mock<InstanceSettings>(),
|
||||
mock<Logger>(),
|
||||
mock<OutboundHttp>(),
|
||||
);
|
||||
const service = new AgentRunTracingService(agentsConfig, otelService);
|
||||
|
||||
const built = await service.build(baseMetadata);
|
||||
expect(built).toBeDefined();
|
||||
|
||||
// No provider registered — the global tracer falls back to OTel's
|
||||
// no-op implementation, so starting a span records nothing anywhere.
|
||||
// No provider started — the module hands out OTel's no-op tracer, so
|
||||
// starting a span records nothing anywhere.
|
||||
(built?.tracer as Tracer).startActiveSpan('test-span', (span) => {
|
||||
expect(span.isRecording()).toBe(false);
|
||||
span.end();
|
||||
|
|
@ -137,7 +147,7 @@ describe('AgentRunTracingService (real OTel provider)', () => {
|
|||
const otel = OtelTestProvider.create();
|
||||
try {
|
||||
const agentsConfig = mock<AgentsConfig>({ tracingEnabled: false });
|
||||
const service = new AgentRunTracingService(agentsConfig);
|
||||
const service = new AgentRunTracingService(agentsConfig, otel.asOtelService());
|
||||
|
||||
const built = await service.build(baseMetadata);
|
||||
expect(built).toBeUndefined();
|
||||
|
|
|
|||
|
|
@ -1,16 +1,17 @@
|
|||
import type { AgentsConfig } from '@n8n/config';
|
||||
import { trace } from '@opentelemetry/api';
|
||||
import type { Tracer } from '@opentelemetry/api';
|
||||
import { mock } from 'vitest-mock-extended';
|
||||
|
||||
import type { OtelService } from '@/modules/otel/otel.service';
|
||||
|
||||
import { AgentRunTracingService, modelIdFromSnapshot } from '../agent-run-tracing.service';
|
||||
|
||||
vi.mock('@opentelemetry/api', () => ({
|
||||
trace: { getTracer: vi.fn(() => ({ startSpan: vi.fn(), startActiveSpan: vi.fn() })) },
|
||||
}));
|
||||
const otelService = mock<OtelService>();
|
||||
|
||||
describe('AgentRunTracingService', () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
otelService.getTracer.mockReturnValue(mock<Tracer>());
|
||||
});
|
||||
|
||||
const baseMetadata = {
|
||||
|
|
@ -21,38 +22,39 @@ describe('AgentRunTracingService', () => {
|
|||
};
|
||||
|
||||
it('exposes whether tracing is enabled via the `enabled` getter', () => {
|
||||
expect(new AgentRunTracingService(mock<AgentsConfig>({ tracingEnabled: true })).enabled).toBe(
|
||||
true,
|
||||
);
|
||||
expect(new AgentRunTracingService(mock<AgentsConfig>({ tracingEnabled: false })).enabled).toBe(
|
||||
false,
|
||||
);
|
||||
expect(
|
||||
new AgentRunTracingService(mock<AgentsConfig>({ tracingEnabled: true }), otelService).enabled,
|
||||
).toBe(true);
|
||||
expect(
|
||||
new AgentRunTracingService(mock<AgentsConfig>({ tracingEnabled: false }), otelService)
|
||||
.enabled,
|
||||
).toBe(false);
|
||||
});
|
||||
|
||||
it('returns undefined when agent tracing is disabled, without touching the tracer', async () => {
|
||||
const agentsConfig = mock<AgentsConfig>({ tracingEnabled: false });
|
||||
const service = new AgentRunTracingService(agentsConfig);
|
||||
const service = new AgentRunTracingService(agentsConfig, otelService);
|
||||
|
||||
const built = await service.build(baseMetadata);
|
||||
|
||||
expect(built).toBeUndefined();
|
||||
expect(trace.getTracer).not.toHaveBeenCalled();
|
||||
expect(otelService.getTracer).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('builds telemetry from a fresh tracer fetch when tracing is enabled', async () => {
|
||||
const agentsConfig = mock<AgentsConfig>({ tracingEnabled: true });
|
||||
const service = new AgentRunTracingService(agentsConfig);
|
||||
const service = new AgentRunTracingService(agentsConfig, otelService);
|
||||
|
||||
await service.build(baseMetadata);
|
||||
await service.build(baseMetadata);
|
||||
|
||||
expect(trace.getTracer).toHaveBeenCalledTimes(2);
|
||||
expect(trace.getTracer).toHaveBeenCalledWith('@n8n/agents');
|
||||
expect(otelService.getTracer).toHaveBeenCalledTimes(2);
|
||||
expect(otelService.getTracer).toHaveBeenCalledWith('@n8n/agents');
|
||||
});
|
||||
|
||||
it('never marks the built telemetry as LangSmith, so root spans stay generic', async () => {
|
||||
const agentsConfig = mock<AgentsConfig>({ tracingEnabled: true });
|
||||
const service = new AgentRunTracingService(agentsConfig);
|
||||
const service = new AgentRunTracingService(agentsConfig, otelService);
|
||||
|
||||
const built = await service.build(baseMetadata);
|
||||
|
||||
|
|
@ -62,7 +64,7 @@ describe('AgentRunTracingService', () => {
|
|||
|
||||
it('includes execution_id/workflow_id/node_id only when source is "workflow"', async () => {
|
||||
const agentsConfig = mock<AgentsConfig>({ tracingEnabled: true });
|
||||
const service = new AgentRunTracingService(agentsConfig);
|
||||
const service = new AgentRunTracingService(agentsConfig, otelService);
|
||||
|
||||
const workflowBuilt = await service.build({
|
||||
...baseMetadata,
|
||||
|
|
@ -95,7 +97,7 @@ describe('AgentRunTracingService', () => {
|
|||
|
||||
it('omits execution_id/workflow_id/node_id individually when not provided, even for a workflow source', async () => {
|
||||
const agentsConfig = mock<AgentsConfig>({ tracingEnabled: true });
|
||||
const service = new AgentRunTracingService(agentsConfig);
|
||||
const service = new AgentRunTracingService(agentsConfig, otelService);
|
||||
|
||||
const noneProvided = await service.build({
|
||||
...baseMetadata,
|
||||
|
|
@ -150,7 +152,7 @@ describe('AgentRunTracingService', () => {
|
|||
|
||||
it('rejects workflow-only fields at compile time for a non-workflow source', async () => {
|
||||
const agentsConfig = mock<AgentsConfig>({ tracingEnabled: true });
|
||||
const service = new AgentRunTracingService(agentsConfig);
|
||||
const service = new AgentRunTracingService(agentsConfig, otelService);
|
||||
|
||||
// @ts-expect-error executionId only exists on the 'workflow' branch of
|
||||
// the AgentRunTracingMetadata union — a non-workflow source can't carry
|
||||
|
|
@ -160,7 +162,7 @@ describe('AgentRunTracingService', () => {
|
|||
|
||||
it('marks workflow-sourced runs as rootAnchored: false only when a parent context was found, and leaves other sources anchored', async () => {
|
||||
const agentsConfig = mock<AgentsConfig>({ tracingEnabled: true });
|
||||
const service = new AgentRunTracingService(agentsConfig);
|
||||
const service = new AgentRunTracingService(agentsConfig, otelService);
|
||||
|
||||
const workflowBuilt = await service.build({
|
||||
...baseMetadata,
|
||||
|
|
@ -184,7 +186,7 @@ describe('AgentRunTracingService', () => {
|
|||
|
||||
it('includes user_id and model_id only when provided', async () => {
|
||||
const agentsConfig = mock<AgentsConfig>({ tracingEnabled: true });
|
||||
const service = new AgentRunTracingService(agentsConfig);
|
||||
const service = new AgentRunTracingService(agentsConfig, otelService);
|
||||
|
||||
const built = await service.build({
|
||||
...baseMetadata,
|
||||
|
|
|
|||
|
|
@ -2,7 +2,8 @@ import type { AttributeValue, BuiltTelemetry } from '@n8n/agents';
|
|||
import { Telemetry } from '@n8n/agents';
|
||||
import { AgentsConfig } from '@n8n/config';
|
||||
import { Service } from '@n8n/di';
|
||||
import { trace } from '@opentelemetry/api';
|
||||
|
||||
import { OtelService } from '@/modules/otel/otel.service';
|
||||
|
||||
const AGENTS_TRACER_NAME = '@n8n/agents';
|
||||
|
||||
|
|
@ -49,13 +50,16 @@ function isWorkflowTracingMetadata(
|
|||
}
|
||||
|
||||
/**
|
||||
* Builds the per-run OTel telemetry handed to an agent run, riding along with
|
||||
* the workflow OTel module's already-registered global tracer provider — this
|
||||
* service never creates, registers, or shuts down a provider itself.
|
||||
* Builds the per-run OTel telemetry handed to an agent run on the workflow OTel
|
||||
* module's tracer provider — this service never creates, registers, or shuts
|
||||
* down a provider itself.
|
||||
*/
|
||||
@Service()
|
||||
export class AgentRunTracingService {
|
||||
constructor(private readonly agentsConfig: AgentsConfig) {}
|
||||
constructor(
|
||||
private readonly agentsConfig: AgentsConfig,
|
||||
private readonly otelService: OtelService,
|
||||
) {}
|
||||
|
||||
/** Whether agent tracing is enabled — lets callers skip tracing-only work (e.g. a DB lookup) upfront. */
|
||||
get enabled(): boolean {
|
||||
|
|
@ -67,9 +71,8 @@ export class AgentRunTracingService {
|
|||
* `ExecutionOptions.telemetry` entirely in that case, indistinguishable
|
||||
* from "OTel module off" to the SDK (both are no-ops).
|
||||
*
|
||||
* Fetches the tracer fresh on every call rather than caching it: a tracer
|
||||
* obtained before any provider is registered would otherwise go stale
|
||||
* across an OTel module restart. Fetching fresh sidesteps that entirely.
|
||||
* Fetches the tracer fresh on every call rather than caching it, so a
|
||||
* tracer never goes stale across an OTel module restart.
|
||||
*/
|
||||
async build(metadata: AgentRunTracingMetadata): Promise<BuiltTelemetry | undefined> {
|
||||
if (!this.agentsConfig.tracingEnabled) return undefined;
|
||||
|
|
@ -91,7 +94,7 @@ export class AgentRunTracingService {
|
|||
};
|
||||
|
||||
const built = await new Telemetry()
|
||||
.tracer(trace.getTracer(AGENTS_TRACER_NAME))
|
||||
.tracer(this.otelService.getTracer(AGENTS_TRACER_NAME))
|
||||
.metadata(attributes)
|
||||
.recordInputs(this.agentsConfig.tracingRecordInputs)
|
||||
.recordOutputs(this.agentsConfig.tracingRecordOutputs)
|
||||
|
|
|
|||
|
|
@ -12,72 +12,114 @@ Given OTEL often involves events triggered from elsewhere within the n8n system
|
|||
### Attributes
|
||||
All attributes are listed in `otel.constants.ts`
|
||||
|
||||
### Global API ownership
|
||||
|
||||
`OtelService` starts a `NodeTracerProvider` and hands out tracers through `getTracer()`.
|
||||
The module's own spans never depend on the global `@opentelemetry/api` registry.
|
||||
|
||||
At start, the service checks whether another library already registered a global tracer
|
||||
provider. Sentry does this in `ErrorReporter.init`. A user SDK loaded with `--require` does
|
||||
it too.
|
||||
|
||||
- The slot is free: the service registers its provider, an `AsyncLocalStorageContextManager`
|
||||
and the default W3C propagators. A restart swaps only the tracer provider. Shutdown leaves
|
||||
the globals in place.
|
||||
- The slot is taken: the service keeps its provider private and logs one info line. It never
|
||||
disables or replaces the other library's globals.
|
||||
|
||||
Consequences:
|
||||
|
||||
- Workflow spans go to the module exporter only. To get them into Sentry, point the OTel
|
||||
settings at Sentry's OTLP endpoint.
|
||||
- Workflow root spans start on `ROOT_CONTEXT`. They never join a trace that is active in the
|
||||
process, such as a Sentry request span or an HTTP server span from a user SDK. An inbound
|
||||
`traceparent` header still sets the parent.
|
||||
- Persisted trace context and outbound headers always use the W3C `traceparent` and
|
||||
`tracestate` format, whoever owns the global propagator. `OTEL_PROPAGATORS` has no effect
|
||||
on them.
|
||||
- The resource uses the env, process and host detectors. `OTEL_NODE_RESOURCE_DETECTORS` has
|
||||
no effect.
|
||||
|
||||
### Module architecture
|
||||
```mermaid
|
||||
graph TD
|
||||
subgraph Module Layer
|
||||
MOD["OtelModule
|
||||
@BackendModule"]
|
||||
end
|
||||
graph TD
|
||||
subgraph Module Layer
|
||||
MOD["OtelModule
|
||||
@BackendModule on main, worker and webhook"]
|
||||
end
|
||||
|
||||
subgraph Configuration
|
||||
CFG["OtelConfig
|
||||
env vars to typed config"]
|
||||
end
|
||||
subgraph Configuration
|
||||
CFG["OtelConfig
|
||||
env vars to typed config"]
|
||||
SET["OtelSettingsService
|
||||
DB row merged with config, env vars win"]
|
||||
CTRL["OtelSettingsController
|
||||
main only: GET and PUT settings, POST test trace"]
|
||||
end
|
||||
|
||||
subgraph SDK Layer
|
||||
SVC["OtelService
|
||||
owns NodeSDK lifecycle"]
|
||||
SDK["OpenTelemetry NodeSDK
|
||||
exporter, sampler, resource"]
|
||||
end
|
||||
subgraph SDK Layer
|
||||
SVC["OtelService
|
||||
owns the NodeTracerProvider, getTracer()"]
|
||||
SDK["NodeTracerProvider
|
||||
OTLP exporter, sampler, resource"]
|
||||
API["Global @opentelemetry/api
|
||||
registered only when the slot is free"]
|
||||
end
|
||||
|
||||
subgraph Instrumentation Layer
|
||||
INST["N8nWorkflowInstrumentation
|
||||
@OnLifecycleEvent listeners"]
|
||||
REG["SpanRegistry
|
||||
Map of executionId to Span"]
|
||||
end
|
||||
subgraph Instrumentation Layer
|
||||
LCH["OtelLifecycleHandler
|
||||
@OnLifecycleEvent and @OnPubSubEvent"]
|
||||
ELT["ExecutionLevelTracer
|
||||
span maps by execution id, W3C propagator"]
|
||||
TCS["TraceContextService
|
||||
traceparent on the execution row"]
|
||||
end
|
||||
|
||||
subgraph Handler Layer
|
||||
IFC{{"SpanHandler interface"}}
|
||||
WS["WorkflowStartHandler"]
|
||||
WE["WorkflowEndHandler"]
|
||||
end
|
||||
subgraph n8n Core
|
||||
LC(("Lifecycle events
|
||||
workflowExecuteBefore, Resume, After
|
||||
nodeExecuteBefore, After"))
|
||||
PUB(("Pub/Sub
|
||||
reload-otel-config"))
|
||||
REQ["Request helpers
|
||||
module context injectTraceHeaders"]
|
||||
AGT["AgentRunTracingService
|
||||
AgentWorkflowExecutionService"]
|
||||
end
|
||||
|
||||
subgraph n8n Core
|
||||
LC(("Lifecycle Events
|
||||
workflowExecuteBefore
|
||||
workflowExecuteAfter"))
|
||||
end
|
||||
COL[("OTLP collector")]
|
||||
|
||||
MOD -- "1. check enabled" --> CFG
|
||||
MOD -- "2. init SDK" --> SVC
|
||||
SVC -- "creates" --> SDK
|
||||
MOD -- "3. register listeners" --> INST
|
||||
MOD -- "1. load settings" --> SET
|
||||
SET --> CFG
|
||||
MOD -- "2. start provider" --> SVC
|
||||
SVC -- "creates" --> SDK
|
||||
SVC -. "registers when free" .-> API
|
||||
MOD -- "3. import handler" --> LCH
|
||||
MOD -- "4. main only" --> CTRL
|
||||
CTRL -- "save, then publish" --> PUB
|
||||
PUB -. "fires" .-> LCH
|
||||
LCH -- "restart()" --> SVC
|
||||
|
||||
LC -. "fires event" .-> INST
|
||||
INST -- "dispatches to" --> IFC
|
||||
IFC -. "implemented by" .-> WS
|
||||
IFC -. "implemented by" .-> WE
|
||||
LC -. "fires" .-> LCH
|
||||
LCH -- "start and end workflow and node spans" --> ELT
|
||||
LCH -- "persist and read" --> TCS
|
||||
ELT -- "getTracer()" --> SVC
|
||||
REQ -- "outbound traceparent" --> ELT
|
||||
AGT -- "getActiveContext()" --> ELT
|
||||
AGT -- "getTracer()" --> SVC
|
||||
SDK -. "exports spans" .-> COL
|
||||
|
||||
WS -- "startSpan, store" --> REG
|
||||
WE -- "retrieve, enrich, end" --> REG
|
||||
REG -. "spans exported via" .-> SDK
|
||||
classDef module fill:#4a9eff,color:#fff
|
||||
classDef config fill:#f5a623,color:#fff
|
||||
classDef sdk fill:#7b68ee,color:#fff
|
||||
classDef inst fill:#50c878,color:#fff
|
||||
classDef core fill:#888,color:#fff
|
||||
|
||||
classDef module fill:#4a9eff,color:#fff
|
||||
classDef config fill:#f5a623,color:#fff
|
||||
classDef sdk fill:#7b68ee,color:#fff
|
||||
classDef inst fill:#50c878,color:#fff
|
||||
classDef handler fill:#ff6b6b,color:#fff
|
||||
classDef core fill:#888,color:#fff
|
||||
|
||||
class MOD module
|
||||
class CFG config
|
||||
class SVC,SDK sdk
|
||||
class INST,REG inst
|
||||
class IFC,WS,WE handler
|
||||
class LC core
|
||||
class MOD module
|
||||
class CFG,SET,CTRL config
|
||||
class SVC,SDK,API sdk
|
||||
class LCH,ELT,TCS inst
|
||||
class LC,PUB,REQ,AGT,COL core
|
||||
```
|
||||
|
||||
#### Manual validation
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
import type { Logger } from '@n8n/backend-common';
|
||||
import { SpanStatusCode, trace } from '@opentelemetry/api';
|
||||
import type { TextMapPropagator } from '@opentelemetry/api';
|
||||
import { context, propagation, SpanStatusCode, trace } from '@opentelemetry/api';
|
||||
import { hrTimeToMilliseconds } from '@opentelemetry/core';
|
||||
import { mock } from 'vitest-mock-extended';
|
||||
|
||||
|
|
@ -21,7 +22,7 @@ describe('ExecutionLevelTracer', () => {
|
|||
};
|
||||
|
||||
beforeAll(() => {
|
||||
otel = OtelTestProvider.create();
|
||||
otel = OtelTestProvider.create({ withContextManager: true });
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
|
|
@ -30,7 +31,7 @@ describe('ExecutionLevelTracer', () => {
|
|||
|
||||
beforeEach(() => {
|
||||
otel.reset();
|
||||
tracer = new ExecutionLevelTracer(makeOtelSettingsService(), logger);
|
||||
tracer = new ExecutionLevelTracer(otel.asOtelService(), makeOtelSettingsService(), logger);
|
||||
});
|
||||
|
||||
const inboundTracingContext = {
|
||||
|
|
@ -901,6 +902,7 @@ describe('ExecutionLevelTracer', () => {
|
|||
|
||||
it('should no-op when injectOutbound is false', () => {
|
||||
const noInjectTracer = new ExecutionLevelTracer(
|
||||
otel.asOtelService(),
|
||||
makeOtelSettingsService({ injectOutbound: false }),
|
||||
logger,
|
||||
);
|
||||
|
|
@ -947,6 +949,59 @@ describe('ExecutionLevelTracer', () => {
|
|||
});
|
||||
});
|
||||
|
||||
describe('trace context ownership', () => {
|
||||
const traceparentPattern = /^00-[0-9a-f]{32}-[0-9a-f]{16}-01$/;
|
||||
|
||||
it('should start a new trace when no tracing context is given, even inside an active foreign span', () => {
|
||||
const foreignSpan = trace.getTracer('foreign').startSpan('GET /webhook');
|
||||
|
||||
context.with(trace.setSpan(context.active(), foreignSpan), () => {
|
||||
tracer.startWorkflow({ executionId: 'exec-root', workflow: defaultWorkflow });
|
||||
});
|
||||
tracer.endWorkflow({
|
||||
executionId: 'exec-root',
|
||||
status: 'success',
|
||||
mode: 'webhook',
|
||||
isRetry: false,
|
||||
});
|
||||
foreignSpan.end();
|
||||
|
||||
const workflowSpan = otel.getFinishedSpans().find((s) => s.name === 'workflow.execute')!;
|
||||
expect(workflowSpan.parentSpanContext).toBeUndefined();
|
||||
expect(workflowSpan.spanContext().traceId).not.toBe(foreignSpan.spanContext().traceId);
|
||||
});
|
||||
|
||||
it('should emit W3C trace context while another library owns the global propagator', () => {
|
||||
const foreignPropagator: TextMapPropagator = {
|
||||
inject: (_ctx, carrier, setter) => setter.set(carrier, 'sentry-trace', 'foreign'),
|
||||
extract: (ctx) => ctx,
|
||||
fields: () => ['sentry-trace'],
|
||||
};
|
||||
propagation.setGlobalPropagator(foreignPropagator);
|
||||
|
||||
try {
|
||||
const persisted = tracer.startWorkflow({
|
||||
executionId: 'exec-w3c',
|
||||
workflow: defaultWorkflow,
|
||||
});
|
||||
const headers: Record<string, string> = {};
|
||||
tracer.injectTraceHeaders('exec-w3c', undefined, headers);
|
||||
tracer.endWorkflow({
|
||||
executionId: 'exec-w3c',
|
||||
status: 'success',
|
||||
mode: 'webhook',
|
||||
isRetry: false,
|
||||
});
|
||||
|
||||
expect(persisted.traceparent).toMatch(traceparentPattern);
|
||||
expect(headers.traceparent).toMatch(traceparentPattern);
|
||||
expect(headers['sentry-trace']).toBeUndefined();
|
||||
} finally {
|
||||
propagation.disable();
|
||||
}
|
||||
});
|
||||
});
|
||||
|
||||
describe('getActiveContext', () => {
|
||||
it('returns a context carrying the node span when one is active', () => {
|
||||
tracer.startWorkflow({ executionId: 'exec-ctx-node', workflow: defaultWorkflow });
|
||||
|
|
|
|||
|
|
@ -1170,14 +1170,13 @@ describe('onReloadOtelConfig', () => {
|
|||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
it('restarts the OTel SDK and refreshes the tracer', async () => {
|
||||
it('restarts the OTel tracer provider', async () => {
|
||||
const handler = makeHandler();
|
||||
otelService.restart.mockResolvedValue(undefined);
|
||||
|
||||
await handler.onReloadOtelConfig();
|
||||
|
||||
expect(otelService.restart).toHaveBeenCalledTimes(1);
|
||||
expect(tracer.refreshTracer).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,225 @@
|
|||
import type { Logger } from '@n8n/backend-common';
|
||||
import type { OutboundHttp } from '@n8n/backend-network';
|
||||
import type { Context } from '@opentelemetry/api';
|
||||
import {
|
||||
context,
|
||||
createContextKey,
|
||||
propagation,
|
||||
ProxyTracerProvider,
|
||||
ROOT_CONTEXT,
|
||||
trace,
|
||||
} from '@opentelemetry/api';
|
||||
import type { ReadableSpan } from '@opentelemetry/sdk-trace-base';
|
||||
import {
|
||||
InMemorySpanExporter,
|
||||
NodeTracerProvider,
|
||||
SimpleSpanProcessor,
|
||||
} from '@opentelemetry/sdk-trace-node';
|
||||
import { mock } from 'vitest-mock-extended';
|
||||
import type { InstanceSettings } from 'n8n-core';
|
||||
|
||||
import type { OtelSettingsService } from '../otel-settings.service';
|
||||
import type { OtelConfig } from '../otel.config';
|
||||
import { ATTR } from '../otel.constants';
|
||||
import { OtelService } from '../otel.service';
|
||||
|
||||
import { N8N_VERSION } from '@/constants';
|
||||
|
||||
const { exportedSpans } = vi.hoisted(() => ({ exportedSpans: [] as ReadableSpan[] }));
|
||||
|
||||
vi.mock('@opentelemetry/exporter-trace-otlp-proto', () => ({
|
||||
OTLPTraceExporter: vi.fn().mockImplementation(function () {
|
||||
return {
|
||||
export: (spans: ReadableSpan[], resultCallback: (result: { code: number }) => void) => {
|
||||
exportedSpans.push(...spans);
|
||||
resultCallback({ code: 0 });
|
||||
},
|
||||
shutdown: async () => {},
|
||||
forceFlush: async () => {},
|
||||
};
|
||||
}),
|
||||
}));
|
||||
|
||||
vi.mock('@opentelemetry/sdk-trace-base', async (importActual) => {
|
||||
const actual = await importActual<typeof import('@opentelemetry/sdk-trace-base')>();
|
||||
return { ...actual, BatchSpanProcessor: actual.SimpleSpanProcessor };
|
||||
});
|
||||
|
||||
vi.mock('@opentelemetry/resources', async (importActual) => {
|
||||
const actual = await importActual<typeof import('@opentelemetry/resources')>();
|
||||
return { ...actual, detectResources: () => actual.emptyResource() };
|
||||
});
|
||||
|
||||
const enabledSettings: OtelConfig = {
|
||||
enabled: true,
|
||||
exporterProtocol: 'http/protobuf',
|
||||
exporterEndpoint: 'http://localhost:4318',
|
||||
exporterTracingPath: '/v1/traces',
|
||||
exporterHeaders: '',
|
||||
exporterServiceName: 'n8n',
|
||||
tracesSampleRate: 1,
|
||||
startupConnectivityTimeoutMs: 2_000,
|
||||
includeNodeSpans: true,
|
||||
injectOutbound: true,
|
||||
productionExecutionsOnly: true,
|
||||
};
|
||||
|
||||
const disabledSettings: OtelConfig = { ...enabledSettings, enabled: false };
|
||||
|
||||
const traceparentPattern = /^00-[0-9a-f]{32}-[0-9a-f]{16}-01$/;
|
||||
const markedContext = ROOT_CONTEXT.setValue(createContextKey('marker'), 'marked');
|
||||
|
||||
function activeContextInside(ctx: Context): Context {
|
||||
return context.with(ctx, () => context.active());
|
||||
}
|
||||
|
||||
function globalTracerProviderDelegate() {
|
||||
const globalProvider = trace.getTracerProvider();
|
||||
return globalProvider instanceof ProxyTracerProvider
|
||||
? globalProvider.getDelegate()
|
||||
: globalProvider;
|
||||
}
|
||||
|
||||
function registerForeignProvider() {
|
||||
const exporter = new InMemorySpanExporter();
|
||||
const provider = new NodeTracerProvider({ spanProcessors: [new SimpleSpanProcessor(exporter)] });
|
||||
provider.register();
|
||||
return { provider, exporter };
|
||||
}
|
||||
|
||||
function exportedSpanNames() {
|
||||
return exportedSpans.map((span) => span.name);
|
||||
}
|
||||
|
||||
describe('OtelService tracer provider', () => {
|
||||
const fetchMock = vi.fn();
|
||||
const outboundHttp = {
|
||||
transport: () => ({ asCustomFetch: () => fetchMock }),
|
||||
} as unknown as OutboundHttp;
|
||||
let otelSettingsService: ReturnType<typeof mock<OtelSettingsService>>;
|
||||
let logger: ReturnType<typeof mock<Logger>>;
|
||||
let service: OtelService;
|
||||
let foreign: ReturnType<typeof registerForeignProvider> | undefined;
|
||||
|
||||
beforeEach(() => {
|
||||
exportedSpans.length = 0;
|
||||
fetchMock.mockResolvedValue({ ok: true });
|
||||
otelSettingsService = mock<OtelSettingsService>();
|
||||
otelSettingsService.loadSettings.mockResolvedValue(enabledSettings);
|
||||
logger = mock<Logger>();
|
||||
service = new OtelService(
|
||||
otelSettingsService,
|
||||
mock<InstanceSettings>({ instanceId: 'inst-1', instanceType: 'main' }),
|
||||
logger,
|
||||
outboundHttp,
|
||||
);
|
||||
});
|
||||
|
||||
afterEach(async () => {
|
||||
await service.shutdown();
|
||||
await foreign?.provider.shutdown();
|
||||
foreign = undefined;
|
||||
trace.disable();
|
||||
context.disable();
|
||||
propagation.disable();
|
||||
});
|
||||
|
||||
describe('when another library owns the global API', () => {
|
||||
beforeEach(() => {
|
||||
foreign = registerForeignProvider();
|
||||
});
|
||||
|
||||
it('exports n8n spans through its own provider and leaves the foreign exporter empty', async () => {
|
||||
await service.init();
|
||||
|
||||
service.getTracer('n8n-workflow').startSpan('workflow.execute').end();
|
||||
|
||||
expect(exportedSpanNames()).toEqual(['workflow.execute']);
|
||||
expect(exportedSpans[0].resource.attributes).toMatchObject({
|
||||
[ATTR.OTEL_SERVICE_NAME]: 'n8n',
|
||||
[ATTR.OTEL_SERVICE_VERSION]: N8N_VERSION,
|
||||
[ATTR.INSTANCE_ID]: 'inst-1',
|
||||
[ATTR.INSTANCE_ROLE]: 'main',
|
||||
});
|
||||
expect(foreign?.exporter.getFinishedSpans()).toHaveLength(0);
|
||||
});
|
||||
|
||||
it('keeps exporting after a restart and leaves the foreign globals in place', async () => {
|
||||
await service.init();
|
||||
await service.restart();
|
||||
|
||||
service.getTracer('n8n-workflow').startSpan('workflow.execute').end();
|
||||
trace.getTracer('foreign').startSpan('GET /webhook').end();
|
||||
|
||||
expect(exportedSpanNames()).toEqual(['workflow.execute']);
|
||||
expect(foreign?.exporter.getFinishedSpans().map((span) => span.name)).toEqual([
|
||||
'GET /webhook',
|
||||
]);
|
||||
expect(globalTracerProviderDelegate()).toBe(foreign?.provider);
|
||||
expect(activeContextInside(markedContext)).toBe(markedContext);
|
||||
});
|
||||
|
||||
it('logs the foreign owner once for the lifetime of the service', async () => {
|
||||
await service.init();
|
||||
await service.restart();
|
||||
|
||||
expect(logger.info).toHaveBeenCalledTimes(1);
|
||||
expect(logger.info).toHaveBeenCalledWith(
|
||||
expect.stringContaining('runs on its own tracer provider'),
|
||||
);
|
||||
});
|
||||
});
|
||||
|
||||
describe('when the global API is free', () => {
|
||||
it('registers its provider, so third-party tracers export through the module', async () => {
|
||||
await service.init();
|
||||
|
||||
trace.getTracer('third-party').startSpan('http.request').end();
|
||||
const headers: Record<string, string> = {};
|
||||
const span = service.getTracer('n8n-workflow').startSpan('workflow.execute');
|
||||
propagation.inject(trace.setSpan(context.active(), span), headers);
|
||||
span.end();
|
||||
|
||||
expect(exportedSpanNames()).toEqual(['http.request', 'workflow.execute']);
|
||||
expect(headers.traceparent).toMatch(traceparentPattern);
|
||||
expect(activeContextInside(markedContext)).toBe(markedContext);
|
||||
expect(logger.info).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('swaps the registered provider on restart and keeps the context manager', async () => {
|
||||
await service.init();
|
||||
await service.restart();
|
||||
|
||||
trace.getTracer('third-party').startSpan('http.request').end();
|
||||
|
||||
expect(exportedSpanNames()).toEqual(['http.request']);
|
||||
expect(activeContextInside(markedContext)).toBe(markedContext);
|
||||
});
|
||||
|
||||
it('keeps the context manager after shutdown and hands out no-op tracers', async () => {
|
||||
await service.init();
|
||||
await service.shutdown();
|
||||
|
||||
expect(activeContextInside(markedContext)).toBe(markedContext);
|
||||
expect(service.getTracer('n8n-workflow').startSpan('workflow.execute').isRecording()).toBe(
|
||||
false,
|
||||
);
|
||||
});
|
||||
|
||||
it('registers on the first enabled restart after a disabled boot', async () => {
|
||||
otelSettingsService.loadSettings
|
||||
.mockResolvedValueOnce(disabledSettings)
|
||||
.mockResolvedValueOnce(enabledSettings);
|
||||
await service.init();
|
||||
expect(service.getTracer('n8n-workflow').startSpan('workflow.execute').isRecording()).toBe(
|
||||
false,
|
||||
);
|
||||
|
||||
await service.restart();
|
||||
service.getTracer('n8n-workflow').startSpan('workflow.execute').end();
|
||||
trace.getTracer('third-party').startSpan('http.request').end();
|
||||
|
||||
expect(exportedSpanNames()).toEqual(['workflow.execute', 'http.request']);
|
||||
});
|
||||
});
|
||||
});
|
||||
|
|
@ -1,9 +1,10 @@
|
|||
import type { Logger } from '@n8n/backend-common';
|
||||
import type { OutboundHttp } from '@n8n/backend-network';
|
||||
import { context, diag, metrics, propagation, trace } from '@opentelemetry/api';
|
||||
import { diag } from '@opentelemetry/api';
|
||||
import { OTLPTraceExporter as OTLPGrpcTraceExporter } from '@opentelemetry/exporter-trace-otlp-grpc';
|
||||
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-proto';
|
||||
import { BasicTracerProvider } from '@opentelemetry/sdk-trace-base';
|
||||
import { NodeTracerProvider } from '@opentelemetry/sdk-trace-node';
|
||||
import { mock } from 'vitest-mock-extended';
|
||||
import type { InstanceSettings } from 'n8n-core';
|
||||
|
||||
|
|
@ -12,8 +13,9 @@ import type { OtelConfig } from '../otel.config';
|
|||
import { ATTR, OTEL_TEST_SPAN_NAME } from '../otel.constants';
|
||||
import { OtelService } from '../otel.service';
|
||||
|
||||
const start = vi.fn();
|
||||
const register = vi.fn();
|
||||
const shutdown = vi.fn();
|
||||
const providerGetTracer = vi.fn();
|
||||
|
||||
// The connectivity-check fetch, obtained via outboundHttp.transport().asCustomFetch().
|
||||
// checkEndpointReachability ignores the response and only catches network errors.
|
||||
|
|
@ -86,15 +88,6 @@ const mockSpanEnd = vi.fn();
|
|||
const mockStartSpan = vi.fn();
|
||||
const mockGetTracer = vi.fn();
|
||||
|
||||
vi.mock('@opentelemetry/sdk-node', () => ({
|
||||
NodeSDK: vi.fn().mockImplementation(function () {
|
||||
return {
|
||||
start,
|
||||
shutdown,
|
||||
};
|
||||
}),
|
||||
}));
|
||||
|
||||
vi.mock('@opentelemetry/exporter-trace-otlp-proto', () => ({
|
||||
OTLPTraceExporter: vi.fn().mockImplementation(function () {
|
||||
return {
|
||||
|
|
@ -122,6 +115,7 @@ vi.mock('@grpc/grpc-js', () => ({
|
|||
}));
|
||||
|
||||
vi.mock('@opentelemetry/sdk-trace-base', () => ({
|
||||
BatchSpanProcessor: vi.fn(),
|
||||
BasicTracerProvider: vi.fn().mockImplementation(function (config: {
|
||||
spanProcessors?: unknown[];
|
||||
}) {
|
||||
|
|
@ -136,21 +130,24 @@ vi.mock('@opentelemetry/sdk-trace-base', () => ({
|
|||
}));
|
||||
|
||||
vi.mock('@opentelemetry/resources', () => ({
|
||||
resourceFromAttributes: vi.fn().mockReturnValue({}),
|
||||
resourceFromAttributes: vi.fn().mockReturnValue({ merge: () => ({}) }),
|
||||
detectResources: vi.fn().mockReturnValue({}),
|
||||
envDetector: {},
|
||||
processDetector: {},
|
||||
hostDetector: {},
|
||||
}));
|
||||
|
||||
vi.mock('@opentelemetry/sdk-trace-node', () => ({
|
||||
TraceIdRatioBasedSampler: vi.fn().mockImplementation(function () {
|
||||
return {};
|
||||
}),
|
||||
NodeTracerProvider: vi.fn().mockImplementation(function () {
|
||||
return { register, shutdown, getTracer: providerGetTracer };
|
||||
}),
|
||||
}));
|
||||
|
||||
vi.mock('@opentelemetry/api', async () => ({
|
||||
...(await vi.importActual<typeof import('@opentelemetry/api')>('@opentelemetry/api')),
|
||||
trace: { disable: vi.fn() },
|
||||
context: { disable: vi.fn() },
|
||||
propagation: { disable: vi.fn() },
|
||||
metrics: { disable: vi.fn() },
|
||||
DiagLogLevel: { WARN: 'WARN' },
|
||||
diag: { setLogger: vi.fn() },
|
||||
}));
|
||||
|
|
@ -201,20 +198,20 @@ describe('OtelService', () => {
|
|||
});
|
||||
|
||||
describe('init', () => {
|
||||
it('does not start SDK when enabled is false', async () => {
|
||||
it('does not start a tracer provider when enabled is false', async () => {
|
||||
otelSettingsService.loadSettings.mockResolvedValue(disabledSettings);
|
||||
|
||||
await service.init();
|
||||
|
||||
expect(start).not.toHaveBeenCalled();
|
||||
expect(NodeTracerProvider).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it('starts SDK when enabled is true', async () => {
|
||||
it('starts a tracer provider when enabled is true', async () => {
|
||||
otelSettingsService.loadSettings.mockResolvedValue(enabledSettings);
|
||||
|
||||
await service.init();
|
||||
|
||||
expect(start).toHaveBeenCalledTimes(1);
|
||||
expect(NodeTracerProvider).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('builds the HTTP exporter with the traces path and no gRPC exporter', async () => {
|
||||
|
|
@ -241,7 +238,7 @@ describe('OtelService', () => {
|
|||
await service.init();
|
||||
await flushPromises();
|
||||
|
||||
expect(start).toHaveBeenCalledTimes(1);
|
||||
expect(NodeTracerProvider).toHaveBeenCalledTimes(1);
|
||||
expect(logger.error).toHaveBeenCalledWith(
|
||||
'Failed to connect to OpenTelemetry OTLP endpoint during startup',
|
||||
expect.objectContaining({ endpoint: 'http://localhost:4318/v1/traces' }),
|
||||
|
|
@ -259,7 +256,7 @@ describe('OtelService', () => {
|
|||
expect.objectContaining({ url: 'http://collector.example.com:4317' }),
|
||||
);
|
||||
expect(OTLPTraceExporter).not.toHaveBeenCalled();
|
||||
expect(start).toHaveBeenCalledTimes(1);
|
||||
expect(NodeTracerProvider).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('lowercases only the endpoint scheme before it reaches the exporter', async () => {
|
||||
|
|
@ -408,7 +405,7 @@ describe('OtelService', () => {
|
|||
await service.init();
|
||||
await flushPromises();
|
||||
|
||||
expect(start).toHaveBeenCalledTimes(1);
|
||||
expect(NodeTracerProvider).toHaveBeenCalledTimes(1);
|
||||
expect(logger.error).toHaveBeenCalledTimes(1);
|
||||
expect(logger.error).toHaveBeenCalledWith(
|
||||
'Failed to connect to OpenTelemetry OTLP endpoint during startup',
|
||||
|
|
@ -427,7 +424,7 @@ describe('OtelService', () => {
|
|||
await flushPromises();
|
||||
|
||||
expect(newClient).not.toHaveBeenCalled();
|
||||
expect(start).toHaveBeenCalledTimes(1);
|
||||
expect(NodeTracerProvider).toHaveBeenCalledTimes(1);
|
||||
expect(logger.error).toHaveBeenCalledWith(
|
||||
'Failed to connect to OpenTelemetry OTLP endpoint during startup',
|
||||
expect.objectContaining({ endpoint: 'not a valid endpoint' }),
|
||||
|
|
@ -436,7 +433,7 @@ describe('OtelService', () => {
|
|||
});
|
||||
|
||||
describe('restart', () => {
|
||||
it('shuts down existing SDK then reloads settings and starts a new one', async () => {
|
||||
it('shuts down the existing tracer provider then reloads settings and starts a new one', async () => {
|
||||
otelSettingsService.loadSettings.mockResolvedValue(enabledSettings);
|
||||
|
||||
await service.init();
|
||||
|
|
@ -447,10 +444,10 @@ describe('OtelService', () => {
|
|||
|
||||
expect(shutdown).toHaveBeenCalledTimes(1);
|
||||
expect(otelSettingsService.loadSettings).toHaveBeenCalledTimes(1);
|
||||
expect(start).toHaveBeenCalledTimes(1);
|
||||
expect(NodeTracerProvider).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('does not start SDK after restart when reloaded settings have enabled=false', async () => {
|
||||
it('does not start a tracer provider after restart when reloaded settings have enabled=false', async () => {
|
||||
otelSettingsService.loadSettings.mockResolvedValueOnce(enabledSettings);
|
||||
otelSettingsService.loadSettings.mockResolvedValueOnce(disabledSettings);
|
||||
|
||||
|
|
@ -459,11 +456,11 @@ describe('OtelService', () => {
|
|||
|
||||
await service.restart();
|
||||
|
||||
expect(start).not.toHaveBeenCalled();
|
||||
expect(NodeTracerProvider).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe('SDK startup failure', () => {
|
||||
describe('tracer provider startup failure', () => {
|
||||
const exporterFailure = () => {
|
||||
vi.mocked(OTLPTraceExporter).mockImplementationOnce(function () {
|
||||
throw new Error('exporter unavailable');
|
||||
|
|
@ -476,7 +473,7 @@ describe('OtelService', () => {
|
|||
|
||||
await expect(service.init()).resolves.not.toThrow();
|
||||
|
||||
expect(start).not.toHaveBeenCalled();
|
||||
expect(NodeTracerProvider).not.toHaveBeenCalled();
|
||||
expect(fetchMock).not.toHaveBeenCalled();
|
||||
expect(logger.error).toHaveBeenCalledTimes(1);
|
||||
expect(logger.error).toHaveBeenCalledWith(
|
||||
|
|
@ -485,9 +482,9 @@ describe('OtelService', () => {
|
|||
);
|
||||
});
|
||||
|
||||
it('tears down partially installed providers when the SDK start call throws', async () => {
|
||||
it('shuts the provider down when registering it with the global API throws', async () => {
|
||||
otelSettingsService.loadSettings.mockResolvedValue(enabledSettings);
|
||||
start.mockImplementationOnce(() => {
|
||||
register.mockImplementationOnce(() => {
|
||||
throw new Error('provider registration failed');
|
||||
});
|
||||
|
||||
|
|
@ -498,7 +495,6 @@ describe('OtelService', () => {
|
|||
{ error: 'provider registration failed' },
|
||||
);
|
||||
expect(shutdown).toHaveBeenCalledTimes(1);
|
||||
expect(trace.disable).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('leaves the service in a non-exporting state when a restart fails', async () => {
|
||||
|
|
@ -509,7 +505,7 @@ describe('OtelService', () => {
|
|||
|
||||
await expect(service.restart()).resolves.not.toThrow();
|
||||
|
||||
expect(start).not.toHaveBeenCalled();
|
||||
expect(NodeTracerProvider).not.toHaveBeenCalled();
|
||||
expect(shutdown).toHaveBeenCalledTimes(1);
|
||||
|
||||
shutdown.mockClear();
|
||||
|
|
@ -520,23 +516,20 @@ describe('OtelService', () => {
|
|||
});
|
||||
|
||||
describe('shutdown', () => {
|
||||
it('disables all four OTel globals so the next SDK start can re-register providers', async () => {
|
||||
it('shuts the running provider down', async () => {
|
||||
otelSettingsService.loadSettings.mockResolvedValue(enabledSettings);
|
||||
await service.init();
|
||||
|
||||
await service.shutdown();
|
||||
|
||||
expect(trace.disable).toHaveBeenCalledTimes(1);
|
||||
expect(context.disable).toHaveBeenCalledTimes(1);
|
||||
expect(propagation.disable).toHaveBeenCalledTimes(1);
|
||||
expect(metrics.disable).toHaveBeenCalledTimes(1);
|
||||
expect(shutdown).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it('does not throw when called before init', async () => {
|
||||
await expect(service.shutdown()).resolves.not.toThrow();
|
||||
});
|
||||
|
||||
it('does not throw when the SDK shutdown rejects (e.g. exporter flush failure)', async () => {
|
||||
it('does not throw when the provider shutdown rejects (e.g. exporter flush failure)', async () => {
|
||||
otelSettingsService.loadSettings.mockResolvedValue(enabledSettings);
|
||||
await service.init();
|
||||
shutdown.mockRejectedValueOnce(new Error('connect ECONNREFUSED 127.0.0.1:9'));
|
||||
|
|
@ -545,6 +538,24 @@ describe('OtelService', () => {
|
|||
});
|
||||
});
|
||||
|
||||
describe('getTracer', () => {
|
||||
it('hands out a no-op tracer before a provider has started', () => {
|
||||
const span = service.getTracer('n8n-workflow').startSpan('workflow.execute');
|
||||
|
||||
expect(span.isRecording()).toBe(false);
|
||||
});
|
||||
|
||||
it("hands out the running provider's tracer", async () => {
|
||||
const tracer = { startSpan: vi.fn() };
|
||||
providerGetTracer.mockReturnValue(tracer);
|
||||
otelSettingsService.loadSettings.mockResolvedValue(enabledSettings);
|
||||
await service.init();
|
||||
|
||||
expect(service.getTracer('n8n-workflow')).toBe(tracer);
|
||||
expect(providerGetTracer).toHaveBeenCalledWith('n8n-workflow');
|
||||
});
|
||||
});
|
||||
|
||||
describe('diagnostics logger', () => {
|
||||
beforeEach(() => {
|
||||
(
|
||||
|
|
@ -806,7 +817,7 @@ describe('OtelService', () => {
|
|||
await service.sendTestTrace(connection);
|
||||
|
||||
expect(BasicTracerProvider).toHaveBeenCalledTimes(1);
|
||||
expect(start).not.toHaveBeenCalled();
|
||||
expect(NodeTracerProvider).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
describe('over gRPC', () => {
|
||||
|
|
|
|||
|
|
@ -23,7 +23,9 @@ import * as utils from '@test-integration/utils';
|
|||
|
||||
import { OtelTestProvider } from './otel-test-provider';
|
||||
import { TestNodeWithTracing } from './test-node-with-tracing';
|
||||
import { OtelSettingsService } from '../../otel-settings.service';
|
||||
import { OtelConfig } from '../../otel.config';
|
||||
import { OtelService } from '../../otel.service';
|
||||
|
||||
const BASE_DIR = path.resolve(__dirname, '../../../../../..');
|
||||
|
||||
|
|
@ -51,6 +53,8 @@ export async function initOtelTestEnvironment() {
|
|||
|
||||
await testModules.loadModules(['otel']);
|
||||
await testDb.init();
|
||||
Container.set(OtelService, otel.asOtelService());
|
||||
await Container.get(OtelSettingsService).loadSettings();
|
||||
await Container.get(ModuleRegistry).initModules('main');
|
||||
Container.get(LicenseState).setLicenseProvider({
|
||||
isLicensed: (feature) => feature === LICENSE_FEATURES.OTEL_CUSTOM_SPAN_ATTRIBUTES,
|
||||
|
|
|
|||
|
|
@ -1,17 +1,19 @@
|
|||
import type { Context, ContextManager } from '@opentelemetry/api';
|
||||
import { context, propagation, ROOT_CONTEXT, trace } from '@opentelemetry/api';
|
||||
import { W3CTraceContextPropagator } from '@opentelemetry/core';
|
||||
import { context, ROOT_CONTEXT, trace } from '@opentelemetry/api';
|
||||
import {
|
||||
BasicTracerProvider,
|
||||
InMemorySpanExporter,
|
||||
NodeTracerProvider,
|
||||
SimpleSpanProcessor,
|
||||
} from '@opentelemetry/sdk-trace-node';
|
||||
import { AsyncLocalStorage } from 'node:async_hooks';
|
||||
import { mock } from 'vitest-mock-extended';
|
||||
|
||||
import type { OtelService } from '../../otel.service';
|
||||
|
||||
/**
|
||||
* Minimal stand-in for `@opentelemetry/context-async-hooks`'
|
||||
* `AsyncLocalStorageContextManager` (the one `NodeSDK` registers by default in
|
||||
* production) — avoids adding that package as a test-only dependency. Without
|
||||
* `AsyncLocalStorageContextManager` (the one `NodeTracerProvider.register()`
|
||||
* installs in production) — avoids adding that package as a test-only dependency. Without
|
||||
* a real context manager registered, `context.with()` is a no-op (the default
|
||||
* `NoopContextManager` ignores the context it's given), so span nesting via
|
||||
* ambient context can't be exercised at all.
|
||||
|
|
@ -46,11 +48,14 @@ class AsyncLocalStorageTestContextManager implements ContextManager {
|
|||
}
|
||||
|
||||
/**
|
||||
* Disposable OTel test harness. Sets up an in-memory tracer
|
||||
* that captures spans for assertion, and tears down cleanly.
|
||||
* Disposable OTel test harness. Sets up an in-memory tracer provider that
|
||||
* captures spans for assertion, registers it as the global tracer provider
|
||||
* (so the module under test sees the slot taken, like it does next to
|
||||
* Sentry), and tears down cleanly.
|
||||
*
|
||||
* Usage:
|
||||
* const otel = OtelTestProvider.create();
|
||||
* const tracer = new ExecutionLevelTracer(otel.asOtelService(), ...);
|
||||
* // ... run code that creates spans ...
|
||||
* expect(otel.getFinishedSpans()).toHaveLength(1);
|
||||
* otel.reset(); // between tests
|
||||
|
|
@ -62,24 +67,28 @@ class AsyncLocalStorageTestContextManager implements ContextManager {
|
|||
*/
|
||||
export class OtelTestProvider {
|
||||
private constructor(
|
||||
private readonly provider: BasicTracerProvider,
|
||||
readonly provider: NodeTracerProvider,
|
||||
private readonly exporter: InMemorySpanExporter,
|
||||
private readonly contextManagerEnabled: boolean,
|
||||
) {}
|
||||
|
||||
static create(options: { withContextManager?: boolean } = {}): OtelTestProvider {
|
||||
const exporter = new InMemorySpanExporter();
|
||||
const provider = new BasicTracerProvider({
|
||||
const provider = new NodeTracerProvider({
|
||||
spanProcessors: [new SimpleSpanProcessor(exporter)],
|
||||
});
|
||||
trace.setGlobalTracerProvider(provider);
|
||||
propagation.setGlobalPropagator(new W3CTraceContextPropagator());
|
||||
if (options.withContextManager) {
|
||||
context.setGlobalContextManager(new AsyncLocalStorageTestContextManager());
|
||||
}
|
||||
return new OtelTestProvider(provider, exporter, options.withContextManager ?? false);
|
||||
}
|
||||
|
||||
/** An `OtelService` stand-in that hands out this provider's tracers. */
|
||||
asOtelService(): OtelService {
|
||||
return mock<OtelService>({ getTracer: (name: string) => this.provider.getTracer(name) });
|
||||
}
|
||||
|
||||
getFinishedSpans() {
|
||||
return this.exporter.getFinishedSpans();
|
||||
}
|
||||
|
|
@ -92,7 +101,6 @@ export class OtelTestProvider {
|
|||
this.exporter.reset();
|
||||
await this.provider.shutdown();
|
||||
trace.disable();
|
||||
propagation.disable();
|
||||
if (this.contextManagerEnabled) context.disable();
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,7 +1,15 @@
|
|||
import { Logger } from '@n8n/backend-common';
|
||||
import { Service } from '@n8n/di';
|
||||
import type { Context, Exception, Span } from '@opentelemetry/api';
|
||||
import { context, propagation, SpanStatusCode, trace } from '@opentelemetry/api';
|
||||
import {
|
||||
context,
|
||||
defaultTextMapGetter,
|
||||
defaultTextMapSetter,
|
||||
ROOT_CONTEXT,
|
||||
SpanStatusCode,
|
||||
trace,
|
||||
} from '@opentelemetry/api';
|
||||
import { W3CTraceContextPropagator } from '@opentelemetry/core';
|
||||
import type { ExecutionStatus } from 'n8n-workflow';
|
||||
|
||||
import { WorkflowCrashedError } from '@/errors/workflow-crashed.error';
|
||||
|
|
@ -16,9 +24,11 @@ import {
|
|||
} from './execution-level-tracer.types';
|
||||
import { OtelSettingsService } from './otel-settings.service';
|
||||
import { ATTR } from './otel.constants';
|
||||
import { OtelService } from './otel.service';
|
||||
import type { TracingContext } from './tracing-context';
|
||||
|
||||
const TRACER_NAME = 'n8n-workflow';
|
||||
const propagator = new W3CTraceContextPropagator();
|
||||
const UNKNOWN_ERROR_TYPE = 'UnknownError';
|
||||
const OBJECT_ERROR_TYPE = 'Object';
|
||||
|
||||
|
|
@ -32,22 +42,17 @@ type TrackedSpan = { span: Span };
|
|||
export class ExecutionLevelTracer {
|
||||
private readonly activeWorkflowSpans = new Map<string, TrackedSpan>();
|
||||
private readonly activeNodeSpansByExecutionId = new Map<string, Map<string, TrackedSpan>>();
|
||||
private tracer = trace.getTracer(TRACER_NAME);
|
||||
|
||||
/**
|
||||
* Called by OtelService after a SDK restart so this instance picks up the
|
||||
* new NodeTracerProvider. Without this, the cached NodeTracer stays bound
|
||||
* to the old (shutdown) provider and all spans are silently dropped.
|
||||
*/
|
||||
refreshTracer(): void {
|
||||
this.tracer = trace.getTracer(TRACER_NAME);
|
||||
}
|
||||
|
||||
constructor(
|
||||
private readonly otelService: OtelService,
|
||||
private readonly otelSettingsService: OtelSettingsService,
|
||||
private readonly logger: Logger,
|
||||
) {}
|
||||
|
||||
private get tracer() {
|
||||
return this.otelService.getTracer(TRACER_NAME);
|
||||
}
|
||||
|
||||
startWorkflow(params: StartWorkflowParams) {
|
||||
try {
|
||||
const parentCtx = this.parseTraceParentHeaders(params.tracingContext);
|
||||
|
|
@ -271,7 +276,7 @@ export class ExecutionLevelTracer {
|
|||
const span = this.findMostSpecificSpan(executionId, nodeName);
|
||||
if (!span) return;
|
||||
|
||||
propagation.inject(trace.setSpan(context.active(), span), headers);
|
||||
propagator.inject(trace.setSpan(ROOT_CONTEXT, span), headers, defaultTextMapSetter);
|
||||
} catch (error) {
|
||||
this.logger.warn('Failed to inject trace headers', {
|
||||
executionId,
|
||||
|
|
@ -281,15 +286,14 @@ export class ExecutionLevelTracer {
|
|||
}
|
||||
}
|
||||
|
||||
private parseTraceParentHeaders(tracingContext?: TracingContext) {
|
||||
return tracingContext
|
||||
? propagation.extract(context.active(), tracingContext)
|
||||
: context.active();
|
||||
private parseTraceParentHeaders(tracingContext?: TracingContext): Context {
|
||||
if (!tracingContext) return ROOT_CONTEXT;
|
||||
return propagator.extract(ROOT_CONTEXT, tracingContext, defaultTextMapGetter);
|
||||
}
|
||||
|
||||
private buildContinuationLinks(linkTo?: TracingContext) {
|
||||
if (!linkTo) return undefined;
|
||||
const extracted = propagation.extract(context.active(), linkTo);
|
||||
const extracted = propagator.extract(ROOT_CONTEXT, linkTo, defaultTextMapGetter);
|
||||
const spanContext = trace.getSpanContext(extracted);
|
||||
if (!spanContext) return undefined;
|
||||
return [
|
||||
|
|
@ -348,7 +352,7 @@ function buildNodeEndAttributes(params: EndNodeParams): Record<string, string |
|
|||
|
||||
function toTracingParentContext(span: Span): TracingContext {
|
||||
const carrier: Record<string, string> = {};
|
||||
propagation.inject(trace.setSpan(context.active(), span), carrier);
|
||||
propagator.inject(trace.setSpan(ROOT_CONTEXT, span), carrier, defaultTextMapSetter);
|
||||
return { traceparent: carrier.traceparent, tracestate: carrier.tracestate };
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -62,7 +62,6 @@ export class OtelLifecycleHandler {
|
|||
@OnPubSubEvent('reload-otel-config')
|
||||
async onReloadOtelConfig(): Promise<void> {
|
||||
await this.otelService.restart();
|
||||
this.tracer.refreshTracer();
|
||||
}
|
||||
|
||||
private shouldTrace(ctx: { type: string; mode: WorkflowExecuteMode }): boolean {
|
||||
|
|
|
|||
|
|
@ -2,18 +2,24 @@ import type { Metadata } from '@grpc/grpc-js';
|
|||
import { Logger } from '@n8n/backend-common';
|
||||
import { OutboundHttp } from '@n8n/backend-network';
|
||||
import { Service } from '@n8n/di';
|
||||
import type { DiagLogger } from '@opentelemetry/api';
|
||||
import { DiagLogLevel, diag, context, metrics, propagation, trace } from '@opentelemetry/api';
|
||||
import type { DiagLogger, Tracer, TracerProvider } from '@opentelemetry/api';
|
||||
import { DiagLogLevel, ProxyTracerProvider, diag, trace } from '@opentelemetry/api';
|
||||
import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-proto';
|
||||
import { resourceFromAttributes } from '@opentelemetry/resources';
|
||||
import { NodeSDK } from '@opentelemetry/sdk-node';
|
||||
import {
|
||||
detectResources,
|
||||
envDetector,
|
||||
hostDetector,
|
||||
processDetector,
|
||||
resourceFromAttributes,
|
||||
} from '@opentelemetry/resources';
|
||||
import {
|
||||
BasicTracerProvider,
|
||||
BatchSpanProcessor,
|
||||
type ReadableSpan,
|
||||
type SpanExporter,
|
||||
type SpanProcessor,
|
||||
} from '@opentelemetry/sdk-trace-base';
|
||||
import { TraceIdRatioBasedSampler } from '@opentelemetry/sdk-trace-node';
|
||||
import { NodeTracerProvider, TraceIdRatioBasedSampler } from '@opentelemetry/sdk-trace-node';
|
||||
import { InstanceSettings } from 'n8n-core';
|
||||
import { OperationalError } from 'n8n-workflow';
|
||||
|
||||
|
|
@ -29,10 +35,22 @@ export type OtelTestTraceResult = { success: true } | { success: false; error: s
|
|||
const stripEmptyResolutionNote = (message: string) =>
|
||||
message.replace(/\s*Resolution note:\s*$/, '');
|
||||
|
||||
// Deprecated in `@opentelemetry/api` 1.x, but the only public source of a no-op tracer provider.
|
||||
const noopTracerProvider = new ProxyTracerProvider();
|
||||
|
||||
function registeredGlobalTracerProvider(): TracerProvider | undefined {
|
||||
const globalProvider = trace.getTracerProvider();
|
||||
const delegate =
|
||||
globalProvider instanceof ProxyTracerProvider ? globalProvider.getDelegate() : globalProvider;
|
||||
return delegate === noopTracerProvider.getDelegate() ? undefined : delegate;
|
||||
}
|
||||
|
||||
@Service()
|
||||
export class OtelService {
|
||||
private static isDiagnosticsLoggerConfigured = false;
|
||||
private sdk?: NodeSDK;
|
||||
private provider?: NodeTracerProvider;
|
||||
private ownsGlobalApi = false;
|
||||
private hasLoggedForeignGlobalApiOwner = false;
|
||||
|
||||
constructor(
|
||||
private readonly otelSettingsService: OtelSettingsService,
|
||||
|
|
@ -52,9 +70,13 @@ export class OtelService {
|
|||
await this.start(settings);
|
||||
}
|
||||
|
||||
getTracer(name: string): Tracer {
|
||||
return (this.provider ?? noopTracerProvider).getTracer(name);
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends a single `n8n.test_trace` span to the given OTLP endpoint and waits
|
||||
* for the exporter's result. Unlike the long-running SDK (which batches spans
|
||||
* for the exporter's result. Unlike the long-running provider (which batches spans
|
||||
* fire-and-forget), this uses a throwaway provider/exporter so the collector's
|
||||
* response — success, a rejection (HTTP status or gRPC status code), or a
|
||||
* network error — can be reported back to the caller. Runs independently of
|
||||
|
|
@ -111,9 +133,9 @@ export class OtelService {
|
|||
if (!settings.enabled) return;
|
||||
|
||||
this.configureDiagnosticsLogger();
|
||||
let sdk: NodeSDK;
|
||||
let provider: NodeTracerProvider;
|
||||
try {
|
||||
sdk = await this.startSdk(settings);
|
||||
provider = await this.startProvider(settings);
|
||||
} catch (error) {
|
||||
this.logger.error('Failed to start OpenTelemetry tracing, so tracing stays off', {
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
|
|
@ -122,47 +144,67 @@ export class OtelService {
|
|||
return;
|
||||
}
|
||||
|
||||
void this.checkEndpointReachability(settings, sdk);
|
||||
void this.checkEndpointReachability(settings, provider);
|
||||
}
|
||||
|
||||
async shutdown(): Promise<void> {
|
||||
// Cleared before the flush, so a probe that fails meanwhile sees that its SDK is gone.
|
||||
const sdk = this.sdk;
|
||||
this.sdk = undefined;
|
||||
// Cleared before the flush, so a probe that fails meanwhile sees that its provider is gone.
|
||||
const provider = this.provider;
|
||||
this.provider = undefined;
|
||||
try {
|
||||
await sdk?.shutdown();
|
||||
await provider?.shutdown();
|
||||
} catch (error) {
|
||||
this.logger.warn(
|
||||
'Failed to cleanly shut down OpenTelemetry SDK (exporter flush may have failed)',
|
||||
'Failed to cleanly shut down OpenTelemetry tracer provider (exporter flush may have failed)',
|
||||
{ error: error instanceof Error ? error.message : String(error) },
|
||||
);
|
||||
} finally {
|
||||
// Unregister the global providers so the next NodeSDK.start() can register
|
||||
// new ones. Without this, OTel's allowOverride=false guard blocks
|
||||
// re-registration and the restart silently fails.
|
||||
trace?.disable();
|
||||
context?.disable();
|
||||
propagation?.disable();
|
||||
metrics?.disable();
|
||||
}
|
||||
}
|
||||
|
||||
private async startSdk(settings: OtelConfig): Promise<NodeSDK> {
|
||||
private async startProvider(settings: OtelConfig): Promise<NodeTracerProvider> {
|
||||
const traceExporter = await this.createTraceExporter(settings);
|
||||
|
||||
this.sdk = new NodeSDK({
|
||||
resource: resourceFromAttributes({
|
||||
[ATTR.OTEL_SERVICE_NAME]: settings.exporterServiceName,
|
||||
[ATTR.OTEL_SERVICE_VERSION]: N8N_VERSION,
|
||||
[ATTR.INSTANCE_ID]: this.instanceSettings.instanceId,
|
||||
[ATTR.INSTANCE_ROLE]: this.instanceSettings.instanceType,
|
||||
}),
|
||||
traceExporter,
|
||||
this.provider = new NodeTracerProvider({
|
||||
resource: this.buildResource(settings.exporterServiceName),
|
||||
sampler: new TraceIdRatioBasedSampler(settings.tracesSampleRate),
|
||||
spanProcessors: [new BatchSpanProcessor(traceExporter)],
|
||||
});
|
||||
this.registerGlobalApi(this.provider);
|
||||
return this.provider;
|
||||
}
|
||||
|
||||
this.sdk.start();
|
||||
return this.sdk;
|
||||
private buildResource(serviceName: string) {
|
||||
return resourceFromAttributes({
|
||||
[ATTR.OTEL_SERVICE_NAME]: serviceName,
|
||||
[ATTR.OTEL_SERVICE_VERSION]: N8N_VERSION,
|
||||
[ATTR.INSTANCE_ID]: this.instanceSettings.instanceId,
|
||||
[ATTR.INSTANCE_ROLE]: this.instanceSettings.instanceType,
|
||||
}).merge(detectResources({ detectors: [envDetector, processDetector, hostDetector] }));
|
||||
}
|
||||
|
||||
private registerGlobalApi(provider: NodeTracerProvider): void {
|
||||
if (this.ownsGlobalApi) {
|
||||
trace.disable();
|
||||
this.ownsGlobalApi = trace.setGlobalTracerProvider(provider);
|
||||
return;
|
||||
}
|
||||
|
||||
if (registeredGlobalTracerProvider()) {
|
||||
this.logForeignGlobalApiOwner();
|
||||
return;
|
||||
}
|
||||
|
||||
provider.register();
|
||||
this.ownsGlobalApi = registeredGlobalTracerProvider() === provider;
|
||||
}
|
||||
|
||||
private logForeignGlobalApiOwner(): void {
|
||||
if (this.hasLoggedForeignGlobalApiOwner) return;
|
||||
|
||||
this.hasLoggedForeignGlobalApiOwner = true;
|
||||
this.logger.info(
|
||||
'Another library owns the global OpenTelemetry API, so n8n workflow tracing runs on its own tracer provider',
|
||||
);
|
||||
}
|
||||
|
||||
private async createTraceExporter(
|
||||
|
|
@ -284,7 +326,7 @@ export class OtelService {
|
|||
|
||||
private async checkEndpointReachability(
|
||||
settings: OtelConfig,
|
||||
startedSdk: NodeSDK,
|
||||
startedProvider: NodeTracerProvider,
|
||||
): Promise<void> {
|
||||
const url = this.resolveExporterUrl(settings);
|
||||
const timeoutMs = settings.startupConnectivityTimeoutMs;
|
||||
|
|
@ -296,8 +338,8 @@ export class OtelService {
|
|||
await this.probeHttpEndpoint(url, timeoutMs);
|
||||
}
|
||||
} catch (error) {
|
||||
// A restart or shutdown replaced the SDK this probe belongs to.
|
||||
if (this.sdk !== startedSdk) return;
|
||||
// A restart or shutdown replaced the provider this probe belongs to.
|
||||
if (this.provider !== startedProvider) return;
|
||||
|
||||
this.logger.error('Failed to connect to OpenTelemetry OTLP endpoint during startup', {
|
||||
endpoint: url,
|
||||
|
|
|
|||
|
|
@ -4453,9 +4453,6 @@ importers:
|
|||
'@opentelemetry/resources':
|
||||
specifier: ^2.10.0
|
||||
version: 2.10.0(@opentelemetry/api@1.9.1)
|
||||
'@opentelemetry/sdk-node':
|
||||
specifier: ^0.221.0
|
||||
version: 0.221.0(@opentelemetry/api@1.9.1)(supports-color@8.1.1)
|
||||
'@opentelemetry/sdk-trace-base':
|
||||
specifier: ^2.10.0
|
||||
version: 2.10.0(@opentelemetry/api@1.9.1)
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user