diff --git a/packages/otel/README.md b/packages/otel/README.md index 846c794..1c8c0c8 100644 --- a/packages/otel/README.md +++ b/packages/otel/README.md @@ -63,6 +63,7 @@ Registers the OpenTelemetry SDK with the specified configuration. Configuration - `propagators`: A set of propagators that may extend inbound and outbound contexts. By default, `@vercel/otel` configures [W3C Trace Context](https://www.w3.org/TR/trace-context/) propagator. - `traceSampler`: The sampler to be used to decide which requests should be traced. By default, all requests are traced. This option can be changed to, for instance, only trace 1% of all requests. - `spanProcessors` and `traceExporter`: The export mechanism for traces. By default, `@vercel/otel` configures the best export mechanism for the environment. For instance, if a [tracing integrations](https://vercel.com/docs/observability/otel-overview) is configured on Vercel, this integration will be automatically used for export; otherwise an [OTLP exporter](https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/#otlp-exporter) can be used if configured in environment variables. +- `experimentalPartialSpans`: Export a snapshot when a span starts, then export the completed span again when it ends. The Vercel runtime exporter still receives only completed spans. By default, spans are exported only after they end. See [API](https://otel.vercel.sh/api/) for more details. diff --git a/packages/otel/src/exporters/exporter-trace-otlp-http-fetch.ts b/packages/otel/src/exporters/exporter-trace-otlp-http-fetch.ts index b7222dc..7dfe820 100644 --- a/packages/otel/src/exporters/exporter-trace-otlp-http-fetch.ts +++ b/packages/otel/src/exporters/exporter-trace-otlp-http-fetch.ts @@ -1,9 +1,9 @@ import type { ReadableSpan, SpanExporter } from "@opentelemetry/sdk-trace-base"; -import { JsonTraceSerializer } from "@opentelemetry/otlp-transformer/build/src/trace/json/trace"; import type { ExportResult } from "@opentelemetry/core"; import { OTLPExporterEdgeBase } from "./otlp-exporter-base"; import { getDefaultUrl } from "./trace-config"; import type { OTLPExporterConfig } from "./config"; +import { serializeTraceServiceRequestJson } from "./trace-service-request"; /** * OTLP exporter for the `http/json` protocol. Compatible with the "edge" runtime. @@ -46,7 +46,7 @@ class Impl extends OTLPExporterEdgeBase { contentType: string; headers?: Record | undefined; } { - const serialized = JsonTraceSerializer.serializeRequest(spans); + const serialized = serializeTraceServiceRequestJson(spans); return { body: new TextDecoder().decode(serialized), contentType: "application/json", diff --git a/packages/otel/src/exporters/exporter-trace-otlp-proto-fetch.ts b/packages/otel/src/exporters/exporter-trace-otlp-proto-fetch.ts index 3846d69..9829520 100644 --- a/packages/otel/src/exporters/exporter-trace-otlp-proto-fetch.ts +++ b/packages/otel/src/exporters/exporter-trace-otlp-proto-fetch.ts @@ -1,11 +1,11 @@ import type { ReadableSpan, SpanExporter } from "@opentelemetry/sdk-trace-base"; import type { IExportTraceServiceRequest } from "@opentelemetry/otlp-transformer/build/src/trace/internal-types"; -import { createExportTraceServiceRequest } from "@opentelemetry/otlp-transformer/build/src/trace/internal"; import type { ExportResult } from "@opentelemetry/core"; import { encodeTraceServiceRequest } from "./proto"; import { OTLPExporterEdgeBase } from "./otlp-exporter-base"; import { getDefaultUrl } from "./trace-config"; import type { OTLPExporterConfig } from "./config"; +import { createTraceServiceRequest } from "./trace-service-request"; /** * OTLP exporter for the `http/protobuf` protocol. Compatible with the "edge" runtime. @@ -40,7 +40,7 @@ class Impl extends OTLPExporterEdgeBase< IExportTraceServiceRequest > { convert(spans: ReadableSpan[]): IExportTraceServiceRequest { - return createExportTraceServiceRequest(spans); + return createTraceServiceRequest(spans); } override toMessage(serviceRequest: IExportTraceServiceRequest): { diff --git a/packages/otel/src/exporters/trace-service-request.ts b/packages/otel/src/exporters/trace-service-request.ts new file mode 100644 index 0000000..2d8601a --- /dev/null +++ b/packages/otel/src/exporters/trace-service-request.ts @@ -0,0 +1,96 @@ +import type { ReadableSpan } from "@opentelemetry/sdk-trace-base"; +import type { + Fixed64, + OtlpEncodingOptions, +} from "@opentelemetry/otlp-transformer/build/src/common/internal-types"; +import { createExportTraceServiceRequest } from "@opentelemetry/otlp-transformer/build/src/trace/internal"; +import type { + IExportTraceServiceRequest, + ISpan, +} from "@opentelemetry/otlp-transformer/build/src/trace/internal-types"; + +export function createTraceServiceRequest( + spans: ReadableSpan[], + options?: OtlpEncodingOptions, +): IExportTraceServiceRequest { + const request = createExportTraceServiceRequest(spans, options); + omitZeroSpanEndTimes(request, getUnendedSpanIds(spans)); + return request; +} + +export function serializeTraceServiceRequestJson( + spans: ReadableSpan[], +): Uint8Array { + const request = createTraceServiceRequest(spans, { + useHex: true, + useLongBits: false, + }); + return new TextEncoder().encode(JSON.stringify(request)); +} + +function getUnendedSpanIds(spans: ReadableSpan[]): Set { + const unendedSpanIds = new Set(); + for (const span of spans) { + if (!span.ended) { + unendedSpanIds.add(spanKeyFromReadableSpan(span)); + } + } + return unendedSpanIds; +} + +function omitZeroSpanEndTimes( + request: IExportTraceServiceRequest, + unendedSpanIds: Set, +): void { + if (unendedSpanIds.size === 0) { + return; + } + + for (const resourceSpans of request.resourceSpans ?? []) { + for (const scopeSpans of resourceSpans.scopeSpans) { + for (const span of scopeSpans.spans ?? []) { + const spanId = spanKeyFromOtlpSpan(span); + if ( + spanId && + unendedSpanIds.has(spanId) && + isZeroFixed64(span.endTimeUnixNano) + ) { + delete (span as { endTimeUnixNano?: unknown }).endTimeUnixNano; + } + } + } + } +} + +function spanKeyFromReadableSpan(span: ReadableSpan): string { + const { traceId, spanId } = span.spanContext(); + return `${traceId}:${spanId}`; +} + +function spanKeyFromOtlpSpan(span: ISpan): string | undefined { + if (!span.traceId || !span.spanId) { + return undefined; + } + return `${encodedSpanIdToHex(span.traceId)}:${encodedSpanIdToHex( + span.spanId, + )}`; +} + +function encodedSpanIdToHex(value: string | Uint8Array): string { + if (typeof value === "string") { + return value; + } + return Array.from(value, (byte) => byte.toString(16).padStart(2, "0")).join( + "", + ); +} + +function isZeroFixed64(value: Fixed64): boolean { + if (typeof value === "number") { + return value === 0; + } + if (typeof value === "string") { + return value === "0"; + } + return value.low === 0 && value.high === 0; +} diff --git a/packages/otel/src/processor/partial-span-processor.test.ts b/packages/otel/src/processor/partial-span-processor.test.ts new file mode 100644 index 0000000..32600d7 --- /dev/null +++ b/packages/otel/src/processor/partial-span-processor.test.ts @@ -0,0 +1,138 @@ +import { describe, expect, it } from "vitest"; +import { ExportResultCode, type ExportResult } from "@opentelemetry/core"; +import type { + ReadableSpan, + SpanExporter, +} from "@opentelemetry/sdk-trace-base"; +import { BasicTracerProvider } from "@opentelemetry/sdk-trace-base"; +import { createTraceServiceRequest } from "../exporters/trace-service-request"; +import { PartialSpanProcessor } from "./partial-span-processor"; + +describe("PartialSpanProcessor", () => { + it("exports a start snapshot and a final snapshot", async () => { + const exportedSpans: ReadableSpan[][] = []; + const exporter = createExporter(exportedSpans); + const provider = new BasicTracerProvider({ + spanProcessors: [ + new PartialSpanProcessor(exporter, { scheduledDelayMillis: 60_000 }), + ], + }); + + const span = provider.getTracer("test").startSpan("partial-span"); + span.setAttribute("phase", "started"); + + await provider.forceFlush(); + + span.setAttribute("phase", "ended"); + span.end(); + await provider.forceFlush(); + + expect(exportedSpans).toHaveLength(2); + expect(exportedSpans[0]?.[0]?.ended).toBe(false); + expect(exportedSpans[0]?.[0]?.attributes.phase).toBe("started"); + expect(exportedSpans[1]?.[0]?.ended).toBe(true); + expect(exportedSpans[1]?.[0]?.attributes.phase).toBe("ended"); + + await provider.shutdown(); + }); + + it("omits the OTLP end time for partial snapshots", async () => { + const exportedSpans: ReadableSpan[][] = []; + const exporter = createExporter(exportedSpans); + const provider = new BasicTracerProvider({ + spanProcessors: [ + new PartialSpanProcessor(exporter, { scheduledDelayMillis: 60_000 }), + ], + }); + + const span = provider.getTracer("test").startSpan("partial-span"); + await provider.forceFlush(); + span.end(); + await provider.forceFlush(); + + const partialSpan = exportedSpans[0]?.[0]; + const finalSpan = exportedSpans[1]?.[0]; + expect(partialSpan).toBeDefined(); + expect(finalSpan).toBeDefined(); + if (!partialSpan || !finalSpan) { + throw new Error("Expected partial and final spans to be exported"); + } + + const request = createTraceServiceRequest([partialSpan, finalSpan], { + useHex: true, + useLongBits: false, + }); + const otlpSpans = request.resourceSpans?.[0]?.scopeSpans[0]?.spans; + + expect(otlpSpans?.[0]).not.toHaveProperty("endTimeUnixNano"); + expect(otlpSpans?.[1]).toHaveProperty("endTimeUnixNano"); + + await provider.shutdown(); + }); + + it("does not export a partial snapshot when the span ends before flush", async () => { + const exportedSpans: ReadableSpan[][] = []; + const exporter = createExporter(exportedSpans); + const provider = new BasicTracerProvider({ + spanProcessors: [ + new PartialSpanProcessor(exporter, { scheduledDelayMillis: 60_000 }), + ], + }); + + const span = provider.getTracer("test").startSpan("short-span"); + span.end(); + await provider.forceFlush(); + + expect(exportedSpans).toHaveLength(1); + expect(exportedSpans[0]?.[0]?.name).toBe("short-span"); + expect(exportedSpans[0]?.[0]?.ended).toBe(true); + + await provider.shutdown(); + }); + + it("keeps a zero OTLP end time for spans that are marked ended", async () => { + const exportedSpans: ReadableSpan[][] = []; + const exporter = createExporter(exportedSpans); + const provider = new BasicTracerProvider({ + spanProcessors: [ + new PartialSpanProcessor(exporter, { scheduledDelayMillis: 60_000 }), + ], + }); + + const span = provider.getTracer("test").startSpan("epoch-ended-span"); + await provider.forceFlush(); + + const partialSpan = exportedSpans[0]?.[0]; + expect(partialSpan).toBeDefined(); + if (!partialSpan) { + throw new Error("Expected a partial span to be exported"); + } + + const endedZeroSpan: ReadableSpan = { ...partialSpan, ended: true }; + const request = createTraceServiceRequest([endedZeroSpan], { + useHex: true, + useLongBits: false, + }); + const otlpSpan = request.resourceSpans?.[0]?.scopeSpans[0]?.spans?.[0]; + + expect(otlpSpan).toHaveProperty("endTimeUnixNano", "0"); + + span.end(); + await provider.shutdown(); + }); +}); + +function createExporter(exportedSpans: ReadableSpan[][]): SpanExporter { + return { + export( + spans: ReadableSpan[], + resultCallback: (result: ExportResult) => void, + ): void { + exportedSpans.push(spans); + resultCallback({ code: ExportResultCode.SUCCESS }); + }, + shutdown(): Promise { + return Promise.resolve(); + }, + }; +} diff --git a/packages/otel/src/processor/partial-span-processor.ts b/packages/otel/src/processor/partial-span-processor.ts new file mode 100644 index 0000000..4bc3e95 --- /dev/null +++ b/packages/otel/src/processor/partial-span-processor.ts @@ -0,0 +1,282 @@ +import { + context, + diag, + type Context, + type HrTime, + type Link, +} from "@opentelemetry/api"; +import { + ExportResultCode, + getNumberFromEnv, + globalErrorHandler, + suppressTracing, + unrefTimer, +} from "@opentelemetry/core"; +import type { + BufferConfig, + ReadableSpan, + Span, + SpanExporter, + SpanProcessor, + TimedEvent, +} from "@opentelemetry/sdk-trace-base"; +import { isSampled } from "../util/sampled"; + +/** @internal */ +export class PartialSpanProcessor implements SpanProcessor { + private readonly maxExportBatchSize: number; + private readonly maxQueueSize: number; + private readonly scheduledDelayMillis: number; + private readonly exportTimeoutMillis: number; + private queue: QueuedSpan[] = []; + private timer: NodeJS.Timeout | undefined; + private flushPromise = Promise.resolve(); + private shutdownStarted = false; + private shutdownPromise: Promise | undefined; + private droppedSpansCount = 0; + + constructor( + private readonly exporter: SpanExporter, + config?: BufferConfig, + ) { + this.maxExportBatchSize = + typeof config?.maxExportBatchSize === "number" + ? config.maxExportBatchSize + : (getNumberFromEnv("OTEL_BSP_MAX_EXPORT_BATCH_SIZE") ?? 512); + this.maxQueueSize = + typeof config?.maxQueueSize === "number" + ? config.maxQueueSize + : (getNumberFromEnv("OTEL_BSP_MAX_QUEUE_SIZE") ?? 2048); + this.scheduledDelayMillis = + typeof config?.scheduledDelayMillis === "number" + ? config.scheduledDelayMillis + : (getNumberFromEnv("OTEL_BSP_SCHEDULE_DELAY") ?? 5000); + this.exportTimeoutMillis = + typeof config?.exportTimeoutMillis === "number" + ? config.exportTimeoutMillis + : (getNumberFromEnv("OTEL_BSP_EXPORT_TIMEOUT") ?? 30000); + + if (this.maxExportBatchSize > this.maxQueueSize) { + diag.warn( + "PartialSpanProcessor: maxExportBatchSize must be smaller or equal to maxQueueSize, setting maxExportBatchSize to match maxQueueSize", + ); + this.maxExportBatchSize = this.maxQueueSize; + } + } + + forceFlush(): Promise { + if (this.shutdownPromise) { + return this.shutdownPromise; + } + return this.flushAll().then(() => this.exporter.forceFlush?.()); + } + + onStart(span: Span, _parentContext: Context): void { + if (this.shutdownStarted || !isSampled(span.spanContext().traceFlags)) { + return; + } + this.addToBuffer(span, true); + } + + onEnd(span: ReadableSpan): void { + if (this.shutdownStarted || !isSampled(span.spanContext().traceFlags)) { + return; + } + this.removePendingPartial(span); + this.addToBuffer(span, false); + } + + shutdown(): Promise { + if (!this.shutdownPromise) { + this.shutdownStarted = true; + this.shutdownPromise = this.flushAll().then(() => this.exporter.shutdown()); + } + return this.shutdownPromise; + } + + private addToBuffer(span: ReadableSpan, partial: boolean): void { + if (this.queue.length >= this.maxQueueSize) { + if (this.droppedSpansCount === 0) { + diag.debug("maxQueueSize reached, dropping spans"); + } + this.droppedSpansCount++; + return; + } + + if (this.droppedSpansCount > 0) { + diag.warn( + `Dropped ${this.droppedSpansCount} spans because maxQueueSize reached`, + ); + this.droppedSpansCount = 0; + } + + this.queue.push({ + span, + partial, + key: spanKey(span), + }); + if (this.queue.length >= this.maxExportBatchSize) { + void this.flushAll().catch(globalErrorHandler); + return; + } + this.maybeStartTimer(); + } + + private flushAll(): Promise { + this.clearTimer(); + const flush = (): Promise => { + const flushNextBatch = (): Promise => { + if (this.queue.length === 0) { + return Promise.resolve(); + } + return this.flushOneBatch().then(flushNextBatch); + }; + return flushNextBatch(); + }; + this.flushPromise = this.flushPromise.then(flush, flush); + return this.flushPromise; + } + + private async flushOneBatch(): Promise { + const entries = this.queue.splice(0, this.maxExportBatchSize); + if (entries.length === 0) { + return; + } + const spans = entries.map(({ span }) => snapshotSpan(span)); + + await context.with(suppressTracing(context.active()), async () => { + const pendingResources: Promise[] = []; + for (const span of spans) { + if ( + span.resource.asyncAttributesPending && + span.resource.waitForAsyncAttributes + ) { + pendingResources.push(span.resource.waitForAsyncAttributes()); + } + } + await Promise.all(pendingResources); + await this.exportSpans(spans); + }); + } + + private exportSpans(spans: ReadableSpan[]): Promise { + return new Promise((resolve, reject) => { + let settled = false; + const timer = setTimeout(() => { + settled = true; + reject(new Error("Timeout")); + }, this.exportTimeoutMillis); + + const finish = (callback: () => void): void => { + if (settled) { + return; + } + settled = true; + clearTimeout(timer); + callback(); + }; + + try { + this.exporter.export(spans, (result) => { + finish(() => { + if (result.code === ExportResultCode.SUCCESS) { + resolve(); + } else { + reject( + result.error ?? + new Error("PartialSpanProcessor: span export failed"), + ); + } + }); + }); + } catch (error) { + finish(() => { + reject(error instanceof Error ? error : new Error(String(error))); + }); + } + }); + } + + private maybeStartTimer(): void { + if (this.timer !== undefined) { + return; + } + this.timer = setTimeout(() => { + void this.flushAll().catch(globalErrorHandler); + }, this.scheduledDelayMillis); + unrefTimer(this.timer); + } + + private clearTimer(): void { + if (this.timer !== undefined) { + clearTimeout(this.timer); + this.timer = undefined; + } + } + + private removePendingPartial(span: ReadableSpan): void { + const key = spanKey(span); + this.queue = this.queue.filter( + (entry) => !entry.partial || entry.key !== key, + ); + } +} + +function spanKey(span: ReadableSpan): string { + const { traceId, spanId } = span.spanContext(); + return `${traceId}:${spanId}`; +} + +interface QueuedSpan { + span: ReadableSpan; + partial: boolean; + key: string; +} + +function snapshotSpan(span: ReadableSpan): ReadableSpan { + const spanContext = { ...span.spanContext() }; + const parentSpanContext = span.parentSpanContext + ? { ...span.parentSpanContext } + : undefined; + const ended = span.ended; + + return { + name: span.name, + kind: span.kind, + spanContext: () => spanContext, + parentSpanContext, + startTime: copyHrTime(span.startTime), + endTime: ended ? copyHrTime(span.endTime) : [0, 0], + status: { ...span.status }, + attributes: { ...span.attributes }, + links: span.links.map(copyLink), + events: span.events.map(copyEvent), + duration: ended ? copyHrTime(span.duration) : [0, 0], + ended, + resource: span.resource, + instrumentationScope: span.instrumentationScope, + droppedAttributesCount: span.droppedAttributesCount, + droppedEventsCount: span.droppedEventsCount, + droppedLinksCount: span.droppedLinksCount, + }; +} + +function copyHrTime(time: HrTime): HrTime { + return [time[0], time[1]]; +} + +function copyEvent(event: TimedEvent): TimedEvent { + return { + ...event, + time: copyHrTime(event.time), + attributes: event.attributes ? { ...event.attributes } : undefined, + }; +} + +function copyLink(link: Link): Link { + return { + ...link, + context: { ...link.context }, + attributes: link.attributes ? { ...link.attributes } : undefined, + }; +} diff --git a/packages/otel/src/sdk.ts b/packages/otel/src/sdk.ts index 887be29..f46c46c 100644 --- a/packages/otel/src/sdk.ts +++ b/packages/otel/src/sdk.ts @@ -57,6 +57,7 @@ import { VercelRuntimePropagator } from "./vercel-request-context/propagator"; import { VercelRuntimeSpanExporter } from "./vercel-request-context/exporter"; import { getStringFromEnv, getStringListFromEnv } from "./utils/env-parser"; import { FilterWhenDrainedSpanProcessor } from "./processor/filter-when-drained-span-processor"; +import { PartialSpanProcessor } from "./processor/partial-span-processor"; interface Env { OTEL_SDK_DISABLED?: string; @@ -450,7 +451,7 @@ function parseSpanProcessor( processors.push( new FilterWhenDrainedSpanProcessor( - new BatchSpanProcessor(exporter), + createExporterSpanProcessor(exporter, configuration), ), ); } @@ -464,7 +465,10 @@ function parseSpanProcessor( ) { processors.push( new FilterWhenDrainedSpanProcessor( - new BatchSpanProcessor(parseTraceExporter(env)), + createExporterSpanProcessor( + parseTraceExporter(env), + configuration, + ), ), ); } @@ -475,11 +479,25 @@ function parseSpanProcessor( }) .filter(isNotNull), ...(configuration.traceExporter && configuration.traceExporter !== "auto" - ? [new BatchSpanProcessor(configuration.traceExporter)] + ? [ + createExporterSpanProcessor( + configuration.traceExporter, + configuration, + ), + ] : []), ]; } +function createExporterSpanProcessor( + exporter: SpanExporter, + configuration: Configuration, +): SpanProcessor { + return configuration.experimentalPartialSpans + ? new PartialSpanProcessor(exporter) + : new BatchSpanProcessor(exporter); +} + /** * This code is moved from the https://github.com/open-telemetry/opentelemetry-js/blob/00e78efd840d3f49d9d4b025a9965e8d3f2913ad/experimental/packages/opentelemetry-sdk-node/src/TracerProviderWithEnvExporter.ts#L41 * due to the https://github.com/open-telemetry/opentelemetry-js/issues/4212 diff --git a/packages/otel/src/types.ts b/packages/otel/src/types.ts index 5eacaae..23b2196 100644 --- a/packages/otel/src/types.ts +++ b/packages/otel/src/types.ts @@ -150,6 +150,14 @@ export interface Configuration { */ traceExporter?: SpanExporterOrName; + /** + * Export a snapshot when a span starts, then export the completed span again when it ends. + * This lets compatible backends receive long-running spans before completion. + * The Vercel runtime exporter still receives only completed spans. + * By default, spans are exported only after they end. + */ + experimentalPartialSpans?: boolean; + spanLimits?: SpanLimits; logRecordProcessors?: LogRecordProcessor[]; diff --git a/packages/otel/src/vercel-request-context/exporter.ts b/packages/otel/src/vercel-request-context/exporter.ts index ff9f959..bb2669a 100644 --- a/packages/otel/src/vercel-request-context/exporter.ts +++ b/packages/otel/src/vercel-request-context/exporter.ts @@ -1,8 +1,8 @@ import { diag } from "@opentelemetry/api"; import type { ReadableSpan, SpanExporter } from "@opentelemetry/sdk-trace-base"; import { ExportResultCode, type ExportResult } from "@opentelemetry/core"; -import { JsonTraceSerializer } from "@opentelemetry/otlp-transformer/build/src/trace/json/trace"; import type { IExportTraceServiceRequest } from "@opentelemetry/otlp-transformer/build/src/trace/internal-types"; +import { serializeTraceServiceRequestJson } from "../exporters/trace-service-request"; import { getVercelRequestContext } from "./api"; export class VercelRuntimeSpanExporter implements SpanExporter { @@ -18,11 +18,8 @@ export class VercelRuntimeSpanExporter implements SpanExporter { } try { - const serializedData = JsonTraceSerializer.serializeRequest(spans); + const serializedData = serializeTraceServiceRequestJson(spans); - if (!serializedData) { - throw new Error("Failed to serialize spans"); - } // Convert back to object format for the Vercel telemetry API const data = JSON.parse( new TextDecoder().decode(serializedData),