feat(webapp): parallelize OTLP ingest transform via a worker pool (flag-gated)

Behind OTEL_TRANSFORM_WORKER_POOL_ENABLED (default off), OTLP protobuf decode + convert
+ enrich (incl. LLM pricing match/calculateCost) run in a worker_threads pool instead of
the request event loop. The main thread stays the single DB reader (broadcasts compiled
model rows to workers via ModelPricingRegistry.loadFromModels) and keeps the existing
consolidated insertMany path, so ClickHouse insert cadence and part count are unchanged.

Adds: otlpTransformWorker (worker entry), otlpWorkerPool (hand-rolled worker_threads pool
with per-worker single-flight, pricing broadcast, respawn-on-crash), exporter exportXRaw
dispatch methods + flag, route wiring, a bundled build:otlpworker esbuild step, and
ModelPricingRegistry.loadFromModels/toSerializable (prisma-optional) + tests.

Local e2e on an isolated stack: flag-on sustained ~2.6x throughput (997 -> 2612 spans/s)
and kept the main-thread health check responsive (0.5-2.5s vs 4.6-10.4s) at identical load;
enriched events with registry-sourced cost verified in ClickHouse via the worker path.

Author: ericallam

apps/webapp/app/routes/otel.v1.logs.ts

@@ -1,7 +1,7 @@
 import type { ActionFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
 import { ExportLogsServiceRequest, ExportLogsServiceResponse } from "@trigger.dev/otlp-importer";
-import { otlpExporter } from "~/v3/otlpExporter.server";
+import { otlpExporter, otlpTransformWorkerPoolEnabled } from "~/v3/otlpExporter.server";
 
 export async function action({ request }: ActionFunctionArgs) {
   try {
@@ -17,6 +17,15 @@ export async function action({ request }: ActionFunctionArgs) {
     } else if (contentType.startsWith("application/x-protobuf")) {
       const buffer = await request.arrayBuffer();
 
+      if (otlpTransformWorkerPoolEnabled) {
+        await exporter.exportLogsRaw(new Uint8Array(buffer));
+
+        return new Response(
+          ExportLogsServiceResponse.encode(ExportLogsServiceResponse.create()).finish(),
+          { status: 200 }
+        );
+      }
+
       const exportRequest = ExportLogsServiceRequest.decode(new Uint8Array(buffer));
 
       const exportResponse = await exporter.exportLogs(exportRequest);

apps/webapp/app/routes/otel.v1.metrics.ts

@@ -4,7 +4,7 @@ import {
   ExportMetricsServiceRequest,
   ExportMetricsServiceResponse,
 } from "@trigger.dev/otlp-importer";
-import { otlpExporter } from "~/v3/otlpExporter.server";
+import { otlpExporter, otlpTransformWorkerPoolEnabled } from "~/v3/otlpExporter.server";
 
 export async function action({ request }: ActionFunctionArgs) {
   try {
@@ -21,6 +21,15 @@ export async function action({ request }: ActionFunctionArgs) {
       const exporter = await otlpExporter;
       const buffer = await request.arrayBuffer();
 
+      if (otlpTransformWorkerPoolEnabled) {
+        await exporter.exportMetricsRaw(new Uint8Array(buffer));
+
+        return new Response(
+          ExportMetricsServiceResponse.encode(ExportMetricsServiceResponse.create()).finish(),
+          { status: 200 }
+        );
+      }
+
       const exportRequest = ExportMetricsServiceRequest.decode(new Uint8Array(buffer));
 
       const exportResponse = await exporter.exportMetrics(exportRequest);

apps/webapp/app/routes/otel.v1.traces.ts

@@ -1,7 +1,7 @@
 import type { ActionFunctionArgs } from "@remix-run/server-runtime";
 import { json } from "@remix-run/server-runtime";
 import { ExportTraceServiceRequest, ExportTraceServiceResponse } from "@trigger.dev/otlp-importer";
-import { otlpExporter } from "~/v3/otlpExporter.server";
+import { otlpExporter, otlpTransformWorkerPoolEnabled } from "~/v3/otlpExporter.server";
 
 export async function action({ request }: ActionFunctionArgs) {
   try {
@@ -17,6 +17,15 @@ export async function action({ request }: ActionFunctionArgs) {
     } else if (contentType.startsWith("application/x-protobuf")) {
       const buffer = await request.arrayBuffer();
 
+      if (otlpTransformWorkerPoolEnabled) {
+        await exporter.exportTracesRaw(new Uint8Array(buffer));
+
+        return new Response(
+          ExportTraceServiceResponse.encode(ExportTraceServiceResponse.create()).finish(),
+          { status: 200 }
+        );
+      }
+
       const exportRequest = ExportTraceServiceRequest.decode(new Uint8Array(buffer));
 
       const exportResponse = await exporter.exportTraces(exportRequest);

apps/webapp/app/v3/otlpExporter.server.ts

@@ -30,6 +30,18 @@ import {
   isBoolValue,
   isStringValue,
 } from "./otlpTransform.server";
+import os from "node:os";
+import { getOtlpWorkerPool } from "./otlpWorkerPool.server";
+import { llmPricingRegistry } from "./llmPricingRegistry.server";
+
+// When enabled, decode+convert+enrich run in a worker pool; the main thread keeps the single
+// consolidated insert path (batching/part-count unchanged). Off = today's single-thread path.
+export const otlpTransformWorkerPoolEnabled =
+  process.env.OTEL_TRANSFORM_WORKER_POOL_ENABLED === "1";
+
+const OTEL_TRANSFORM_WORKER_POOL_SIZE = process.env.OTEL_TRANSFORM_WORKER_POOL_SIZE
+  ? parseInt(process.env.OTEL_TRANSFORM_WORKER_POOL_SIZE, 10)
+  : Math.max(1, (os.cpus()?.length ?? 2) - 2);
 
 type OTLPExporterConfig = {
   clickhouseFactory: ClickhouseFactory;
@@ -110,10 +122,55 @@ class OTLPExporter {
     });
   }
 
+  async exportTracesRaw(payload: Uint8Array): Promise<void> {
+    await this.#exportRawEvents("traces", payload);
+  }
+
+  async exportLogsRaw(payload: Uint8Array): Promise<void> {
+    await this.#exportRawEvents("logs", payload);
+  }
+
+  async exportMetricsRaw(payload: Uint8Array): Promise<void> {
+    await startSpan(this._tracer, "exportMetricsRaw", async (span) => {
+      const pool = await this.#pool();
+      const { rows } = await pool.runTransform("metrics", payload, this.#transformConfig());
+      span.setAttribute("metric_row_count", rows.length);
+      if (rows.length > 0) {
+        await this.#exportMetricRows(rows);
+      }
+    });
+  }
+
+  async #exportRawEvents(kind: "traces" | "logs", payload: Uint8Array): Promise<void> {
+    await startSpan(this._tracer, kind === "traces" ? "exportTracesRaw" : "exportLogsRaw", async (span) => {
+      const pool = await this.#pool();
+      const { eventsWithStores } = await pool.runTransform(kind, payload, this.#transformConfig());
+      const eventCount = await this.#exportEvents(eventsWithStores, true);
+      span.setAttribute("event_count", eventCount);
+    });
+  }
+
+  #transformConfig() {
+    return {
+      spanAttributeValueLengthLimit: this._spanAttributeValueLengthLimit,
+      defaultEventStore: env.EVENT_REPOSITORY_DEFAULT_STORE,
+    };
+  }
+
+  async #pool() {
+    await waitForLlmPricingReady();
+    const models =
+      llmPricingRegistry && llmPricingRegistry.isLoaded ? llmPricingRegistry.toSerializable() : [];
+    return getOtlpWorkerPool(OTEL_TRANSFORM_WORKER_POOL_SIZE, models);
+  }
+
   async #exportEvents(
-    eventsWithStores: { events: Array<CreateEventInput>; taskEventStore: string }[]
+    eventsWithStores: { events: Array<CreateEventInput>; taskEventStore: string }[],
+    alreadyEnriched = false
   ) {
-    await waitForLlmPricingReady();
+    if (!alreadyEnriched) {
+      await waitForLlmPricingReady();
+    }
 
     // Group by unique event repositories
     const routeCache = new Map<string, { key: string; repository: IEventRepository }>();
@@ -150,7 +207,7 @@ class OTLPExporter {
     let eventCount = 0;
 
     for (const [repoKey, { repository, events }] of groups) {
-      const enrichedEvents = enrichCreatableEvents(events);
+      const enrichedEvents = alreadyEnriched ? events : enrichCreatableEvents(events);
 
       this.#logEventsVerbose(enrichedEvents, `exportEvents ${repoKey}`);
 

apps/webapp/app/v3/otlpTransform.server.ts

@@ -18,6 +18,48 @@ import type {
   CreateEventInput,
 } from "./eventRepository/eventRepository.types";
 
+// Filters mirror OTLPExporter's #filterResource* methods, minus the debug logging, so a
+// worker can run them without the server logger.
+export function filterResourceSpans(resourceSpans: ResourceSpans[]): ResourceSpans[] {
+  return resourceSpans.filter((resourceSpan) => {
+    const triggerAttribute = resourceSpan.resource?.attributes.find(
+      (attribute) => attribute.key === SemanticInternalAttributes.TRIGGER
+    );
+    const executionEnvironmentAttribute = resourceSpan.resource?.attributes.find(
+      (attribute) => attribute.key === SemanticInternalAttributes.EXECUTION_ENVIRONMENT
+    );
+
+    if (!triggerAttribute && !executionEnvironmentAttribute) return true;
+
+    const executionEnvironment = isStringValue(executionEnvironmentAttribute?.value)
+      ? executionEnvironmentAttribute.value.stringValue
+      : undefined;
+    if (executionEnvironment === "trigger") return true;
+
+    return isBoolValue(triggerAttribute?.value) ? triggerAttribute.value.boolValue : false;
+  });
+}
+
+export function filterResourceLogs(resourceLogs: ResourceLogs[]): ResourceLogs[] {
+  return resourceLogs.filter((resourceLog) => {
+    const attribute = resourceLog.resource?.attributes.find(
+      (attribute) => attribute.key === SemanticInternalAttributes.TRIGGER
+    );
+    if (!attribute) return false;
+    return isBoolValue(attribute.value) ? attribute.value.boolValue : false;
+  });
+}
+
+export function filterResourceMetrics(resourceMetrics: ResourceMetrics[]): ResourceMetrics[] {
+  return resourceMetrics.filter((rm) => {
+    const triggerAttribute = rm.resource?.attributes.find(
+      (attribute) => attribute.key === SemanticInternalAttributes.TRIGGER
+    );
+    if (!triggerAttribute) return false;
+    return isBoolValue(triggerAttribute.value) ? triggerAttribute.value.boolValue : false;
+  });
+}
+
 export function convertLogsToCreateableEvents(
   resourceLog: ResourceLogs,
   spanAttributeValueLengthLimit: number,

apps/webapp/app/v3/otlpTransformWorker.ts

@@ -0,0 +1,102 @@
+import { parentPort, workerData } from "node:worker_threads";
+import { ModelPricingRegistry } from "@internal/llm-model-catalog";
+import type { LlmModelWithPricing } from "@internal/llm-model-catalog";
+import {
+  ExportLogsServiceRequest,
+  ExportMetricsServiceRequest,
+  ExportTraceServiceRequest,
+} from "@trigger.dev/otlp-importer";
+import {
+  convertLogsToCreateableEvents,
+  convertMetricsToClickhouseRows,
+  convertSpansToCreateableEvents,
+  filterResourceLogs,
+  filterResourceMetrics,
+  filterResourceSpans,
+} from "./otlpTransform.server";
+import { enrichCreatableEvents, setLlmPricingRegistry } from "./utils/enrichCreatableEvents.server";
+
+type TransformTask = {
+  id: number;
+  kind: "traces" | "logs" | "metrics";
+  payload: Uint8Array;
+  spanAttributeValueLengthLimit: number;
+  defaultEventStore: string;
+};
+
+type PricingUpdate = { type: "pricing"; models: LlmModelWithPricing[] };
+
+// The main thread is the only DB reader; it broadcasts the compiled model rows here.
+const registry = new ModelPricingRegistry();
+setLlmPricingRegistry(registry);
+
+function applyPricing(models: LlmModelWithPricing[]) {
+  registry.loadFromModels(models);
+}
+
+if (Array.isArray(workerData?.pricingModels)) {
+  applyPricing(workerData.pricingModels);
+}
+
+function runTask(task: TransformTask) {
+  const bytes = new Uint8Array(task.payload);
+
+  if (task.kind === "traces") {
+    const request = ExportTraceServiceRequest.decode(bytes);
+    const eventsWithStores = filterResourceSpans(request.resourceSpans).flatMap((resourceSpan) =>
+      convertSpansToCreateableEvents(
+        resourceSpan,
+        task.spanAttributeValueLengthLimit,
+        task.defaultEventStore
+      )
+    );
+    for (const group of eventsWithStores) {
+      group.events = enrichCreatableEvents(group.events);
+    }
+    return { eventsWithStores };
+  }
+
+  if (task.kind === "logs") {
+    const request = ExportLogsServiceRequest.decode(bytes);
+    const eventsWithStores = filterResourceLogs(request.resourceLogs).flatMap((resourceLog) =>
+      convertLogsToCreateableEvents(
+        resourceLog,
+        task.spanAttributeValueLengthLimit,
+        task.defaultEventStore
+      )
+    );
+    for (const group of eventsWithStores) {
+      group.events = enrichCreatableEvents(group.events);
+    }
+    return { eventsWithStores };
+  }
+
+  const request = ExportMetricsServiceRequest.decode(bytes);
+  const rows = filterResourceMetrics(request.resourceMetrics).flatMap((resourceMetrics) =>
+    convertMetricsToClickhouseRows(resourceMetrics, task.spanAttributeValueLengthLimit)
+  );
+  return { rows };
+}
+
+if (!parentPort) {
+  throw new Error("otlpTransformWorker must be run as a worker thread");
+}
+
+parentPort.on("message", (message: TransformTask | PricingUpdate) => {
+  if ("type" in message && message.type === "pricing") {
+    applyPricing(message.models);
+    return;
+  }
+
+  const task = message as TransformTask;
+  try {
+    const result = runTask(task);
+    parentPort!.postMessage({ id: task.id, ok: true, result });
+  } catch (error) {
+    parentPort!.postMessage({
+      id: task.id,
+      ok: false,
+      error: error instanceof Error ? error.message : String(error),
+    });
+  }
+});