feat(otel): cap oversized export batches

Author: hassiebp

packages/core/src/utils.ts

@@ -6,6 +6,7 @@ type LangfuseEnvVar =
   | "LANGFUSE_TIMEOUT"
   | "LANGFUSE_FLUSH_AT"
   | "LANGFUSE_FLUSH_INTERVAL"
+  | "LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES"
   | "LANGFUSE_MEDIA_UPLOAD_ENABLED"
   | "LANGFUSE_LOG_LEVEL"
   | "LANGFUSE_DEBUG"

packages/otel/README.md

@@ -19,8 +19,16 @@ npm install @langfuse/otel @opentelemetry/sdk-trace-node
 LANGFUSE_PUBLIC_KEY="pk-lf-..."
 LANGFUSE_SECRET_KEY="sk-lf-..."
 LANGFUSE_BASE_URL="https://cloud.langfuse.com" # 🇪🇺 EU region. 🇺🇸 US: https://us.cloud.langfuse.com
+LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES="67108864" # Optional; defaults to 64 MiB
 ```
 
+The batch byte limit applies to the final serialized OTLP/HTTP JSON request
+before compression. If a batch exceeds the limit, the entire batch is dropped;
+it is not split, retried, or truncated. Invalid values fall back to 64 MiB with
+a warning. Set the limit to a positive decimal safe integer; surrounding
+whitespace is ignored. Custom exporters passed to `LangfuseSpanProcessor`
+bypass this limit because they may use a different wire format or transport.
+
 ## Quickstart
 
 ```typescript

packages/otel/package.json

@@ -41,7 +41,8 @@
     "dist"
   ],
   "dependencies": {
-    "@langfuse/core": "workspace:^"
+    "@langfuse/core": "workspace:^",
+    "@opentelemetry/otlp-transformer": "0.220.0"
   },
   "peerDependencies": {
     "@opentelemetry/api": "^1.9.0",

packages/otel/src/size-limited-span-exporter.ts

@@ -0,0 +1,123 @@
+import { getEnv, getGlobalLogger } from "@langfuse/core";
+import { ExportResultCode } from "@opentelemetry/core";
+import { JsonTraceSerializer } from "@opentelemetry/otlp-transformer";
+import type { ReadableSpan, SpanExporter } from "@opentelemetry/sdk-trace-base";
+
+export const DEFAULT_MAX_BATCH_SIZE_BYTES = 64 * 1024 * 1024;
+
+export function getSerializedBatchSizeBytes(
+  spans: ReadableSpan[],
+): number | undefined {
+  return JsonTraceSerializer.serializeRequest(spans)?.byteLength;
+}
+
+export function resolveMaxBatchSizeBytes(rawValue: string | undefined): number {
+  if (rawValue === undefined) return DEFAULT_MAX_BATCH_SIZE_BYTES;
+
+  const normalizedValue = rawValue.trim();
+  if (!/^\d+$/.test(normalizedValue)) {
+    getGlobalLogger().warn(
+      "Invalid LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES. Using the default limit.",
+      { defaultMaxBatchSizeBytes: DEFAULT_MAX_BATCH_SIZE_BYTES },
+    );
+    return DEFAULT_MAX_BATCH_SIZE_BYTES;
+  }
+
+  const parsedValue = Number(normalizedValue);
+  if (!Number.isSafeInteger(parsedValue) || parsedValue <= 0) {
+    getGlobalLogger().warn(
+      "Invalid LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES. Using the default limit.",
+      { defaultMaxBatchSizeBytes: DEFAULT_MAX_BATCH_SIZE_BYTES },
+    );
+    return DEFAULT_MAX_BATCH_SIZE_BYTES;
+  }
+
+  return parsedValue;
+}
+
+export function resolveMaxBatchSizeBytesFromEnvironment(): number {
+  const processValue =
+    typeof process !== "undefined"
+      ? process.env.LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES
+      : undefined;
+
+  return resolveMaxBatchSizeBytes(
+    processValue !== undefined
+      ? processValue
+      : getEnv("LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES"),
+  );
+}
+
+export class SizeLimitedSpanExporter implements SpanExporter {
+  private readonly delegate: SpanExporter;
+  private readonly maxBatchSizeBytes: number;
+  private readonly serializeRequest: (
+    spans: ReadableSpan[],
+  ) => { byteLength: number } | undefined;
+
+  constructor(params: {
+    delegate: SpanExporter;
+    maxBatchSizeBytes: number;
+    serializeRequest?: (
+      spans: ReadableSpan[],
+    ) => { byteLength: number } | undefined;
+  }) {
+    this.delegate = params.delegate;
+    this.maxBatchSizeBytes = params.maxBatchSizeBytes;
+    this.serializeRequest =
+      params.serializeRequest ?? JsonTraceSerializer.serializeRequest;
+  }
+
+  export(
+    spans: ReadableSpan[],
+    resultCallback: Parameters<SpanExporter["export"]>[1],
+  ): void {
+    let serializedSizeBytes: number | undefined;
+
+    try {
+      serializedSizeBytes = this.serializeRequest(spans)?.byteLength;
+    } catch (error) {
+      resultCallback({
+        code: ExportResultCode.FAILED,
+        error: error instanceof Error ? error : new Error(String(error)),
+      });
+      return;
+    }
+
+    if (serializedSizeBytes === undefined) {
+      resultCallback({
+        code: ExportResultCode.FAILED,
+        error: new Error("Failed to serialize OpenTelemetry span batch."),
+      });
+      return;
+    }
+
+    if (serializedSizeBytes > this.maxBatchSizeBytes) {
+      getGlobalLogger().warn(
+        "Dropping OpenTelemetry span batch because its serialized request exceeds the configured byte limit.",
+        {
+          maxBatchSizeBytes: this.maxBatchSizeBytes,
+          serializedSizeBytes,
+          spanCount: spans.length,
+        },
+      );
+      resultCallback({
+        code: ExportResultCode.FAILED,
+        error: new Error(
+          `Serialized OpenTelemetry span batch size ${serializedSizeBytes} bytes exceeds the configured limit of ${this.maxBatchSizeBytes} bytes.`,
+        ),
+      });
+      return;
+    }
+
+    this.delegate.export(spans, resultCallback);
+  }
+
+  forceFlush(): Promise<void> {
+    return this.delegate.forceFlush?.() ?? Promise.resolve();
+  }
+
+  shutdown(): Promise<void> {
+    return this.delegate.shutdown();
+  }
+}

packages/otel/src/span-processor.ts

@@ -23,6 +23,10 @@ import {
 } from "@opentelemetry/sdk-trace-base";
 
 import { MediaService } from "./MediaService.js";
+import {
+  SizeLimitedSpanExporter,
+  resolveMaxBatchSizeBytesFromEnvironment,
+} from "./size-limited-span-exporter.js";
 import { isDefaultExportSpan } from "./span-filter.js";
 
 /**
@@ -78,6 +82,8 @@ export type ShouldExportSpan = (params: { otelSpan: ReadableSpan }) => boolean;
 export interface LangfuseSpanProcessorParams {
   /**
    * Custom OpenTelemetry span exporter. If not provided, a default OTLP exporter will be used.
+   * Custom exporters bypass Langfuse's OTLP batch byte limit and must enforce
+   * any transport-specific request limits themselves.
    */
   exporter?: SpanExporter;
 
@@ -275,19 +281,22 @@ export class LangfuseSpanProcessor implements SpanProcessor {
         ? !["false", "0"].includes(envMediaUploadEnabled.toLowerCase())
         : true);
 
-    const exporter =
-      params?.exporter ??
-      new OTLPTraceExporter({
-        url: `${baseUrl}/api/public/otel/v1/traces`,
-        headers: {
-          Authorization: `Basic ${authHeaderValue}`,
-          "x-langfuse-sdk-name": "javascript",
-          "x-langfuse-sdk-version": LANGFUSE_SDK_VERSION,
-          "x-langfuse-public-key": publicKey ?? "<missing>",
-          ...params?.additionalHeaders,
-        },
-        timeoutMillis: timeoutSeconds * 1_000,
-      });
+    const exporter = params?.exporter
+      ? params.exporter
+      : new SizeLimitedSpanExporter({
+          maxBatchSizeBytes: resolveMaxBatchSizeBytesFromEnvironment(),
+          delegate: new OTLPTraceExporter({
+            url: `${baseUrl}/api/public/otel/v1/traces`,
+            headers: {
+              Authorization: `Basic ${authHeaderValue}`,
+              "x-langfuse-sdk-name": "javascript",
+              "x-langfuse-sdk-version": LANGFUSE_SDK_VERSION,
+              "x-langfuse-public-key": publicKey ?? "<missing>",
+              ...params?.additionalHeaders,
+            },
+            timeoutMillis: timeoutSeconds * 1_000,
+          }),
+        });
 
     this.processor =
       params?.exportMode === "immediate"

pnpm-lock.yaml

@@ -1,9 +1,14 @@
+import { createServer } from "node:http";
+import type { AddressInfo } from "node:net";
+
 import {
   LogLevel,
   configureGlobalLogger,
   resetGlobalLogger,
 } from "@langfuse/core";
+import { LangfuseSpanProcessor } from "@langfuse/otel";
 import { trace } from "@opentelemetry/api";
+import { NodeSDK } from "@opentelemetry/sdk-node";
 import { startObservation } from "@langfuse/tracing";
 import { describe, it, expect, beforeEach, afterEach, vi } from "vitest";
 
@@ -28,6 +33,7 @@ describe("LangfuseSpanProcessor E2E Tests", () => {
 
   afterEach(async () => {
     delete process.env.LANGFUSE_MEDIA_UPLOAD_ENABLED;
+    delete process.env.LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES;
     await teardownTestEnvironment(testEnv);
     vi.restoreAllMocks();
     resetGlobalLogger();
@@ -718,6 +724,57 @@ describe("LangfuseSpanProcessor E2E Tests", () => {
   });
 
   describe("Export Mode Selection", () => {
+    it("does not send an oversized request through the default exporter", async () => {
+      await teardownTestEnvironment(testEnv);
+      process.env.LANGFUSE_OTEL_MAX_BATCH_SIZE_BYTES = "1";
+      const warn = vi
+        .spyOn(console, "warn")
+        .mockImplementation(() => undefined);
+      let requestCount = 0;
+      const server = createServer((request, response) => {
+        requestCount += 1;
+        request.resume();
+        response.writeHead(200);
+        response.end();
+      });
+      await new Promise<void>((resolve) =>
+        server.listen(0, "127.0.0.1", resolve),
+      );
+      const { port } = server.address() as AddressInfo;
+      const spanProcessor = new LangfuseSpanProcessor({
+        baseUrl: `http://127.0.0.1:${port}`,
+        publicKey: "pk-lf-test",
+        secretKey: "sk-lf-test",
+        exportMode: "immediate",
+        mediaUploadEnabled: false,
+      });
+      const sdk = new NodeSDK({
+        spanProcessor,
+        instrumentations: [],
+      });
+      sdk.start();
+
+      try {
+        const span = startObservation("oversized-default-export");
+        span.end();
+        await spanProcessor.forceFlush();
+
+        expect(requestCount).toBe(0);
+        expect(warn).toHaveBeenCalledWith(
+          expect.stringContaining("Dropping OpenTelemetry span batch"),
+          expect.objectContaining({
+            maxBatchSizeBytes: 1,
+            spanCount: 1,
+          }),
+        );
+      } finally {
+        await sdk.shutdown();
+        await new Promise<void>((resolve, reject) =>
+          server.close((error) => (error ? reject(error) : resolve())),
+        );
+      }
+    });
+
     it("should use BatchSpanProcessor by default", async () => {
       // Default testEnv uses batched mode
       const span1 = startObservation("default-batch-1");