From 7e74dd0f97a00c4491353999865c50efc6f1664a Mon Sep 17 00:00:00 2001 From: "jinye.djy" Date: Fri, 1 May 2026 09:41:36 +0800 Subject: [PATCH 1/7] feat(telemetry): define HTTP OTLP endpoint behavior and signal routing - Add resolveHttpOtlpUrl() that appends /v1/traces, /v1/logs, /v1/metrics to base HTTP OTLP endpoints per the OpenTelemetry specification - Add per-signal endpoint overrides (otlpTracesEndpoint, otlpLogsEndpoint, otlpMetricsEndpoint) for backends with non-standard paths (e.g. Alibaba Cloud) - Add LogToSpanProcessor that bridges OTel log records to spans for traces-only backends, with session-based traceId correlation and error status propagation - Auto-wire LogToSpanProcessor when traces URL exists but logs URL doesn't - Validate per-signal URLs gracefully (log error + skip, don't crash) - Preserve query strings when appending signal paths to URLs - Guard gRPC branch against missing base endpoint with per-signal config - Update telemetry documentation with signal routing semantics and Alibaba Cloud HTTP per-signal endpoint examples Closes #3734 Co-authored-by: Qwen-Coder --- docs/developers/development/telemetry.md | 60 ++++- packages/core/src/config/config.test.ts | 35 +++ packages/core/src/config/config.ts | 25 +- packages/core/src/telemetry/config.test.ts | 54 ++++- packages/core/src/telemetry/config.ts | 20 ++ .../telemetry/log-to-span-processor.test.ts | 228 ++++++++++++++++++ .../src/telemetry/log-to-span-processor.ts | 209 ++++++++++++++++ packages/core/src/telemetry/sdk.test.ts | 139 ++++++++++- packages/core/src/telemetry/sdk.ts | 187 ++++++++++---- 9 files changed, 885 insertions(+), 72 deletions(-) create mode 100644 packages/core/src/telemetry/log-to-span-processor.test.ts create mode 100644 packages/core/src/telemetry/log-to-span-processor.ts diff --git a/docs/developers/development/telemetry.md b/docs/developers/development/telemetry.md index 006a668a761..93f8f10692b 100644 --- a/docs/developers/development/telemetry.md +++ b/docs/developers/development/telemetry.md @@ -58,20 +58,36 @@ observability framework — Qwen Code's observability system provides: All telemetry behavior is controlled through your `.qwen/settings.json` file. These settings can be overridden by environment variables or CLI flags. -| Setting | Environment Variable | CLI Flag | Description | Values | Default | -| -------------- | ------------------------------ | -------------------------------------------------------- | ------------------------------------------------- | ----------------- | ----------------------- | -| `enabled` | `QWEN_TELEMETRY_ENABLED` | `--telemetry` / `--no-telemetry` | Enable or disable telemetry | `true`/`false` | `false` | -| `target` | `QWEN_TELEMETRY_TARGET` | `--telemetry-target ` | Where to send telemetry data | `"gcp"`/`"local"` | `"local"` | -| `otlpEndpoint` | `QWEN_TELEMETRY_OTLP_ENDPOINT` | `--telemetry-otlp-endpoint ` | OTLP collector endpoint | URL string | `http://localhost:4317` | -| `otlpProtocol` | `QWEN_TELEMETRY_OTLP_PROTOCOL` | `--telemetry-otlp-protocol ` | OTLP transport protocol | `"grpc"`/`"http"` | `"grpc"` | -| `outfile` | `QWEN_TELEMETRY_OUTFILE` | `--telemetry-outfile ` | Save telemetry to file (overrides `otlpEndpoint`) | file path | - | -| `logPrompts` | `QWEN_TELEMETRY_LOG_PROMPTS` | `--telemetry-log-prompts` / `--no-telemetry-log-prompts` | Include prompts in telemetry logs | `true`/`false` | `true` | -| `useCollector` | `QWEN_TELEMETRY_USE_COLLECTOR` | - | Use external OTLP collector (advanced) | `true`/`false` | `false` | +| Setting | Environment Variable | CLI Flag | Description | Values | Default | +| --------------------- | -------------------------------------- | -------------------------------------------------------- | ---------------------------------------------------- | ----------------- | ----------------------- | +| `enabled` | `QWEN_TELEMETRY_ENABLED` | `--telemetry` / `--no-telemetry` | Enable or disable telemetry | `true`/`false` | `false` | +| `target` | `QWEN_TELEMETRY_TARGET` | `--telemetry-target ` | Where to send telemetry data | `"gcp"`/`"local"` | `"local"` | +| `otlpEndpoint` | `QWEN_TELEMETRY_OTLP_ENDPOINT` | `--telemetry-otlp-endpoint ` | OTLP collector endpoint | URL string | `http://localhost:4317` | +| `otlpProtocol` | `QWEN_TELEMETRY_OTLP_PROTOCOL` | `--telemetry-otlp-protocol ` | OTLP transport protocol | `"grpc"`/`"http"` | `"grpc"` | +| `otlpTracesEndpoint` | `QWEN_TELEMETRY_OTLP_TRACES_ENDPOINT` | - | Per-signal endpoint override for traces (HTTP only) | URL string | - | +| `otlpLogsEndpoint` | `QWEN_TELEMETRY_OTLP_LOGS_ENDPOINT` | - | Per-signal endpoint override for logs (HTTP only) | URL string | - | +| `otlpMetricsEndpoint` | `QWEN_TELEMETRY_OTLP_METRICS_ENDPOINT` | - | Per-signal endpoint override for metrics (HTTP only) | URL string | - | +| `outfile` | `QWEN_TELEMETRY_OUTFILE` | `--telemetry-outfile ` | Save telemetry to file (overrides `otlpEndpoint`) | file path | - | +| `logPrompts` | `QWEN_TELEMETRY_LOG_PROMPTS` | `--telemetry-log-prompts` / `--no-telemetry-log-prompts` | Include prompts in telemetry logs | `true`/`false` | `true` | +| `useCollector` | `QWEN_TELEMETRY_USE_COLLECTOR` | - | Use external OTLP collector (advanced) | `true`/`false` | `false` | **Note on boolean environment variables:** For the boolean settings (`enabled`, `logPrompts`, `useCollector`), setting the corresponding environment variable to `true` or `1` will enable the feature. Any other value will disable it. +**HTTP OTLP signal routing:** When using HTTP protocol (`otlpProtocol: "http"`), +Qwen Code automatically appends signal-specific paths (`/v1/traces`, `/v1/logs`, +`/v1/metrics`) to the base `otlpEndpoint`. For example, `http://collector:4318` +becomes `http://collector:4318/v1/traces` for traces. If the URL already ends +with a signal path, it is used as-is. Per-signal endpoint overrides +(`otlpTracesEndpoint`, etc.) take precedence over the base endpoint and are used +verbatim. gRPC protocol uses service-based routing and does not append paths. + +The per-signal endpoint environment variables also accept the standard +OpenTelemetry names: `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT`, +`OTEL_EXPORTER_OTLP_LOGS_ENDPOINT`, `OTEL_EXPORTER_OTLP_METRICS_ENDPOINT`. +The `QWEN_TELEMETRY_OTLP_*` variants take precedence over the `OTEL_*` variants. + For detailed information about all configuration options, see the [Configuration Guide](./cli/configuration.md). @@ -91,6 +107,9 @@ sent to Alibaba Cloud. 1. Enable telemetry in your `.qwen/settings.json` and set the OTLP endpoint: + + **Option A: gRPC protocol** (standard OTLP endpoint): + ```json { "telemetry": { @@ -101,6 +120,29 @@ sent to Alibaba Cloud. } } ``` + + **Option B: HTTP protocol with per-signal endpoints** (for backends + that use non-standard paths, e.g., `/api/otlp/traces` instead of + `/v1/traces`): + + ```json + { + "telemetry": { + "enabled": true, + "otlpProtocol": "http", + "otlpTracesEndpoint": "http:////api/otlp/traces", + "otlpLogsEndpoint": "http:////api/otlp/logs", + "otlpMetricsEndpoint": "http:////api/otlp/metrics" + } + } + ``` + + > **Note:** When using HTTP protocol with only `otlpEndpoint` (no + > per-signal overrides), Qwen Code appends standard OTLP paths + > (`/v1/traces`, `/v1/logs`, `/v1/metrics`) to the base URL. If your + > backend uses different paths, use per-signal endpoint overrides as + > shown in Option B. + 2. If your Alibaba Cloud endpoint requires authentication, provide OTLP headers through standard OpenTelemetry environment variables such as `OTEL_EXPORTER_OTLP_HEADERS` (or the signal-specific variants). Qwen diff --git a/packages/core/src/config/config.test.ts b/packages/core/src/config/config.test.ts index c1eb09231a4..62663eced3b 100644 --- a/packages/core/src/config/config.test.ts +++ b/packages/core/src/config/config.test.ts @@ -1023,6 +1023,41 @@ describe('Server Config (config.ts)', () => { }); }); + describe('Per-Signal OTLP Endpoint Configuration', () => { + it('should return per-signal endpoints when provided', () => { + const params: ConfigParameters = { + ...baseParams, + telemetry: { + enabled: true, + otlpTracesEndpoint: 'http://traces:4318/v1/traces', + otlpLogsEndpoint: 'http://logs:4318/v1/logs', + otlpMetricsEndpoint: 'http://metrics:4318/v1/metrics', + }, + }; + const config = new Config(params); + expect(config.getTelemetryOtlpTracesEndpoint()).toBe( + 'http://traces:4318/v1/traces', + ); + expect(config.getTelemetryOtlpLogsEndpoint()).toBe( + 'http://logs:4318/v1/logs', + ); + expect(config.getTelemetryOtlpMetricsEndpoint()).toBe( + 'http://metrics:4318/v1/metrics', + ); + }); + + it('should return undefined when per-signal endpoints are not provided', () => { + const params: ConfigParameters = { + ...baseParams, + telemetry: { enabled: true }, + }; + const config = new Config(params); + expect(config.getTelemetryOtlpTracesEndpoint()).toBeUndefined(); + expect(config.getTelemetryOtlpLogsEndpoint()).toBeUndefined(); + expect(config.getTelemetryOtlpMetricsEndpoint()).toBeUndefined(); + }); + }); + describe('UseRipgrep Configuration', () => { it('should default useRipgrep to true when not provided', () => { const config = new Config(baseParams); diff --git a/packages/core/src/config/config.ts b/packages/core/src/config/config.ts index f4b2c42fc33..c8cadc519c8 100644 --- a/packages/core/src/config/config.ts +++ b/packages/core/src/config/config.ts @@ -215,6 +215,12 @@ export interface TelemetrySettings { target?: TelemetryTarget; otlpEndpoint?: string; otlpProtocol?: 'grpc' | 'http'; + /** Per-signal endpoint override for traces (HTTP only). Used as-is without path appending. */ + otlpTracesEndpoint?: string; + /** Per-signal endpoint override for logs (HTTP only). Used as-is without path appending. */ + otlpLogsEndpoint?: string; + /** Per-signal endpoint override for metrics (HTTP only). Used as-is without path appending. */ + otlpMetricsEndpoint?: string; logPrompts?: boolean; outfile?: string; useCollector?: boolean; @@ -741,8 +747,11 @@ export class Config { this.telemetrySettings = { enabled: params.telemetry?.enabled ?? false, target: params.telemetry?.target ?? DEFAULT_TELEMETRY_TARGET, - otlpEndpoint: params.telemetry?.otlpEndpoint ?? DEFAULT_OTLP_ENDPOINT, + otlpEndpoint: params.telemetry?.otlpEndpoint, otlpProtocol: params.telemetry?.otlpProtocol, + otlpTracesEndpoint: params.telemetry?.otlpTracesEndpoint, + otlpLogsEndpoint: params.telemetry?.otlpLogsEndpoint, + otlpMetricsEndpoint: params.telemetry?.otlpMetricsEndpoint, logPrompts: params.telemetry?.logPrompts ?? true, outfile: params.telemetry?.outfile, useCollector: params.telemetry?.useCollector, @@ -1975,7 +1984,7 @@ export class Config { return this.telemetrySettings.logPrompts ?? true; } - getTelemetryOtlpEndpoint(): string { + getTelemetryOtlpEndpoint(): string | undefined { return this.telemetrySettings.otlpEndpoint ?? DEFAULT_OTLP_ENDPOINT; } @@ -1983,6 +1992,18 @@ export class Config { return this.telemetrySettings.otlpProtocol ?? 'grpc'; } + getTelemetryOtlpTracesEndpoint(): string | undefined { + return this.telemetrySettings.otlpTracesEndpoint; + } + + getTelemetryOtlpLogsEndpoint(): string | undefined { + return this.telemetrySettings.otlpLogsEndpoint; + } + + getTelemetryOtlpMetricsEndpoint(): string | undefined { + return this.telemetrySettings.otlpMetricsEndpoint; + } + getTelemetryTarget(): TelemetryTarget { return this.telemetrySettings.target ?? DEFAULT_TELEMETRY_TARGET; } diff --git a/packages/core/src/telemetry/config.test.ts b/packages/core/src/telemetry/config.test.ts index 443282fd439..f1c0d3f07fc 100644 --- a/packages/core/src/telemetry/config.test.ts +++ b/packages/core/src/telemetry/config.test.ts @@ -65,7 +65,12 @@ describe('telemetry/config helpers', () => { useCollector: false, }; const resolved = await resolveTelemetrySettings({ settings }); - expect(resolved).toEqual(settings); + expect(resolved).toEqual({ + ...settings, + otlpTracesEndpoint: undefined, + otlpLogsEndpoint: undefined, + otlpMetricsEndpoint: undefined, + }); }); it('uses env over settings and argv over env', async () => { @@ -102,6 +107,9 @@ describe('telemetry/config helpers', () => { target: TelemetryTarget.GCP, otlpEndpoint: 'http://env:4317', otlpProtocol: 'http', + otlpTracesEndpoint: undefined, + otlpLogsEndpoint: undefined, + otlpMetricsEndpoint: undefined, logPrompts: true, outfile: 'env.log', useCollector: true, @@ -117,6 +125,9 @@ describe('telemetry/config helpers', () => { target: TelemetryTarget.LOCAL, otlpEndpoint: 'http://argv:4317', otlpProtocol: 'grpc', + otlpTracesEndpoint: undefined, + otlpLogsEndpoint: undefined, + otlpMetricsEndpoint: undefined, logPrompts: false, outfile: 'argv.log', useCollector: true, // from env as no argv option @@ -151,5 +162,46 @@ describe('telemetry/config helpers', () => { /Invalid telemetry target/i, ); }); + + it('resolves per-signal endpoints from OTEL_ env vars', async () => { + const env = { + OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: 'http://traces:4318/v1/traces', + OTEL_EXPORTER_OTLP_LOGS_ENDPOINT: 'http://logs:4318/v1/logs', + } as Record; + + const resolved = await resolveTelemetrySettings({ env }); + expect(resolved.otlpTracesEndpoint).toBe('http://traces:4318/v1/traces'); + expect(resolved.otlpLogsEndpoint).toBe('http://logs:4318/v1/logs'); + expect(resolved.otlpMetricsEndpoint).toBeUndefined(); + }); + + it('QWEN_ env vars take precedence over OTEL_ vars for per-signal endpoints', async () => { + const env = { + QWEN_TELEMETRY_OTLP_TRACES_ENDPOINT: + 'http://qwen-traces:4318/v1/traces', + OTEL_EXPORTER_OTLP_TRACES_ENDPOINT: 'http://otel-traces:4318/v1/traces', + } as Record; + + const resolved = await resolveTelemetrySettings({ env }); + expect(resolved.otlpTracesEndpoint).toBe( + 'http://qwen-traces:4318/v1/traces', + ); + }); + + it('resolves per-signal endpoints from settings', async () => { + const settings = { + otlpTracesEndpoint: 'http://traces-settings:4318/v1/traces', + otlpMetricsEndpoint: 'http://metrics-settings:4318/v1/metrics', + }; + + const resolved = await resolveTelemetrySettings({ settings }); + expect(resolved.otlpTracesEndpoint).toBe( + 'http://traces-settings:4318/v1/traces', + ); + expect(resolved.otlpLogsEndpoint).toBeUndefined(); + expect(resolved.otlpMetricsEndpoint).toBe( + 'http://metrics-settings:4318/v1/metrics', + ); + }); }); }); diff --git a/packages/core/src/telemetry/config.ts b/packages/core/src/telemetry/config.ts index f1037e74276..267124ed70e 100644 --- a/packages/core/src/telemetry/config.ts +++ b/packages/core/src/telemetry/config.ts @@ -106,11 +106,31 @@ export async function resolveTelemetrySettings(options: { parseBooleanEnvFlag(env['QWEN_TELEMETRY_USE_COLLECTOR']) ?? settings.useCollector; + // Per-signal endpoint overrides (HTTP only). + // Priority: QWEN_ env var > standard OTEL_ env var > settings.json + const otlpTracesEndpoint = + env['QWEN_TELEMETRY_OTLP_TRACES_ENDPOINT'] ?? + env['OTEL_EXPORTER_OTLP_TRACES_ENDPOINT'] ?? + settings.otlpTracesEndpoint; + + const otlpLogsEndpoint = + env['QWEN_TELEMETRY_OTLP_LOGS_ENDPOINT'] ?? + env['OTEL_EXPORTER_OTLP_LOGS_ENDPOINT'] ?? + settings.otlpLogsEndpoint; + + const otlpMetricsEndpoint = + env['QWEN_TELEMETRY_OTLP_METRICS_ENDPOINT'] ?? + env['OTEL_EXPORTER_OTLP_METRICS_ENDPOINT'] ?? + settings.otlpMetricsEndpoint; + return { enabled, target, otlpEndpoint, otlpProtocol, + otlpTracesEndpoint, + otlpLogsEndpoint, + otlpMetricsEndpoint, logPrompts, outfile, useCollector, diff --git a/packages/core/src/telemetry/log-to-span-processor.test.ts b/packages/core/src/telemetry/log-to-span-processor.test.ts new file mode 100644 index 00000000000..20b9c9fb396 --- /dev/null +++ b/packages/core/src/telemetry/log-to-span-processor.test.ts @@ -0,0 +1,228 @@ +/** + * @license + * Copyright 2025 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; +import { SpanKind, SpanStatusCode } from '@opentelemetry/api'; +import { LogToSpanProcessor } from './log-to-span-processor.js'; +import type { ReadableLogRecord } from '@opentelemetry/sdk-logs'; +import type { SpanExporter } from '@opentelemetry/sdk-trace-base'; + +describe('LogToSpanProcessor', () => { + let processor: LogToSpanProcessor; + let mockExporter: SpanExporter; + let exportedSpans: Array>; + + beforeEach(() => { + exportedSpans = []; + mockExporter = { + export: vi.fn((spans, cb) => { + exportedSpans.push(...spans); + cb({ code: 0 }); + }), + shutdown: vi.fn().mockResolvedValue(undefined), + forceFlush: vi.fn().mockResolvedValue(undefined), + } as unknown as SpanExporter; + processor = new LogToSpanProcessor(mockExporter, 60000); // long interval to avoid auto-flush + }); + + afterEach(async () => { + await processor.shutdown(); + }); + + it('converts a log record to a span on flush', async () => { + const logRecord = { + body: 'test event', + hrTime: [1000, 500000000] as [number, number], + attributes: { + key1: 'value1', + key2: 42, + key3: true, + }, + } as unknown as ReadableLogRecord; + + processor.onEmit(logRecord); + await processor.forceFlush(); + + expect(exportedSpans).toHaveLength(1); + const span = exportedSpans[0]; + expect(span.name).toBe('test event'); + expect(span.kind).toBe(SpanKind.INTERNAL); + expect(span.attributes.key1).toBe('value1'); + expect(span.attributes.key2).toBe(42); + expect(span.attributes.key3).toBe(true); + expect(span.attributes['log.bridge']).toBe(true); + expect(span.startTime).toEqual([1000, 500000000]); + // Instant span: end time == start time + expect(span.endTime).toEqual([1000, 500000000]); + expect(span.status.code).toBe(SpanStatusCode.OK); + }); + + it('uses duration_ms to compute span end time', async () => { + const logRecord = { + body: 'api response', + hrTime: [1000, 0] as [number, number], + attributes: { + duration_ms: 250, + model: 'test-model', + }, + } as unknown as ReadableLogRecord; + + processor.onEmit(logRecord); + await processor.forceFlush(); + + const span = exportedSpans[0]; + // 250ms = 250_000_000 nanoseconds + expect(span.endTime).toEqual([1000, 250000000]); + }); + + it('handles duration_ms that causes second rollover', async () => { + const logRecord = { + body: 'long operation', + hrTime: [1000, 900000000] as [number, number], + attributes: { + duration_ms: 500, + }, + } as unknown as ReadableLogRecord; + + processor.onEmit(logRecord); + await processor.forceFlush(); + + const span = exportedSpans[0]; + // 900_000_000 + 500_000_000 = 1_400_000_000 → [1001, 400000000] + expect(span.endTime).toEqual([1001, 400000000]); + }); + + it('serializes object attributes to JSON', async () => { + const logRecord = { + body: 'event with object', + hrTime: [1000, 0] as [number, number], + attributes: { + metadata: { nested: true }, + }, + } as unknown as ReadableLogRecord; + + processor.onEmit(logRecord); + await processor.forceFlush(); + + expect(exportedSpans[0].attributes.metadata).toBe('{"nested":true}'); + }); + + it('skips null and undefined attributes', async () => { + const logRecord = { + body: 'event', + hrTime: [1000, 0] as [number, number], + attributes: { + valid: 'yes', + nullVal: null, + undefinedVal: undefined, + }, + } as unknown as ReadableLogRecord; + + processor.onEmit(logRecord); + await processor.forceFlush(); + + const attrs = exportedSpans[0].attributes; + expect(attrs.valid).toBe('yes'); + expect(attrs).not.toHaveProperty('nullVal'); + expect(attrs).not.toHaveProperty('undefinedVal'); + expect(attrs['log.bridge']).toBe(true); + }); + + it('uses "unknown" as span name when body is missing', async () => { + const logRecord = { + body: undefined, + hrTime: [1000, 0] as [number, number], + attributes: {}, + } as unknown as ReadableLogRecord; + + processor.onEmit(logRecord); + await processor.forceFlush(); + + expect(exportedSpans[0].name).toBe('unknown'); + }); + + it('generates unique trace IDs without session.id', async () => { + const logRecord1 = { + body: 'event1', + hrTime: [1000, 0] as [number, number], + attributes: {}, + } as unknown as ReadableLogRecord; + const logRecord2 = { + body: 'event2', + hrTime: [1001, 0] as [number, number], + attributes: {}, + } as unknown as ReadableLogRecord; + + processor.onEmit(logRecord1); + processor.onEmit(logRecord2); + await processor.forceFlush(); + + const ctx1 = exportedSpans[0].spanContext(); + const ctx2 = exportedSpans[1].spanContext(); + expect(ctx1.traceId).toHaveLength(32); + expect(ctx1.spanId).toHaveLength(16); + expect(ctx1.traceId).not.toBe(ctx2.traceId); + }); + + it('derives same traceId from same session.id', async () => { + const logRecord1 = { + body: 'event1', + hrTime: [1000, 0] as [number, number], + attributes: { 'session.id': 'session-abc' }, + } as unknown as ReadableLogRecord; + const logRecord2 = { + body: 'event2', + hrTime: [1001, 0] as [number, number], + attributes: { 'session.id': 'session-abc' }, + } as unknown as ReadableLogRecord; + + processor.onEmit(logRecord1); + processor.onEmit(logRecord2); + await processor.forceFlush(); + + const ctx1 = exportedSpans[0].spanContext(); + const ctx2 = exportedSpans[1].spanContext(); + // Same session → same traceId + expect(ctx1.traceId).toBe(ctx2.traceId); + // Different spanIds + expect(ctx1.spanId).not.toBe(ctx2.spanId); + }); + + it('derives different traceIds from different session.ids', async () => { + const logRecord1 = { + body: 'event1', + hrTime: [1000, 0] as [number, number], + attributes: { 'session.id': 'session-abc' }, + } as unknown as ReadableLogRecord; + const logRecord2 = { + body: 'event2', + hrTime: [1001, 0] as [number, number], + attributes: { 'session.id': 'session-xyz' }, + } as unknown as ReadableLogRecord; + + processor.onEmit(logRecord1); + processor.onEmit(logRecord2); + await processor.forceFlush(); + + const ctx1 = exportedSpans[0].spanContext(); + const ctx2 = exportedSpans[1].spanContext(); + expect(ctx1.traceId).not.toBe(ctx2.traceId); + }); + + it('shutdown flushes remaining spans and shuts down exporter', async () => { + const logRecord = { + body: 'final event', + hrTime: [1000, 0] as [number, number], + attributes: {}, + } as unknown as ReadableLogRecord; + + processor.onEmit(logRecord); + await processor.shutdown(); + + expect(exportedSpans).toHaveLength(1); + expect(mockExporter.shutdown).toHaveBeenCalled(); + }); +}); diff --git a/packages/core/src/telemetry/log-to-span-processor.ts b/packages/core/src/telemetry/log-to-span-processor.ts new file mode 100644 index 00000000000..ecb82373c12 --- /dev/null +++ b/packages/core/src/telemetry/log-to-span-processor.ts @@ -0,0 +1,209 @@ +/** + * @license + * Copyright 2025 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import { SpanKind, SpanStatusCode, type HrTime } from '@opentelemetry/api'; +import type { + LogRecordProcessor, + ReadableLogRecord, +} from '@opentelemetry/sdk-logs'; +import type { SpanExporter, ReadableSpan } from '@opentelemetry/sdk-trace-base'; +import { + type Resource, + resourceFromAttributes, +} from '@opentelemetry/resources'; + +import { createHash } from 'node:crypto'; + +import { SERVICE_NAME } from './constants.js'; + +/** + * A LogRecordProcessor that converts each OTel log record into a span + * and exports it directly through the provided SpanExporter. + * + * This bridges the gap for backends (e.g., Alibaba Cloud) that support + * traces and metrics but not logs over OTLP. Instead of going through + * the global TracerProvider (which can break in bundled environments), + * this processor directly constructs ReadableSpan objects and feeds + * them to the exporter. + * + * When a log record has a `duration_ms` attribute, the resulting span + * will have a matching duration. Otherwise, the span is instantaneous. + */ +export class LogToSpanProcessor implements LogRecordProcessor { + private buffer: ReadableSpanLike[] = []; + private flushTimer: ReturnType | undefined; + private readonly flushIntervalMs: number; + + constructor( + private readonly spanExporter: SpanExporter, + flushIntervalMs = 5000, + ) { + this.flushIntervalMs = flushIntervalMs; + this.flushTimer = setInterval(() => { + void this.flush(); + }, this.flushIntervalMs); + this.flushTimer.unref(); + } + + onEmit(logRecord: ReadableLogRecord): void { + const name = String(logRecord.body ?? 'unknown'); + const startTime = logRecord.hrTime; + + const attributes: Record = {}; + if (logRecord.attributes) { + for (const [key, value] of Object.entries(logRecord.attributes)) { + if (value !== undefined && value !== null) { + attributes[key] = + typeof value === 'object' ? JSON.stringify(value) : value; + } + } + } + attributes['log.bridge'] = true; + + let endTime = startTime; + const durationMs = logRecord.attributes?.['duration_ms']; + if (typeof durationMs === 'number' && durationMs > 0) { + const [secs, nanos] = startTime; + const durationNanos = durationMs * 1_000_000; + const endNanos = nanos + durationNanos; + endTime = [secs + Math.floor(endNanos / 1e9), endNanos % 1e9] as HrTime; + } + + // Derive traceId from session.id so all events in one session + // appear under a single trace. spanId is random per event. + const sessionId = logRecord.attributes?.['session.id']; + const traceId = sessionId + ? deriveTraceId(String(sessionId)) + : randomHexString(32); + const spanId = randomHexString(16); + + this.buffer.push({ + name, + kind: SpanKind.INTERNAL, + spanContext: () => ({ + traceId, + spanId, + traceFlags: 1, // SAMPLED + }), + startTime, + endTime, + duration: hrTimeDiff(startTime, endTime), + attributes, + status: deriveSpanStatus(logRecord.attributes), + events: [], + links: [], + resource: logRecord.resource ?? resourceFromAttributes({}), + instrumentationScope: { + name: SERVICE_NAME, + version: '', + }, + ended: true, + parentSpanContext: undefined, + droppedAttributesCount: 0, + droppedEventsCount: 0, + droppedLinksCount: 0, + recordException: () => {}, + }); + } + + private flush(): Promise { + if (this.buffer.length === 0) return Promise.resolve(); + const spans = this.buffer.splice(0); + return new Promise((resolve) => { + this.spanExporter.export(spans as unknown as ReadableSpan[], (result) => { + if (result.code !== 0) { + process.stderr.write( + `[LogToSpan] export failed: code=${result.code} error=${result.error?.message ?? 'unknown'}\n`, + ); + } + resolve(); + }); + }); + } + + async shutdown(): Promise { + if (this.flushTimer) { + clearInterval(this.flushTimer); + this.flushTimer = undefined; + } + await this.flush(); + await this.spanExporter.shutdown(); + } + + async forceFlush(): Promise { + await this.flush(); + await this.spanExporter.forceFlush?.(); + } +} + +interface ReadableSpanLike { + name: string; + kind: SpanKind; + spanContext: () => { traceId: string; spanId: string; traceFlags: number }; + startTime: HrTime; + endTime: HrTime; + duration: HrTime; + attributes: Record; + status: { code: SpanStatusCode; message?: string }; + events: never[]; + links: never[]; + resource: Resource; + instrumentationScope: { name: string; version?: string; schemaUrl?: string }; + ended: boolean; + parentSpanContext?: { traceId: string; spanId: string; traceFlags: number }; + droppedAttributesCount: number; + droppedEventsCount: number; + droppedLinksCount: number; + recordException: () => void; +} + +function randomHexString(length: number): string { + const bytes = new Uint8Array(length / 2); + crypto.getRandomValues(bytes); + return Array.from(bytes, (b) => b.toString(16).padStart(2, '0')).join(''); +} + +/** + * Derive a deterministic 32-char hex traceId from a session ID. + * All events in the same session will share this traceId, + * making them appear under a single trace in the backend. + */ +function deriveTraceId(sessionId: string): string { + return createHash('md5').update(sessionId).digest('hex'); +} + +/** + * Derive span status from log record attributes. + * Marks the span as ERROR when common error indicators are present. + */ +function deriveSpanStatus(attrs: Record | undefined): { + code: SpanStatusCode; + message?: string; +} { + if (!attrs) return { code: SpanStatusCode.OK }; + if ( + attrs['success'] === false || + attrs['error'] !== undefined || + attrs['error_message'] !== undefined || + attrs['error_type'] !== undefined + ) { + const msg = String( + attrs['error_message'] ?? attrs['error'] ?? attrs['error_type'] ?? '', + ); + return { code: SpanStatusCode.ERROR, ...(msg && { message: msg }) }; + } + return { code: SpanStatusCode.OK }; +} + +function hrTimeDiff(start: HrTime, end: HrTime): HrTime { + let secs = end[0] - start[0]; + let nanos = end[1] - start[1]; + if (nanos < 0) { + secs -= 1; + nanos += 1e9; + } + return [secs, nanos] as HrTime; +} diff --git a/packages/core/src/telemetry/sdk.test.ts b/packages/core/src/telemetry/sdk.test.ts index 9274631dbba..2d760597b28 100644 --- a/packages/core/src/telemetry/sdk.test.ts +++ b/packages/core/src/telemetry/sdk.test.ts @@ -6,7 +6,11 @@ import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; import type { Config } from '../config/config.js'; -import { initializeTelemetry, shutdownTelemetry } from './sdk.js'; +import { + initializeTelemetry, + shutdownTelemetry, + resolveHttpOtlpUrl, +} from './sdk.js'; import { OTLPTraceExporter } from '@opentelemetry/exporter-trace-otlp-grpc'; import { OTLPLogExporter } from '@opentelemetry/exporter-logs-otlp-grpc'; import { OTLPMetricExporter } from '@opentelemetry/exporter-metrics-otlp-grpc'; @@ -27,6 +31,68 @@ vi.mock('@opentelemetry/exporter-logs-otlp-http'); vi.mock('@opentelemetry/exporter-metrics-otlp-http'); vi.mock('@opentelemetry/sdk-node'); vi.mock('./gcp-exporters.js'); +vi.mock('./log-to-span-processor.js'); + +import { LogToSpanProcessor } from './log-to-span-processor.js'; + +describe('resolveHttpOtlpUrl', () => { + it('appends signal path to base collector URL', () => { + expect(resolveHttpOtlpUrl('http://collector:4318', 'traces')).toBe( + 'http://collector:4318/v1/traces', + ); + expect(resolveHttpOtlpUrl('http://collector:4318', 'logs')).toBe( + 'http://collector:4318/v1/logs', + ); + expect(resolveHttpOtlpUrl('http://collector:4318', 'metrics')).toBe( + 'http://collector:4318/v1/metrics', + ); + }); + + it('handles trailing slash in base URL', () => { + expect(resolveHttpOtlpUrl('http://collector:4318/', 'traces')).toBe( + 'http://collector:4318/v1/traces', + ); + expect(resolveHttpOtlpUrl('http://collector:4318/', 'logs')).toBe( + 'http://collector:4318/v1/logs', + ); + }); + + it('preserves explicit full signal path URL', () => { + expect( + resolveHttpOtlpUrl('http://collector:4318/v1/traces', 'traces'), + ).toBe('http://collector:4318/v1/traces'); + expect(resolveHttpOtlpUrl('http://collector:4318/v1/logs', 'logs')).toBe( + 'http://collector:4318/v1/logs', + ); + expect( + resolveHttpOtlpUrl('http://collector:4318/v1/metrics', 'metrics'), + ).toBe('http://collector:4318/v1/metrics'); + }); + + it('appends signal path when URL has a non-signal custom path', () => { + expect( + resolveHttpOtlpUrl('http://collector:4318/custom/prefix', 'traces'), + ).toBe('http://collector:4318/custom/prefix/v1/traces'); + }); + + it('handles HTTPS URLs', () => { + expect(resolveHttpOtlpUrl('https://otel.example.com', 'logs')).toBe( + 'https://otel.example.com/v1/logs', + ); + expect(resolveHttpOtlpUrl('https://otel.example.com:4318', 'metrics')).toBe( + 'https://otel.example.com:4318/v1/metrics', + ); + }); + + it('preserves query strings when appending signal paths', () => { + expect(resolveHttpOtlpUrl('https://host/otlp?token=abc', 'traces')).toBe( + 'https://host/otlp/v1/traces?token=abc', + ); + expect( + resolveHttpOtlpUrl('https://host/otlp?token=abc&foo=bar', 'logs'), + ).toBe('https://host/otlp/v1/logs?token=abc&foo=bar'); + }); +}); describe('Telemetry SDK', () => { let mockConfig: Config; @@ -37,6 +103,9 @@ describe('Telemetry SDK', () => { getTelemetryEnabled: () => true, getTelemetryOtlpEndpoint: () => 'http://localhost:4317', getTelemetryOtlpProtocol: () => 'grpc', + getTelemetryOtlpTracesEndpoint: () => undefined, + getTelemetryOtlpLogsEndpoint: () => undefined, + getTelemetryOtlpMetricsEndpoint: () => undefined, getTelemetryTarget: () => 'local', getTelemetryUseCollector: () => false, getTelemetryOutfile: () => undefined, @@ -67,7 +136,7 @@ describe('Telemetry SDK', () => { expect(NodeSDK.prototype.start).toHaveBeenCalled(); }); - it('should use HTTP exporters when protocol is http', () => { + it('should use HTTP exporters with signal-specific paths when protocol is http', () => { vi.spyOn(mockConfig, 'getTelemetryEnabled').mockReturnValue(true); vi.spyOn(mockConfig, 'getTelemetryOtlpProtocol').mockReturnValue('http'); vi.spyOn(mockConfig, 'getTelemetryOtlpEndpoint').mockReturnValue( @@ -77,13 +146,13 @@ describe('Telemetry SDK', () => { initializeTelemetry(mockConfig); expect(OTLPTraceExporterHttp).toHaveBeenCalledWith({ - url: 'http://localhost:4318/', + url: 'http://localhost:4318/v1/traces', }); expect(OTLPLogExporterHttp).toHaveBeenCalledWith({ - url: 'http://localhost:4318/', + url: 'http://localhost:4318/v1/logs', }); expect(OTLPMetricExporterHttp).toHaveBeenCalledWith({ - url: 'http://localhost:4318/', + url: 'http://localhost:4318/v1/metrics', }); expect(NodeSDK.prototype.start).toHaveBeenCalled(); }); @@ -98,15 +167,71 @@ describe('Telemetry SDK', () => { ); }); - it('should parse HTTP endpoint correctly', () => { + it('should append signal paths to HTTP endpoint', () => { vi.spyOn(mockConfig, 'getTelemetryOtlpProtocol').mockReturnValue('http'); vi.spyOn(mockConfig, 'getTelemetryOtlpEndpoint').mockReturnValue( 'https://my-collector.com', ); initializeTelemetry(mockConfig); expect(OTLPTraceExporterHttp).toHaveBeenCalledWith( - expect.objectContaining({ url: 'https://my-collector.com/' }), + expect.objectContaining({ url: 'https://my-collector.com/v1/traces' }), + ); + expect(OTLPLogExporterHttp).toHaveBeenCalledWith( + expect.objectContaining({ url: 'https://my-collector.com/v1/logs' }), + ); + expect(OTLPMetricExporterHttp).toHaveBeenCalledWith( + expect.objectContaining({ url: 'https://my-collector.com/v1/metrics' }), + ); + }); + + it('should use per-signal endpoint overrides when provided', () => { + vi.spyOn(mockConfig, 'getTelemetryOtlpProtocol').mockReturnValue('http'); + vi.spyOn(mockConfig, 'getTelemetryOtlpEndpoint').mockReturnValue( + 'http://default-collector:4318', + ); + vi.spyOn(mockConfig, 'getTelemetryOtlpTracesEndpoint').mockReturnValue( + 'http://traces-collector:4318/v1/traces', ); + + initializeTelemetry(mockConfig); + + // Traces uses the per-signal override + expect(OTLPTraceExporterHttp).toHaveBeenCalledWith({ + url: 'http://traces-collector:4318/v1/traces', + }); + // Logs and metrics use the base endpoint with paths appended + expect(OTLPLogExporterHttp).toHaveBeenCalledWith({ + url: 'http://default-collector:4318/v1/logs', + }); + expect(OTLPMetricExporterHttp).toHaveBeenCalledWith({ + url: 'http://default-collector:4318/v1/metrics', + }); + }); + + it('should use per-signal overrides without base endpoint', () => { + vi.spyOn(mockConfig, 'getTelemetryOtlpProtocol').mockReturnValue('http'); + vi.spyOn(mockConfig, 'getTelemetryOtlpEndpoint').mockReturnValue(''); + vi.spyOn(mockConfig, 'getTelemetryOtlpTracesEndpoint').mockReturnValue( + 'http://traces-host/token/api/otlp/traces', + ); + vi.spyOn(mockConfig, 'getTelemetryOtlpMetricsEndpoint').mockReturnValue( + 'http://metrics-host/token/api/otlp/metrics', + ); + // logs has no override and no base endpoint + + initializeTelemetry(mockConfig); + + // Traces and metrics use per-signal override + expect(OTLPTraceExporterHttp).toHaveBeenCalledWith({ + url: 'http://traces-host/token/api/otlp/traces', + }); + expect(OTLPMetricExporterHttp).toHaveBeenCalledWith({ + url: 'http://metrics-host/token/api/otlp/metrics', + }); + // Logs falls back to LogToSpanProcessor (bridges logs → spans) + expect(OTLPLogExporterHttp).not.toHaveBeenCalled(); + expect(LogToSpanProcessor).toHaveBeenCalled(); + expect(NodeSDK.prototype.start).toHaveBeenCalled(); }); it('should use OTLP exporters when target is gcp but useCollector is true', () => { diff --git a/packages/core/src/telemetry/sdk.ts b/packages/core/src/telemetry/sdk.ts index 3dba2acc456..c2c7f33838a 100644 --- a/packages/core/src/telemetry/sdk.ts +++ b/packages/core/src/telemetry/sdk.ts @@ -15,18 +15,9 @@ import { CompressionAlgorithm } from '@opentelemetry/otlp-exporter-base'; import { NodeSDK } from '@opentelemetry/sdk-node'; import { SemanticResourceAttributes } from '@opentelemetry/semantic-conventions'; import { resourceFromAttributes } from '@opentelemetry/resources'; -import { - BatchSpanProcessor, - ConsoleSpanExporter, -} from '@opentelemetry/sdk-trace-node'; -import { - BatchLogRecordProcessor, - ConsoleLogRecordExporter, -} from '@opentelemetry/sdk-logs'; -import { - ConsoleMetricExporter, - PeriodicExportingMetricReader, -} from '@opentelemetry/sdk-metrics'; +import { BatchSpanProcessor } from '@opentelemetry/sdk-trace-node'; +import { BatchLogRecordProcessor } from '@opentelemetry/sdk-logs'; +import { PeriodicExportingMetricReader } from '@opentelemetry/sdk-metrics'; import { HttpInstrumentation } from '@opentelemetry/instrumentation-http'; import type { Config } from '../config/config.js'; import { SERVICE_NAME } from './constants.js'; @@ -37,9 +28,44 @@ import { FileSpanExporter, } from './file-exporters.js'; import { createDebugLogger } from '../utils/debugLogger.js'; +import { LogToSpanProcessor } from './log-to-span-processor.js'; // For troubleshooting, set the log level to DiagLogLevel.DEBUG -diag.setLogger(new DiagConsoleLogger(), DiagLogLevel.INFO); +diag.setLogger(new DiagConsoleLogger(), DiagLogLevel.ERROR); + +/** + * Standard OTLP HTTP signal-specific paths per the OpenTelemetry specification. + * gRPC uses service-based routing so no path appending is needed. + */ +const OTLP_SIGNAL_PATHS = { + traces: 'v1/traces', + logs: 'v1/logs', + metrics: 'v1/metrics', +} as const; + +type OtlpSignal = keyof typeof OTLP_SIGNAL_PATHS; + +/** + * Resolve the final URL for an HTTP OTLP exporter. + * + * - If the URL path already ends with the signal-specific path (e.g., /v1/traces), + * use it as-is. This supports explicit full-path configuration. + * - Otherwise, append the signal-specific path to the base URL. + */ +export function resolveHttpOtlpUrl( + baseEndpoint: string, + signal: OtlpSignal, +): string { + const signalPath = OTLP_SIGNAL_PATHS[signal]; + const url = new URL(baseEndpoint); + const normalizedPath = url.pathname.replace(/\/+$/, ''); + if (normalizedPath.endsWith(signalPath)) { + return url.href; + } + // Append the signal path to the URL pathname, preserving query/hash. + url.pathname = normalizedPath + '/' + signalPath; + return url.href; +} let sdk: NodeSDK | undefined; let telemetryInitialized = false; @@ -73,6 +99,21 @@ function parseOtlpEndpoint( } } +/** + * Validate a URL string. Returns the URL if valid, undefined otherwise. + * Logs an error for invalid URLs instead of throwing. + */ +function validateUrl(url: string | undefined): string | undefined { + if (!url) return undefined; + try { + new URL(url); + return url; + } catch { + diag.error('Invalid OTLP signal endpoint URL, skipping:', url); + return undefined; + } +} + export function initializeTelemetry(config: Config): void { if (telemetryInitialized || !config.getTelemetryEnabled()) { return; @@ -89,51 +130,93 @@ export function initializeTelemetry(config: Config): void { const otlpProtocol = config.getTelemetryOtlpProtocol(); const parsedEndpoint = parseOtlpEndpoint(otlpEndpoint, otlpProtocol); const telemetryOutfile = config.getTelemetryOutfile(); - const useOtlp = !!parsedEndpoint && !telemetryOutfile; + const hasPerSignalEndpoint = + !!config.getTelemetryOtlpTracesEndpoint() || + !!config.getTelemetryOtlpLogsEndpoint() || + !!config.getTelemetryOtlpMetricsEndpoint(); + const useOtlp = + (!!parsedEndpoint || hasPerSignalEndpoint) && !telemetryOutfile; let spanExporter: | OTLPTraceExporter | OTLPTraceExporterHttp | FileSpanExporter - | ConsoleSpanExporter; + | undefined; let logExporter: | OTLPLogExporter | OTLPLogExporterHttp | FileLogExporter - | ConsoleLogRecordExporter; - let metricReader: PeriodicExportingMetricReader; + | undefined; + let metricReader: PeriodicExportingMetricReader | undefined; + let logToSpanProcessor: LogToSpanProcessor | undefined; if (useOtlp) { if (otlpProtocol === 'http') { - spanExporter = new OTLPTraceExporterHttp({ - url: parsedEndpoint, - }); - logExporter = new OTLPLogExporterHttp({ - url: parsedEndpoint, - }); - metricReader = new PeriodicExportingMetricReader({ - exporter: new OTLPMetricExporterHttp({ - url: parsedEndpoint, - }), - exportIntervalMillis: 10000, - }); + const tracesUrl = validateUrl( + config.getTelemetryOtlpTracesEndpoint() ?? + (parsedEndpoint + ? resolveHttpOtlpUrl(parsedEndpoint, 'traces') + : undefined), + ); + const logsUrl = validateUrl( + config.getTelemetryOtlpLogsEndpoint() ?? + (parsedEndpoint + ? resolveHttpOtlpUrl(parsedEndpoint, 'logs') + : undefined), + ); + const metricsUrl = validateUrl( + config.getTelemetryOtlpMetricsEndpoint() ?? + (parsedEndpoint + ? resolveHttpOtlpUrl(parsedEndpoint, 'metrics') + : undefined), + ); + + debugLogger.debug( + `OTLP HTTP endpoints: traces=${tracesUrl ?? 'none'}, logs=${logsUrl ?? 'none'}, metrics=${metricsUrl ?? 'none'}`, + ); + + if (tracesUrl) { + spanExporter = new OTLPTraceExporterHttp({ url: tracesUrl }); + } + if (logsUrl) { + logExporter = new OTLPLogExporterHttp({ url: logsUrl }); + } else if (tracesUrl) { + // Bridge: no logs endpoint but traces endpoint exists. + // Convert log records to spans and export via a dedicated trace exporter. + logToSpanProcessor = new LogToSpanProcessor( + new OTLPTraceExporterHttp({ url: tracesUrl }), + ); + } + if (metricsUrl) { + metricReader = new PeriodicExportingMetricReader({ + exporter: new OTLPMetricExporterHttp({ url: metricsUrl }), + exportIntervalMillis: 10000, + }); + } } else { - // grpc - spanExporter = new OTLPTraceExporter({ - url: parsedEndpoint, - compression: CompressionAlgorithm.GZIP, - }); - logExporter = new OTLPLogExporter({ - url: parsedEndpoint, - compression: CompressionAlgorithm.GZIP, - }); - metricReader = new PeriodicExportingMetricReader({ - exporter: new OTLPMetricExporter({ + // grpc — per-signal endpoints are not supported with gRPC protocol. + if (!parsedEndpoint) { + debugLogger.warn( + 'Per-signal OTLP endpoints are only supported with HTTP protocol. ' + + 'Set otlpProtocol to "http" or provide a base otlpEndpoint for gRPC.', + ); + } else { + spanExporter = new OTLPTraceExporter({ + url: parsedEndpoint, + compression: CompressionAlgorithm.GZIP, + }); + logExporter = new OTLPLogExporter({ url: parsedEndpoint, compression: CompressionAlgorithm.GZIP, - }), - exportIntervalMillis: 10000, - }); + }); + metricReader = new PeriodicExportingMetricReader({ + exporter: new OTLPMetricExporter({ + url: parsedEndpoint, + compression: CompressionAlgorithm.GZIP, + }), + exportIntervalMillis: 10000, + }); + } } } else if (telemetryOutfile) { spanExporter = new FileSpanExporter(telemetryOutfile); @@ -142,20 +225,18 @@ export function initializeTelemetry(config: Config): void { exporter: new FileMetricExporter(telemetryOutfile), exportIntervalMillis: 10000, }); - } else { - spanExporter = new ConsoleSpanExporter(); - logExporter = new ConsoleLogRecordExporter(); - metricReader = new PeriodicExportingMetricReader({ - exporter: new ConsoleMetricExporter(), - exportIntervalMillis: 10000, - }); } + // If no exporter is configured for a signal, it is silently skipped. sdk = new NodeSDK({ resource, - spanProcessors: [new BatchSpanProcessor(spanExporter)], - logRecordProcessors: [new BatchLogRecordProcessor(logExporter)], - metricReader, + spanProcessors: spanExporter ? [new BatchSpanProcessor(spanExporter)] : [], + logRecordProcessors: logExporter + ? [new BatchLogRecordProcessor(logExporter)] + : logToSpanProcessor + ? [logToSpanProcessor] + : [], + ...(metricReader && { metricReader }), instrumentations: [new HttpInstrumentation()], }); From f1878c66c608819c9307727aebffd70e19617490 Mon Sep 17 00:00:00 2001 From: "jinye.djy" Date: Fri, 1 May 2026 09:49:37 +0800 Subject: [PATCH 2/7] fix(telemetry): fix TS noPropertyAccessFromIndexSignature errors in tests Use typed ExportedSpan interface and bracket notation for index signature properties to satisfy strict TypeScript checks in CI. Co-authored-by: Qwen-Coder --- .../telemetry/log-to-span-processor.test.ts | 24 +++++++++++++------ 1 file changed, 17 insertions(+), 7 deletions(-) diff --git a/packages/core/src/telemetry/log-to-span-processor.test.ts b/packages/core/src/telemetry/log-to-span-processor.test.ts index 20b9c9fb396..51c30541fa6 100644 --- a/packages/core/src/telemetry/log-to-span-processor.test.ts +++ b/packages/core/src/telemetry/log-to-span-processor.test.ts @@ -5,15 +5,25 @@ */ import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; -import { SpanKind, SpanStatusCode } from '@opentelemetry/api'; +import { SpanKind, SpanStatusCode, type HrTime } from '@opentelemetry/api'; import { LogToSpanProcessor } from './log-to-span-processor.js'; import type { ReadableLogRecord } from '@opentelemetry/sdk-logs'; import type { SpanExporter } from '@opentelemetry/sdk-trace-base'; +interface ExportedSpan { + name: string; + kind: number; + spanContext: () => { traceId: string; spanId: string; traceFlags: number }; + startTime: HrTime; + endTime: HrTime; + attributes: Record; + status: { code: number; message?: string }; +} + describe('LogToSpanProcessor', () => { let processor: LogToSpanProcessor; let mockExporter: SpanExporter; - let exportedSpans: Array>; + let exportedSpans: ExportedSpan[]; beforeEach(() => { exportedSpans = []; @@ -50,9 +60,9 @@ describe('LogToSpanProcessor', () => { const span = exportedSpans[0]; expect(span.name).toBe('test event'); expect(span.kind).toBe(SpanKind.INTERNAL); - expect(span.attributes.key1).toBe('value1'); - expect(span.attributes.key2).toBe(42); - expect(span.attributes.key3).toBe(true); + expect(span.attributes['key1']).toBe('value1'); + expect(span.attributes['key2']).toBe(42); + expect(span.attributes['key3']).toBe(true); expect(span.attributes['log.bridge']).toBe(true); expect(span.startTime).toEqual([1000, 500000000]); // Instant span: end time == start time @@ -107,7 +117,7 @@ describe('LogToSpanProcessor', () => { processor.onEmit(logRecord); await processor.forceFlush(); - expect(exportedSpans[0].attributes.metadata).toBe('{"nested":true}'); + expect(exportedSpans[0].attributes['metadata']).toBe('{"nested":true}'); }); it('skips null and undefined attributes', async () => { @@ -125,7 +135,7 @@ describe('LogToSpanProcessor', () => { await processor.forceFlush(); const attrs = exportedSpans[0].attributes; - expect(attrs.valid).toBe('yes'); + expect(attrs['valid']).toBe('yes'); expect(attrs).not.toHaveProperty('nullVal'); expect(attrs).not.toHaveProperty('undefinedVal'); expect(attrs['log.bridge']).toBe(true); From 032aa789657a136149a9a6a15f6cd75e8c0c243d Mon Sep 17 00:00:00 2001 From: "jinye.djy" Date: Fri, 1 May 2026 10:00:57 +0800 Subject: [PATCH 3/7] fix(telemetry): replace MD5 with SHA-256 for traceId derivation CodeQL flagged MD5 as a weak cryptographic algorithm when used with session.id (considered sensitive data). Switch to SHA-256 truncated to 32 hex chars to satisfy CodeQL while maintaining the same traceId format required by the OTel specification. Co-authored-by: Qwen-Coder --- packages/core/src/telemetry/log-to-span-processor.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/packages/core/src/telemetry/log-to-span-processor.ts b/packages/core/src/telemetry/log-to-span-processor.ts index ecb82373c12..2c36903675a 100644 --- a/packages/core/src/telemetry/log-to-span-processor.ts +++ b/packages/core/src/telemetry/log-to-span-processor.ts @@ -170,9 +170,11 @@ function randomHexString(length: number): string { * Derive a deterministic 32-char hex traceId from a session ID. * All events in the same session will share this traceId, * making them appear under a single trace in the backend. + * Uses SHA-256 truncated to 32 hex chars (128 bits) to match the + * OTel trace ID format. */ function deriveTraceId(sessionId: string): string { - return createHash('md5').update(sessionId).digest('hex'); + return createHash('sha256').update(sessionId).digest('hex').slice(0, 32); } /** From 89fb6ad9f12d83b234950d95499db5b1acc60614 Mon Sep 17 00:00:00 2001 From: "jinye.djy" Date: Fri, 1 May 2026 15:17:50 +0800 Subject: [PATCH 4/7] fix(telemetry): address review feedback for LogToSpanProcessor robustness - Wrap JSON.stringify in try/catch to handle circular refs and BigInt - Add export timeout (30s) and try/catch to prevent hung shutdown - Track in-flight exports to avoid interval-vs-shutdown race condition - Fix deriveSpanStatus: use truthy checks (!!), drop success===false heuristic since declined tool calls are normal, not errors - Enforce http(s) scheme in validateUrl to reject file:/javascript: URLs - Change DiagLogLevel from ERROR to WARN to preserve operational diagnostics - Preserve logRecord.instrumentationScope instead of hardcoding - Forward severityNumber/severityText as span attributes - Add tests for circular refs, error status edge cases, severity Co-authored-by: Qwen-Coder --- .../telemetry/log-to-span-processor.test.ts | 112 +++++++++++++----- .../src/telemetry/log-to-span-processor.ts | 86 +++++++++++--- packages/core/src/telemetry/sdk.ts | 16 ++- 3 files changed, 164 insertions(+), 50 deletions(-) diff --git a/packages/core/src/telemetry/log-to-span-processor.test.ts b/packages/core/src/telemetry/log-to-span-processor.test.ts index 51c30541fa6..9996b78a5b3 100644 --- a/packages/core/src/telemetry/log-to-span-processor.test.ts +++ b/packages/core/src/telemetry/log-to-span-processor.test.ts @@ -35,7 +35,7 @@ describe('LogToSpanProcessor', () => { shutdown: vi.fn().mockResolvedValue(undefined), forceFlush: vi.fn().mockResolvedValue(undefined), } as unknown as SpanExporter; - processor = new LogToSpanProcessor(mockExporter, 60000); // long interval to avoid auto-flush + processor = new LogToSpanProcessor(mockExporter, 60000); }); afterEach(async () => { @@ -46,11 +46,7 @@ describe('LogToSpanProcessor', () => { const logRecord = { body: 'test event', hrTime: [1000, 500000000] as [number, number], - attributes: { - key1: 'value1', - key2: 42, - key3: true, - }, + attributes: { key1: 'value1', key2: 42, key3: true }, } as unknown as ReadableLogRecord; processor.onEmit(logRecord); @@ -65,7 +61,6 @@ describe('LogToSpanProcessor', () => { expect(span.attributes['key3']).toBe(true); expect(span.attributes['log.bridge']).toBe(true); expect(span.startTime).toEqual([1000, 500000000]); - // Instant span: end time == start time expect(span.endTime).toEqual([1000, 500000000]); expect(span.status.code).toBe(SpanStatusCode.OK); }); @@ -74,44 +69,33 @@ describe('LogToSpanProcessor', () => { const logRecord = { body: 'api response', hrTime: [1000, 0] as [number, number], - attributes: { - duration_ms: 250, - model: 'test-model', - }, + attributes: { duration_ms: 250 }, } as unknown as ReadableLogRecord; processor.onEmit(logRecord); await processor.forceFlush(); - const span = exportedSpans[0]; - // 250ms = 250_000_000 nanoseconds - expect(span.endTime).toEqual([1000, 250000000]); + expect(exportedSpans[0].endTime).toEqual([1000, 250000000]); }); it('handles duration_ms that causes second rollover', async () => { const logRecord = { body: 'long operation', hrTime: [1000, 900000000] as [number, number], - attributes: { - duration_ms: 500, - }, + attributes: { duration_ms: 500 }, } as unknown as ReadableLogRecord; processor.onEmit(logRecord); await processor.forceFlush(); - const span = exportedSpans[0]; - // 900_000_000 + 500_000_000 = 1_400_000_000 → [1001, 400000000] - expect(span.endTime).toEqual([1001, 400000000]); + expect(exportedSpans[0].endTime).toEqual([1001, 400000000]); }); it('serializes object attributes to JSON', async () => { const logRecord = { body: 'event with object', hrTime: [1000, 0] as [number, number], - attributes: { - metadata: { nested: true }, - }, + attributes: { metadata: { nested: true } }, } as unknown as ReadableLogRecord; processor.onEmit(logRecord); @@ -120,15 +104,26 @@ describe('LogToSpanProcessor', () => { expect(exportedSpans[0].attributes['metadata']).toBe('{"nested":true}'); }); + it('handles unserializable object attributes safely', async () => { + const circular: Record = {}; + circular['self'] = circular; + const logRecord = { + body: 'event', + hrTime: [1000, 0] as [number, number], + attributes: { bad: circular }, + } as unknown as ReadableLogRecord; + + processor.onEmit(logRecord); + await processor.forceFlush(); + + expect(exportedSpans[0].attributes['bad']).toBe('[unserializable]'); + }); + it('skips null and undefined attributes', async () => { const logRecord = { body: 'event', hrTime: [1000, 0] as [number, number], - attributes: { - valid: 'yes', - nullVal: null, - undefinedVal: undefined, - }, + attributes: { valid: 'yes', nullVal: null, undefinedVal: undefined }, } as unknown as ReadableLogRecord; processor.onEmit(logRecord); @@ -195,9 +190,7 @@ describe('LogToSpanProcessor', () => { const ctx1 = exportedSpans[0].spanContext(); const ctx2 = exportedSpans[1].spanContext(); - // Same session → same traceId expect(ctx1.traceId).toBe(ctx2.traceId); - // Different spanIds expect(ctx1.spanId).not.toBe(ctx2.spanId); }); @@ -222,6 +215,65 @@ describe('LogToSpanProcessor', () => { expect(ctx1.traceId).not.toBe(ctx2.traceId); }); + it('sets ERROR status for truthy error attributes', async () => { + const logRecord = { + body: 'api error', + hrTime: [1000, 0] as [number, number], + attributes: { + error_message: 'connection refused', + error_type: 'NETWORK', + }, + } as unknown as ReadableLogRecord; + + processor.onEmit(logRecord); + await processor.forceFlush(); + + expect(exportedSpans[0].status.code).toBe(SpanStatusCode.ERROR); + expect(exportedSpans[0].status.message).toBe('connection refused'); + }); + + it('does not set ERROR for success: false (normal decline)', async () => { + const logRecord = { + body: 'tool call declined', + hrTime: [1000, 0] as [number, number], + attributes: { success: false, function_name: 'bash' }, + } as unknown as ReadableLogRecord; + + processor.onEmit(logRecord); + await processor.forceFlush(); + + expect(exportedSpans[0].status.code).toBe(SpanStatusCode.OK); + }); + + it('does not set ERROR for falsy error attributes', async () => { + const logRecord = { + body: 'ok event', + hrTime: [1000, 0] as [number, number], + attributes: { error: null, error_message: '', error_type: '' }, + } as unknown as ReadableLogRecord; + + processor.onEmit(logRecord); + await processor.forceFlush(); + + expect(exportedSpans[0].status.code).toBe(SpanStatusCode.OK); + }); + + it('preserves severity attributes', async () => { + const logRecord = { + body: 'event', + hrTime: [1000, 0] as [number, number], + attributes: {}, + severityNumber: 9, + severityText: 'INFO', + } as unknown as ReadableLogRecord; + + processor.onEmit(logRecord); + await processor.forceFlush(); + + expect(exportedSpans[0].attributes['log.severity_number']).toBe(9); + expect(exportedSpans[0].attributes['log.severity_text']).toBe('INFO'); + }); + it('shutdown flushes remaining spans and shuts down exporter', async () => { const logRecord = { body: 'final event', diff --git a/packages/core/src/telemetry/log-to-span-processor.ts b/packages/core/src/telemetry/log-to-span-processor.ts index 2c36903675a..98fe737465a 100644 --- a/packages/core/src/telemetry/log-to-span-processor.ts +++ b/packages/core/src/telemetry/log-to-span-processor.ts @@ -19,6 +19,8 @@ import { createHash } from 'node:crypto'; import { SERVICE_NAME } from './constants.js'; +const EXPORT_TIMEOUT_MS = 30_000; + /** * A LogRecordProcessor that converts each OTel log record into a span * and exports it directly through the provided SpanExporter. @@ -35,6 +37,7 @@ import { SERVICE_NAME } from './constants.js'; export class LogToSpanProcessor implements LogRecordProcessor { private buffer: ReadableSpanLike[] = []; private flushTimer: ReturnType | undefined; + private inFlightExport: Promise | undefined; private readonly flushIntervalMs: number; constructor( @@ -57,12 +60,22 @@ export class LogToSpanProcessor implements LogRecordProcessor { for (const [key, value] of Object.entries(logRecord.attributes)) { if (value !== undefined && value !== null) { attributes[key] = - typeof value === 'object' ? JSON.stringify(value) : value; + typeof value === 'object' + ? safeStringify(value) + : (value as string | number | boolean); } } } attributes['log.bridge'] = true; + // Preserve severity so downstream queries can filter by log level. + if (logRecord.severityNumber !== undefined) { + attributes['log.severity_number'] = logRecord.severityNumber; + } + if (logRecord.severityText) { + attributes['log.severity_text'] = logRecord.severityText; + } + let endTime = startTime; const durationMs = logRecord.attributes?.['duration_ms']; if (typeof durationMs === 'number' && durationMs > 0) { @@ -96,7 +109,7 @@ export class LogToSpanProcessor implements LogRecordProcessor { events: [], links: [], resource: logRecord.resource ?? resourceFromAttributes({}), - instrumentationScope: { + instrumentationScope: logRecord.instrumentationScope ?? { name: SERVICE_NAME, version: '', }, @@ -112,16 +125,38 @@ export class LogToSpanProcessor implements LogRecordProcessor { private flush(): Promise { if (this.buffer.length === 0) return Promise.resolve(); const spans = this.buffer.splice(0); - return new Promise((resolve) => { - this.spanExporter.export(spans as unknown as ReadableSpan[], (result) => { - if (result.code !== 0) { - process.stderr.write( - `[LogToSpan] export failed: code=${result.code} error=${result.error?.message ?? 'unknown'}\n`, - ); - } + const exportPromise = new Promise((resolve) => { + const timeout = setTimeout(() => { + process.stderr.write( + `[LogToSpan] export timeout after ${EXPORT_TIMEOUT_MS}ms\n`, + ); + resolve(); + }, EXPORT_TIMEOUT_MS); + timeout.unref(); + + try { + this.spanExporter.export( + spans as unknown as ReadableSpan[], + (result) => { + clearTimeout(timeout); + if (result.code !== 0) { + process.stderr.write( + `[LogToSpan] export failed: code=${result.code} error=${result.error?.message ?? 'unknown'}\n`, + ); + } + resolve(); + }, + ); + } catch (err) { + clearTimeout(timeout); + process.stderr.write( + `[LogToSpan] export threw: ${err instanceof Error ? err.message : String(err)}\n`, + ); resolve(); - }); + } }); + this.inFlightExport = exportPromise; + return exportPromise; } async shutdown(): Promise { @@ -129,11 +164,18 @@ export class LogToSpanProcessor implements LogRecordProcessor { clearInterval(this.flushTimer); this.flushTimer = undefined; } + // Wait for any in-flight interval-triggered export before final flush. + if (this.inFlightExport) { + await this.inFlightExport; + } await this.flush(); await this.spanExporter.shutdown(); } async forceFlush(): Promise { + if (this.inFlightExport) { + await this.inFlightExport; + } await this.flush(); await this.spanExporter.forceFlush?.(); } @@ -160,6 +202,18 @@ interface ReadableSpanLike { recordException: () => void; } +/** + * Safely stringify an object value for use as a span attribute. + * Handles circular references and BigInt without throwing. + */ +function safeStringify(value: unknown): string { + try { + return JSON.stringify(value); + } catch { + return '[unserializable]'; + } +} + function randomHexString(length: number): string { const bytes = new Uint8Array(length / 2); crypto.getRandomValues(bytes); @@ -179,19 +233,17 @@ function deriveTraceId(sessionId: string): string { /** * Derive span status from log record attributes. - * Marks the span as ERROR when common error indicators are present. + * Marks the span as ERROR when explicit error indicators are present + * (truthy `error`, `error_message`, or `error_type` attributes). + * Does NOT treat `success: false` as an error — declined/cancelled + * operations are a normal outcome, not failures. */ function deriveSpanStatus(attrs: Record | undefined): { code: SpanStatusCode; message?: string; } { if (!attrs) return { code: SpanStatusCode.OK }; - if ( - attrs['success'] === false || - attrs['error'] !== undefined || - attrs['error_message'] !== undefined || - attrs['error_type'] !== undefined - ) { + if (!!attrs['error'] || !!attrs['error_message'] || !!attrs['error_type']) { const msg = String( attrs['error_message'] ?? attrs['error'] ?? attrs['error_type'] ?? '', ); diff --git a/packages/core/src/telemetry/sdk.ts b/packages/core/src/telemetry/sdk.ts index c2c7f33838a..0ffb88fa7db 100644 --- a/packages/core/src/telemetry/sdk.ts +++ b/packages/core/src/telemetry/sdk.ts @@ -31,7 +31,7 @@ import { createDebugLogger } from '../utils/debugLogger.js'; import { LogToSpanProcessor } from './log-to-span-processor.js'; // For troubleshooting, set the log level to DiagLogLevel.DEBUG -diag.setLogger(new DiagConsoleLogger(), DiagLogLevel.ERROR); +diag.setLogger(new DiagConsoleLogger(), DiagLogLevel.WARN); /** * Standard OTLP HTTP signal-specific paths per the OpenTelemetry specification. @@ -100,13 +100,23 @@ function parseOtlpEndpoint( } /** - * Validate a URL string. Returns the URL if valid, undefined otherwise. + * Validate a URL string. Returns the URL if valid http(s), undefined otherwise. * Logs an error for invalid URLs instead of throwing. */ function validateUrl(url: string | undefined): string | undefined { if (!url) return undefined; try { - new URL(url); + const parsed = new URL(url); + if (parsed.protocol !== 'http:' && parsed.protocol !== 'https:') { + diag.error( + `OTLP endpoint must use http or https, got ${parsed.protocol}`, + ); + return undefined; + } + if (!parsed.hostname) { + diag.error('OTLP endpoint missing hostname'); + return undefined; + } return url; } catch { diag.error('Invalid OTLP signal endpoint URL, skipping:', url); From 9530b7c9e3bae4cb8a4cf3798c41dd87e19e4a82 Mon Sep 17 00:00:00 2001 From: "jinye.djy" Date: Fri, 1 May 2026 17:51:33 +0800 Subject: [PATCH 5/7] fix(telemetry): flush sdk shutdown through cleanup Remove async process exit handlers from telemetry initialization and route SDK shutdown through Config cleanup so normal CLI exit paths await pending telemetry exports. Keep shutdown idempotent while an SDK shutdown is in flight. Co-authored-by: Qwen-Coder --- packages/core/src/config/config.test.ts | 14 ++++++++++ packages/core/src/config/config.ts | 14 +++++++--- packages/core/src/telemetry/sdk.test.ts | 31 +++++++++++++++++++++ packages/core/src/telemetry/sdk.ts | 36 ++++++++++++------------- 4 files changed, 73 insertions(+), 22 deletions(-) diff --git a/packages/core/src/config/config.test.ts b/packages/core/src/config/config.test.ts index 62663eced3b..914ae42017a 100644 --- a/packages/core/src/config/config.test.ts +++ b/packages/core/src/config/config.test.ts @@ -15,6 +15,7 @@ import { DEFAULT_TELEMETRY_TARGET, DEFAULT_OTLP_ENDPOINT, QwenLogger, + shutdownTelemetry, } from '../telemetry/index.js'; import type { ContentGenerator, @@ -178,6 +179,7 @@ vi.mock('../telemetry/index.js', async (importOriginal) => { return { ...actual, initializeTelemetry: vi.fn(), + shutdownTelemetry: vi.fn().mockResolvedValue(undefined), uiTelemetryService: { getLastPromptTokenCount: vi.fn(), }, @@ -848,6 +850,18 @@ describe('Server Config (config.ts)', () => { expect(config.getTelemetryEnabled()).toBe(true); }); + it('Config shutdown should flush telemetry even before initialization completes', async () => { + const paramsWithTelemetry: ConfigParameters = { + ...baseParams, + telemetry: { enabled: true }, + }; + const config = new Config(paramsWithTelemetry); + + await config.shutdown(); + + expect(shutdownTelemetry).toHaveBeenCalledTimes(1); + }); + it('Config constructor should set telemetry to false when provided as false', () => { const paramsWithTelemetry: ConfigParameters = { ...baseParams, diff --git a/packages/core/src/config/config.ts b/packages/core/src/config/config.ts index c8cadc519c8..cce3a505d41 100644 --- a/packages/core/src/config/config.ts +++ b/packages/core/src/config/config.ts @@ -66,6 +66,7 @@ import { DEFAULT_OTLP_ENDPOINT, DEFAULT_TELEMETRY_TARGET, initializeTelemetry, + shutdownTelemetry, logStartSession, logRipgrepFallback, RipgrepFallbackEvent, @@ -1617,11 +1618,12 @@ export class Config { * It handles the case where initialization was not completed. */ async shutdown(): Promise { - if (!this.initialized) { - // Nothing to clean up if not initialized - return; - } try { + if (!this.initialized) { + // Nothing else to clean up if not initialized. + return; + } + // Finalize the current session's metadata before cleanup, then drain // the async write queue so no records are lost on exit. try { @@ -1644,6 +1646,10 @@ export class Config { } catch (error) { // Log but don't throw - cleanup should be best-effort this.debugLogger.error('Error during Config shutdown:', error); + } finally { + if (this.telemetrySettings.enabled) { + await shutdownTelemetry(); + } } } diff --git a/packages/core/src/telemetry/sdk.test.ts b/packages/core/src/telemetry/sdk.test.ts index 2d760597b28..89fdb9ff1a9 100644 --- a/packages/core/src/telemetry/sdk.test.ts +++ b/packages/core/src/telemetry/sdk.test.ts @@ -8,6 +8,7 @@ import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; import type { Config } from '../config/config.js'; import { initializeTelemetry, + isTelemetrySdkInitialized, shutdownTelemetry, resolveHttpOtlpUrl, } from './sdk.js'; @@ -270,4 +271,34 @@ describe('Telemetry SDK', () => { expect(OTLPMetricExporterHttp).not.toHaveBeenCalled(); expect(NodeSDK.prototype.start).toHaveBeenCalled(); }); + + it('should not register async process shutdown handlers', () => { + const processOnSpy = vi.spyOn(process, 'on'); + try { + initializeTelemetry(mockConfig); + + expect(processOnSpy).not.toHaveBeenCalledWith( + 'SIGTERM', + expect.any(Function), + ); + expect(processOnSpy).not.toHaveBeenCalledWith( + 'SIGINT', + expect.any(Function), + ); + expect(processOnSpy).not.toHaveBeenCalledWith( + 'exit', + expect.any(Function), + ); + } finally { + processOnSpy.mockRestore(); + } + }); + + it('should mark telemetry uninitialized after shutdown', async () => { + initializeTelemetry(mockConfig); + + await shutdownTelemetry(); + + expect(isTelemetrySdkInitialized()).toBe(false); + }); }); diff --git a/packages/core/src/telemetry/sdk.ts b/packages/core/src/telemetry/sdk.ts index 0ffb88fa7db..a70a3f4e28e 100644 --- a/packages/core/src/telemetry/sdk.ts +++ b/packages/core/src/telemetry/sdk.ts @@ -69,6 +69,7 @@ export function resolveHttpOtlpUrl( let sdk: NodeSDK | undefined; let telemetryInitialized = false; +let telemetryShutdownPromise: Promise | undefined; export function isTelemetrySdkInitialized(): boolean { return telemetryInitialized; @@ -258,29 +259,28 @@ export function initializeTelemetry(config: Config): void { } catch (error) { debugLogger.error('Error starting OpenTelemetry SDK:', error); } - - process.on('SIGTERM', () => { - shutdownTelemetry(); - }); - process.on('SIGINT', () => { - shutdownTelemetry(); - }); - process.on('exit', () => { - shutdownTelemetry(); - }); } export async function shutdownTelemetry(): Promise { + if (telemetryShutdownPromise) { + return telemetryShutdownPromise; + } if (!telemetryInitialized || !sdk) { return; } + const currentSdk = sdk; const debugLogger = createDebugLogger('OTEL'); - try { - await sdk.shutdown(); - debugLogger.debug('OpenTelemetry SDK shut down successfully.'); - } catch (error) { - debugLogger.error('Error shutting down SDK:', error); - } finally { - telemetryInitialized = false; - } + telemetryShutdownPromise = (async () => { + try { + await currentSdk.shutdown(); + debugLogger.debug('OpenTelemetry SDK shut down successfully.'); + } catch (error) { + debugLogger.error('Error shutting down SDK:', error); + } finally { + telemetryInitialized = false; + sdk = undefined; + telemetryShutdownPromise = undefined; + } + })(); + return telemetryShutdownPromise; } From ee6608e95933e55f5c09f67fc2f41d273fc29bfc Mon Sep 17 00:00:00 2001 From: "jinye.djy" Date: Fri, 1 May 2026 21:17:21 +0800 Subject: [PATCH 6/7] fix(telemetry): harden bridged log shutdown Co-authored-by: Qwen-Coder --- packages/cli/src/gemini.test.tsx | 93 ++++++++++++++++++ packages/cli/src/gemini.tsx | 46 +++++++-- packages/core/src/config/config.test.ts | 19 +++- packages/core/src/config/config.ts | 3 +- .../telemetry/log-to-span-processor.test.ts | 96 +++++++++++++++++++ .../src/telemetry/log-to-span-processor.ts | 34 ++++++- packages/core/src/telemetry/sdk.ts | 3 +- 7 files changed, 278 insertions(+), 16 deletions(-) diff --git a/packages/cli/src/gemini.test.tsx b/packages/cli/src/gemini.test.tsx index 615e99ace92..e30995bb55f 100644 --- a/packages/cli/src/gemini.test.tsx +++ b/packages/cli/src/gemini.test.tsx @@ -493,11 +493,19 @@ describe('gemini.tsx main function kitty protocol', () => { let setRawModeSpy: MockInstance< (mode: boolean) => NodeJS.ReadStream & { fd: 0 } >; + let initialSigintListeners: NodeJS.SignalsListener[]; + let initialSigtermListeners: NodeJS.SignalsListener[]; beforeEach(() => { // Set no relaunch in tests since process spawning causing issues in tests originalEnvNoRelaunch = process.env['QWEN_CODE_NO_RELAUNCH']; process.env['QWEN_CODE_NO_RELAUNCH'] = 'true'; + initialSigintListeners = process.listeners( + 'SIGINT', + ) as NodeJS.SignalsListener[]; + initialSigtermListeners = process.listeners( + 'SIGTERM', + ) as NodeJS.SignalsListener[]; // eslint-disable-next-line @typescript-eslint/no-explicit-any if (!(process.stdin as any).setRawMode) { @@ -517,12 +525,24 @@ describe('gemini.tsx main function kitty protocol', () => { }); afterEach(() => { + for (const listener of process.listeners('SIGINT')) { + if (!initialSigintListeners.includes(listener)) { + process.removeListener('SIGINT', listener as NodeJS.SignalsListener); + } + } + for (const listener of process.listeners('SIGTERM')) { + if (!initialSigtermListeners.includes(listener)) { + process.removeListener('SIGTERM', listener as NodeJS.SignalsListener); + } + } + // Restore original env variables if (originalEnvNoRelaunch !== undefined) { process.env['QWEN_CODE_NO_RELAUNCH'] = originalEnvNoRelaunch; } else { delete process.env['QWEN_CODE_NO_RELAUNCH']; } + vi.restoreAllMocks(); }); it('should call setRawMode and detectAndEnableKittyProtocol when isInteractive is true', async () => { @@ -618,6 +638,79 @@ describe('gemini.tsx main function kitty protocol', () => { expect(setRawModeSpy).toHaveBeenCalledWith(true); expect(detectAndEnableKittyProtocol).toHaveBeenCalledTimes(1); }); + + it('should run cleanup before exiting on interactive SIGTERM', async () => { + const { loadCliConfig, parseArguments } = await import( + './config/config.js' + ); + const { loadSettings } = await import('./config/settings.js'); + const cleanupModule = await import('./utils/cleanup.js'); + const signalHandlers = new Map void>(); + const processOnceSpy = vi.spyOn(process, 'once').mockImplementation((( + eventName: string | symbol, + listener: (...args: unknown[]) => void, + ) => { + if (eventName === 'SIGTERM' || eventName === 'SIGINT') { + signalHandlers.set(eventName, listener); + } + return process; + }) as typeof process.once); + const processExitSpy = vi + .spyOn(process, 'exit') + .mockImplementation((() => undefined) as unknown as typeof process.exit); + const runExitCleanupMock = vi.mocked(cleanupModule.runExitCleanup); + runExitCleanupMock.mockResolvedValue(undefined); + + vi.mocked(loadCliConfig).mockResolvedValue({ + isInteractive: () => true, + getQuestion: () => '', + getSandbox: () => false, + getDebugMode: () => false, + getListExtensions: () => false, + getMcpServers: () => ({}), + initialize: vi.fn(), + getIdeMode: () => false, + getExperimentalZedIntegration: () => false, + getScreenReader: () => false, + getGeminiMdFileCount: () => 0, + getWarnings: () => [], + getModelsConfig: () => ({ + getCurrentAuthType: () => null, + getGenerationConfig: () => ({}), + }), + getProxy: () => undefined, + getUsageStatisticsEnabled: () => true, + getSessionId: () => 'test-session-id', + } as unknown as Config); + vi.mocked(loadSettings).mockReturnValue({ + errors: [], + merged: { + advanced: {}, + security: { auth: {} }, + ui: {}, + }, + setValue: vi.fn(), + forScope: () => ({ settings: {}, originalSettings: {}, path: '' }), + migrationWarnings: [], + getUserHooks: () => undefined, + getProjectHooks: () => undefined, + } as never); + vi.mocked(parseArguments).mockResolvedValue({ + extensions: undefined, + } as never); + + await main(); + signalHandlers.get('SIGTERM')?.('SIGTERM'); + await Promise.resolve(); + await Promise.resolve(); + + expect(setRawModeSpy).toHaveBeenCalledWith(false); + expect(runExitCleanupMock).toHaveBeenCalledTimes(1); + expect(processExitSpy).toHaveBeenCalledWith(143); + + processOnceSpy.mockRestore(); + processExitSpy.mockRestore(); + }); }); describe('validateDnsResolutionOrder', () => { diff --git a/packages/cli/src/gemini.tsx b/packages/cli/src/gemini.tsx index 1add9158036..ac17bcdb2e8 100644 --- a/packages/cli/src/gemini.tsx +++ b/packages/cli/src/gemini.tsx @@ -164,6 +164,41 @@ ${reason.stack}` }); } +function getSignalExitCode(signal: NodeJS.Signals): number { + return signal === 'SIGINT' ? 130 : 143; +} + +function installInteractiveSignalHandlers(wasRaw: boolean): () => void { + let cleanupStarted = false; + + const handleSignal = (signal: NodeJS.Signals) => { + if (process.stdin.isTTY) { + process.stdin.setRawMode(wasRaw); + } + + if (cleanupStarted) { + return; + } + cleanupStarted = true; + + void runExitCleanup() + .catch((error) => { + debugLogger.error(`Error during ${signal} cleanup:`, error); + }) + .finally(() => { + process.exit(getSignalExitCode(signal)); + }); + }; + + process.once('SIGTERM', handleSignal); + process.once('SIGINT', handleSignal); + + return () => { + process.removeListener('SIGTERM', handleSignal); + process.removeListener('SIGINT', handleSignal); + }; +} + export async function startInteractiveUI( config: Config, settings: LoadedSettings, @@ -559,6 +594,9 @@ export async function main() { const wasRaw = process.stdin.isRaw; let kittyProtocolDetectionComplete: Promise | undefined; let themeAutoDetectionComplete: Promise | undefined; + if (config.isInteractive()) { + registerCleanup(installInteractiveSignalHandlers(wasRaw)); + } if (config.isInteractive() && !wasRaw && process.stdin.isTTY) { // Set this as early as possible to avoid spurious characters from // input showing up in the output. @@ -569,14 +607,6 @@ export async function main() { // Ensure the stdin listener is removed on any exit path (error, signal, etc.) registerCleanup(() => stopAndGetCapturedInput()); - // This cleanup isn't strictly needed but may help in certain situations. - process.on('SIGTERM', () => { - process.stdin.setRawMode(wasRaw); - }); - process.on('SIGINT', () => { - process.stdin.setRawMode(wasRaw); - }); - // Detect and enable Kitty keyboard protocol once at startup. kittyProtocolDetectionComplete = detectAndEnableKittyProtocol(); diff --git a/packages/core/src/config/config.test.ts b/packages/core/src/config/config.test.ts index 914ae42017a..aa74be5253a 100644 --- a/packages/core/src/config/config.test.ts +++ b/packages/core/src/config/config.test.ts @@ -15,6 +15,7 @@ import { DEFAULT_TELEMETRY_TARGET, DEFAULT_OTLP_ENDPOINT, QwenLogger, + isTelemetrySdkInitialized, shutdownTelemetry, } from '../telemetry/index.js'; import type { @@ -179,6 +180,7 @@ vi.mock('../telemetry/index.js', async (importOriginal) => { return { ...actual, initializeTelemetry: vi.fn(), + isTelemetrySdkInitialized: vi.fn(() => false), shutdownTelemetry: vi.fn().mockResolvedValue(undefined), uiTelemetryService: { getLastPromptTokenCount: vi.fn(), @@ -278,6 +280,7 @@ describe('Server Config (config.ts)', () => { beforeEach(() => { // Reset mocks if necessary vi.clearAllMocks(); + vi.mocked(isTelemetrySdkInitialized).mockReturnValue(false); vi.spyOn(QwenLogger.prototype, 'logStartSessionEvent').mockImplementation( async () => undefined, ); @@ -850,11 +853,12 @@ describe('Server Config (config.ts)', () => { expect(config.getTelemetryEnabled()).toBe(true); }); - it('Config shutdown should flush telemetry even before initialization completes', async () => { + it('Config shutdown should flush telemetry when SDK is initialized', async () => { const paramsWithTelemetry: ConfigParameters = { ...baseParams, telemetry: { enabled: true }, }; + vi.mocked(isTelemetrySdkInitialized).mockReturnValue(true); const config = new Config(paramsWithTelemetry); await config.shutdown(); @@ -862,6 +866,19 @@ describe('Server Config (config.ts)', () => { expect(shutdownTelemetry).toHaveBeenCalledTimes(1); }); + it('Config shutdown should skip telemetry shutdown before SDK initialization', async () => { + const paramsWithTelemetry: ConfigParameters = { + ...baseParams, + telemetry: { enabled: true }, + }; + vi.mocked(isTelemetrySdkInitialized).mockReturnValue(false); + const config = new Config(paramsWithTelemetry); + + await config.shutdown(); + + expect(shutdownTelemetry).not.toHaveBeenCalled(); + }); + it('Config constructor should set telemetry to false when provided as false', () => { const paramsWithTelemetry: ConfigParameters = { ...baseParams, diff --git a/packages/core/src/config/config.ts b/packages/core/src/config/config.ts index cce3a505d41..09d67d98d69 100644 --- a/packages/core/src/config/config.ts +++ b/packages/core/src/config/config.ts @@ -65,6 +65,7 @@ import { BackgroundShellRegistry } from '../services/backgroundShellRegistry.js' import { DEFAULT_OTLP_ENDPOINT, DEFAULT_TELEMETRY_TARGET, + isTelemetrySdkInitialized, initializeTelemetry, shutdownTelemetry, logStartSession, @@ -1647,7 +1648,7 @@ export class Config { // Log but don't throw - cleanup should be best-effort this.debugLogger.error('Error during Config shutdown:', error); } finally { - if (this.telemetrySettings.enabled) { + if (isTelemetrySdkInitialized()) { await shutdownTelemetry(); } } diff --git a/packages/core/src/telemetry/log-to-span-processor.test.ts b/packages/core/src/telemetry/log-to-span-processor.test.ts index 9996b78a5b3..181b8cc0ebe 100644 --- a/packages/core/src/telemetry/log-to-span-processor.test.ts +++ b/packages/core/src/telemetry/log-to-span-processor.test.ts @@ -78,6 +78,19 @@ describe('LogToSpanProcessor', () => { expect(exportedSpans[0].endTime).toEqual([1000, 250000000]); }); + it('ignores non-finite duration_ms values', async () => { + const logRecord = { + body: 'api response', + hrTime: [1000, 0] as [number, number], + attributes: { duration_ms: Infinity }, + } as unknown as ReadableLogRecord; + + processor.onEmit(logRecord); + await processor.forceFlush(); + + expect(exportedSpans[0].endTime).toEqual([1000, 0]); + }); + it('handles duration_ms that causes second rollover', async () => { const logRecord = { body: 'long operation', @@ -119,6 +132,29 @@ describe('LogToSpanProcessor', () => { expect(exportedSpans[0].attributes['bad']).toBe('[unserializable]'); }); + it('drops sensitive attributes before exporting bridged spans', async () => { + const logRecord = { + body: 'event', + hrTime: [1000, 0] as [number, number], + attributes: { + prompt: 'secret prompt', + function_args: '{"token":"secret"}', + response_text: 'secret response', + safe: 'visible', + }, + } as unknown as ReadableLogRecord; + + processor.onEmit(logRecord); + await processor.forceFlush(); + + const attrs = exportedSpans[0].attributes; + expect(attrs).not.toHaveProperty('prompt'); + expect(attrs).not.toHaveProperty('function_args'); + expect(attrs).not.toHaveProperty('response_text'); + expect(attrs['safe']).toBe('visible'); + expect(attrs['log.bridge']).toBe(true); + }); + it('skips null and undefined attributes', async () => { const logRecord = { body: 'event', @@ -149,6 +185,20 @@ describe('LogToSpanProcessor', () => { expect(exportedSpans[0].name).toBe('unknown'); }); + it('truncates long span names', async () => { + const longName = 'x'.repeat(200); + const logRecord = { + body: longName, + hrTime: [1000, 0] as [number, number], + attributes: {}, + } as unknown as ReadableLogRecord; + + processor.onEmit(logRecord); + await processor.forceFlush(); + + expect(exportedSpans[0].name).toBe(`${'x'.repeat(128)}...`); + }); + it('generates unique trace IDs without session.id', async () => { const logRecord1 = { body: 'event1', @@ -274,6 +324,52 @@ describe('LogToSpanProcessor', () => { expect(exportedSpans[0].attributes['log.severity_text']).toBe('INFO'); }); + it('reuses in-flight exports and flushes queued spans afterwards', async () => { + await processor.shutdown(); + exportedSpans = []; + const exportCallbacks: Array<(result: { code: number }) => void> = []; + let exportCallCount = 0; + mockExporter = { + export: vi.fn((spans, cb) => { + exportCallCount += 1; + exportedSpans.push(...spans); + if (exportCallCount === 1) { + exportCallbacks.push(cb); + } else { + cb({ code: 0 }); + } + }), + shutdown: vi.fn().mockResolvedValue(undefined), + forceFlush: vi.fn().mockResolvedValue(undefined), + } as unknown as SpanExporter; + processor = new LogToSpanProcessor(mockExporter, 60000); + + processor.onEmit({ + body: 'first', + hrTime: [1000, 0] as [number, number], + attributes: {}, + } as unknown as ReadableLogRecord); + const firstFlush = processor.forceFlush(); + await Promise.resolve(); + + processor.onEmit({ + body: 'second', + hrTime: [1001, 0] as [number, number], + attributes: {}, + } as unknown as ReadableLogRecord); + const secondFlush = processor.forceFlush(); + await Promise.resolve(); + + expect(mockExporter.export).toHaveBeenCalledTimes(1); + expect(exportedSpans.map((span) => span.name)).toEqual(['first']); + + exportCallbacks[0]({ code: 0 }); + await Promise.all([firstFlush, secondFlush]); + + expect(mockExporter.export).toHaveBeenCalledTimes(2); + expect(exportedSpans.map((span) => span.name)).toEqual(['first', 'second']); + }); + it('shutdown flushes remaining spans and shuts down exporter', async () => { const logRecord = { body: 'final event', diff --git a/packages/core/src/telemetry/log-to-span-processor.ts b/packages/core/src/telemetry/log-to-span-processor.ts index 98fe737465a..3033a666418 100644 --- a/packages/core/src/telemetry/log-to-span-processor.ts +++ b/packages/core/src/telemetry/log-to-span-processor.ts @@ -20,6 +20,12 @@ import { createHash } from 'node:crypto'; import { SERVICE_NAME } from './constants.js'; const EXPORT_TIMEOUT_MS = 30_000; +const MAX_SPAN_NAME_LENGTH = 128; +const SENSITIVE_ATTRIBUTE_KEYS = new Set([ + 'prompt', + 'function_args', + 'response_text', +]); /** * A LogRecordProcessor that converts each OTel log record into a span @@ -52,13 +58,17 @@ export class LogToSpanProcessor implements LogRecordProcessor { } onEmit(logRecord: ReadableLogRecord): void { - const name = String(logRecord.body ?? 'unknown'); + const name = sanitizeSpanName(logRecord.body); const startTime = logRecord.hrTime; const attributes: Record = {}; if (logRecord.attributes) { for (const [key, value] of Object.entries(logRecord.attributes)) { - if (value !== undefined && value !== null) { + if ( + value !== undefined && + value !== null && + !SENSITIVE_ATTRIBUTE_KEYS.has(key) + ) { attributes[key] = typeof value === 'object' ? safeStringify(value) @@ -78,7 +88,11 @@ export class LogToSpanProcessor implements LogRecordProcessor { let endTime = startTime; const durationMs = logRecord.attributes?.['duration_ms']; - if (typeof durationMs === 'number' && durationMs > 0) { + if ( + typeof durationMs === 'number' && + Number.isFinite(durationMs) && + durationMs > 0 + ) { const [secs, nanos] = startTime; const durationNanos = durationMs * 1_000_000; const endNanos = nanos + durationNanos; @@ -123,6 +137,7 @@ export class LogToSpanProcessor implements LogRecordProcessor { } private flush(): Promise { + if (this.inFlightExport) return this.inFlightExport; if (this.buffer.length === 0) return Promise.resolve(); const spans = this.buffer.splice(0); const exportPromise = new Promise((resolve) => { @@ -155,8 +170,10 @@ export class LogToSpanProcessor implements LogRecordProcessor { resolve(); } }); - this.inFlightExport = exportPromise; - return exportPromise; + this.inFlightExport = exportPromise.finally(() => { + this.inFlightExport = undefined; + }); + return this.inFlightExport; } async shutdown(): Promise { @@ -202,6 +219,13 @@ interface ReadableSpanLike { recordException: () => void; } +function sanitizeSpanName(body: unknown): string { + const rawName = String(body ?? 'unknown'); + return rawName.length > MAX_SPAN_NAME_LENGTH + ? `${rawName.slice(0, MAX_SPAN_NAME_LENGTH)}...` + : rawName; +} + /** * Safely stringify an object value for use as a span attribute. * Handles circular references and BigInt without throwing. diff --git a/packages/core/src/telemetry/sdk.ts b/packages/core/src/telemetry/sdk.ts index a70a3f4e28e..faf2c68b32f 100644 --- a/packages/core/src/telemetry/sdk.ts +++ b/packages/core/src/telemetry/sdk.ts @@ -193,7 +193,8 @@ export function initializeTelemetry(config: Config): void { logExporter = new OTLPLogExporterHttp({ url: logsUrl }); } else if (tracesUrl) { // Bridge: no logs endpoint but traces endpoint exists. - // Convert log records to spans and export via a dedicated trace exporter. + // Convert log records to spans. Use a dedicated trace exporter so the + // bridge owns its own forceFlush/shutdown lifecycle. logToSpanProcessor = new LogToSpanProcessor( new OTLPTraceExporterHttp({ url: tracesUrl }), ); From 3a0d16e247f7469b31f48a9af4d07801c8a0a443 Mon Sep 17 00:00:00 2001 From: "jinye.djy" Date: Fri, 1 May 2026 21:33:18 +0800 Subject: [PATCH 7/7] fix(telemetry): address review follow-ups Co-authored-by: Qwen-Coder --- packages/cli/src/gemini.test.tsx | 6 ++--- packages/cli/src/gemini.tsx | 15 +++++++++---- .../src/telemetry/log-to-span-processor.ts | 3 ++- packages/core/src/telemetry/sdk.test.ts | 22 +++++++++++++++++++ packages/core/src/telemetry/sdk.ts | 9 +++++--- 5 files changed, 44 insertions(+), 11 deletions(-) diff --git a/packages/cli/src/gemini.test.tsx b/packages/cli/src/gemini.test.tsx index e30995bb55f..6820cbd612c 100644 --- a/packages/cli/src/gemini.test.tsx +++ b/packages/cli/src/gemini.test.tsx @@ -639,7 +639,7 @@ describe('gemini.tsx main function kitty protocol', () => { expect(detectAndEnableKittyProtocol).toHaveBeenCalledTimes(1); }); - it('should run cleanup before exiting on interactive SIGTERM', async () => { + it('should run cleanup before exiting on interactive SIGINT', async () => { const { loadCliConfig, parseArguments } = await import( './config/config.js' ); @@ -700,13 +700,13 @@ describe('gemini.tsx main function kitty protocol', () => { } as never); await main(); - signalHandlers.get('SIGTERM')?.('SIGTERM'); + signalHandlers.get('SIGINT')?.(); await Promise.resolve(); await Promise.resolve(); expect(setRawModeSpy).toHaveBeenCalledWith(false); expect(runExitCleanupMock).toHaveBeenCalledTimes(1); - expect(processExitSpy).toHaveBeenCalledWith(143); + expect(processExitSpy).toHaveBeenCalledWith(130); processOnceSpy.mockRestore(); processExitSpy.mockRestore(); diff --git a/packages/cli/src/gemini.tsx b/packages/cli/src/gemini.tsx index ac17bcdb2e8..1020f65e94c 100644 --- a/packages/cli/src/gemini.tsx +++ b/packages/cli/src/gemini.tsx @@ -190,12 +190,19 @@ function installInteractiveSignalHandlers(wasRaw: boolean): () => void { }); }; - process.once('SIGTERM', handleSignal); - process.once('SIGINT', handleSignal); + const handleSigterm = () => { + handleSignal('SIGTERM'); + }; + const handleSigint = () => { + handleSignal('SIGINT'); + }; + + process.once('SIGTERM', handleSigterm); + process.once('SIGINT', handleSigint); return () => { - process.removeListener('SIGTERM', handleSignal); - process.removeListener('SIGINT', handleSignal); + process.removeListener('SIGTERM', handleSigterm); + process.removeListener('SIGINT', handleSigint); }; } diff --git a/packages/core/src/telemetry/log-to-span-processor.ts b/packages/core/src/telemetry/log-to-span-processor.ts index 3033a666418..cdf57830b14 100644 --- a/packages/core/src/telemetry/log-to-span-processor.ts +++ b/packages/core/src/telemetry/log-to-span-processor.ts @@ -228,7 +228,8 @@ function sanitizeSpanName(body: unknown): string { /** * Safely stringify an object value for use as a span attribute. - * Handles circular references and BigInt without throwing. + * Returns a bounded fallback when JSON serialization fails, such as for + * circular references or BigInt values. */ function safeStringify(value: unknown): string { try { diff --git a/packages/core/src/telemetry/sdk.test.ts b/packages/core/src/telemetry/sdk.test.ts index 89fdb9ff1a9..f88e9b9aa66 100644 --- a/packages/core/src/telemetry/sdk.test.ts +++ b/packages/core/src/telemetry/sdk.test.ts @@ -5,6 +5,7 @@ */ import { describe, it, expect, vi, beforeEach, afterEach } from 'vitest'; +import { diag } from '@opentelemetry/api'; import type { Config } from '../config/config.js'; import { initializeTelemetry, @@ -235,6 +236,27 @@ describe('Telemetry SDK', () => { expect(NodeSDK.prototype.start).toHaveBeenCalled(); }); + it('should warn and skip startup for gRPC per-signal endpoints without base endpoint', () => { + const diagWarnSpy = vi.spyOn(diag, 'warn').mockImplementation(() => {}); + try { + vi.spyOn(mockConfig, 'getTelemetryOtlpProtocol').mockReturnValue('grpc'); + vi.spyOn(mockConfig, 'getTelemetryOtlpEndpoint').mockReturnValue(''); + vi.spyOn(mockConfig, 'getTelemetryOtlpTracesEndpoint').mockReturnValue( + 'http://traces-host/token/api/otlp/traces', + ); + + initializeTelemetry(mockConfig); + + expect(diagWarnSpy).toHaveBeenCalledWith( + expect.stringContaining('Telemetry SDK startup was skipped'), + ); + expect(NodeSDK.prototype.start).not.toHaveBeenCalled(); + expect(isTelemetrySdkInitialized()).toBe(false); + } finally { + diagWarnSpy.mockRestore(); + } + }); + it('should use OTLP exporters when target is gcp but useCollector is true', () => { vi.spyOn(mockConfig, 'getTelemetryTarget').mockReturnValue( TelemetryTarget.GCP, diff --git a/packages/core/src/telemetry/sdk.ts b/packages/core/src/telemetry/sdk.ts index faf2c68b32f..52bdb671282 100644 --- a/packages/core/src/telemetry/sdk.ts +++ b/packages/core/src/telemetry/sdk.ts @@ -208,10 +208,13 @@ export function initializeTelemetry(config: Config): void { } else { // grpc — per-signal endpoints are not supported with gRPC protocol. if (!parsedEndpoint) { - debugLogger.warn( + const warning = 'Per-signal OTLP endpoints are only supported with HTTP protocol. ' + - 'Set otlpProtocol to "http" or provide a base otlpEndpoint for gRPC.', - ); + 'Set otlpProtocol to "http" or provide a base otlpEndpoint for gRPC. ' + + 'Telemetry SDK startup was skipped because no supported gRPC endpoint was configured.'; + diag.warn(warning); + debugLogger.warn(warning); + return; } else { spanExporter = new OTLPTraceExporter({ url: parsedEndpoint,