Add partial span export support

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.
 

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<ReadableSpan, ReadableSpan[]> {
     contentType: string;
     headers?: Record<string, string> | undefined;
   } {
-    const serialized = JsonTraceSerializer.serializeRequest(spans);
+    const serialized = serializeTraceServiceRequestJson(spans);
     return {
       body: new TextDecoder().decode(serialized),
       contentType: "application/json",

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): {

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<string> {
+  const unendedSpanIds = new Set<string>();
+  for (const span of spans) {
+    if (!span.ended) {
+      unendedSpanIds.add(spanKeyFromReadableSpan(span));
+    }
+  }
+  return unendedSpanIds;
+}
+
+function omitZeroSpanEndTimes(
+  request: IExportTraceServiceRequest,
+  unendedSpanIds: Set<string>,
+): 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;
+}

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<void> {
+      return Promise.resolve();
+    },
+  };
+}