From 606661734fac5b8ed6ad8e97af852882c00fb628 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Fri, 10 Jul 2026 22:25:08 +0100 Subject: [PATCH 01/10] refactor(webapp): extract OTLP transform into a worker-safe module Move the pure OTLP decode/convert/attribute helpers out of otlpExporter.server.ts into otlpTransform.server.ts, which has no server-singleton imports (env, clickhouse factory, event repository, prisma) so it can run inside a worker thread. The env-derived event-store default is now passed as a parameter. Behavior is identical; typecheck green. Prep for the OTLP ingest worker-pool spike. --- apps/webapp/app/v3/otlpExporter.server.ts | 954 +-------------------- apps/webapp/app/v3/otlpTransform.server.ts | 888 +++++++++++++++++++ 2 files changed, 906 insertions(+), 936 deletions(-) create mode 100644 apps/webapp/app/v3/otlpTransform.server.ts diff --git a/apps/webapp/app/v3/otlpExporter.server.ts b/apps/webapp/app/v3/otlpExporter.server.ts index 65680bbd2a..4700ab8a3b 100644 --- a/apps/webapp/app/v3/otlpExporter.server.ts +++ b/apps/webapp/app/v3/otlpExporter.server.ts @@ -2,43 +2,34 @@ import type { Tracer } from "@opentelemetry/api"; import { trace } from "@opentelemetry/api"; import { SemanticInternalAttributes } from "@trigger.dev/core/v3"; import type { - AnyValue, ExportLogsServiceRequest, ExportMetricsServiceRequest, ExportTraceServiceRequest, - KeyValue, - ResourceLogs, ResourceMetrics, - ResourceSpans, - Span, - Span_Event, } from "@trigger.dev/otlp-importer"; import { ExportLogsServiceResponse, ExportMetricsServiceResponse, ExportTraceServiceResponse, - SeverityNumber, - Span_SpanKind, - Status_StatusCode, } from "@trigger.dev/otlp-importer"; import type { MetricsV1Input } from "@internal/clickhouse"; import { logger } from "~/services/logger.server"; import type { ClickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server"; import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server"; - -import { generateSpanId } from "./eventRepository/common.server"; import { eventRepository } from "./eventRepository/eventRepository.server"; -import type { - CreatableEventKind, - CreatableEventStatus, - CreateEventInput, - IEventRepository, -} from "./eventRepository/eventRepository.types"; +import type { CreateEventInput, IEventRepository } from "./eventRepository/eventRepository.types"; import { startSpan } from "./tracing.server"; import { enrichCreatableEvents } from "./utils/enrichCreatableEvents.server"; import { waitForLlmPricingReady } from "./llmPricingRegistry.server"; import { env } from "~/env.server"; import { singleton } from "~/utils/singleton"; +import { + convertLogsToCreateableEvents, + convertMetricsToClickhouseRows, + convertSpansToCreateableEvents, + isBoolValue, + isStringValue, +} from "./otlpTransform.server"; type OTLPExporterConfig = { clickhouseFactory: ClickhouseFactory; @@ -65,7 +56,11 @@ class OTLPExporter { const eventsWithStores = this.#filterResourceSpans(request.resourceSpans).flatMap( (resourceSpan) => { - return convertSpansToCreateableEvents(resourceSpan, this._spanAttributeValueLengthLimit); + return convertSpansToCreateableEvents( + resourceSpan, + this._spanAttributeValueLengthLimit, + env.EVENT_REPOSITORY_DEFAULT_STORE + ); } ); @@ -99,7 +94,11 @@ class OTLPExporter { const eventsWithStores = this.#filterResourceLogs(request.resourceLogs).flatMap( (resourceLog) => { - return convertLogsToCreateableEvents(resourceLog, this._spanAttributeValueLengthLimit); + return convertLogsToCreateableEvents( + resourceLog, + this._spanAttributeValueLengthLimit, + env.EVENT_REPOSITORY_DEFAULT_STORE + ); } ); @@ -285,923 +284,6 @@ class OTLPExporter { } } -function convertLogsToCreateableEvents( - resourceLog: ResourceLogs, - spanAttributeValueLengthLimit: number -): { events: Array; taskEventStore: string } { - const resourceAttributes = resourceLog.resource?.attributes ?? []; - - const resourceProperties = extractEventProperties(resourceAttributes); - - const userDefinedResourceAttributes = truncateAttributes( - convertKeyValueItemsToMap(resourceAttributes ?? [], [], undefined, [ - SemanticInternalAttributes.USAGE, - SemanticInternalAttributes.SPAN, - SemanticInternalAttributes.METADATA, - SemanticInternalAttributes.STYLE, - SemanticInternalAttributes.METRIC_EVENTS, - SemanticInternalAttributes.TRIGGER, - "process", - "sdk", - "service", - "ctx", - "cli", - "cloud", - ]), - spanAttributeValueLengthLimit - ); - - const taskEventStore = - extractStringAttribute(resourceAttributes, [SemanticInternalAttributes.TASK_EVENT_STORE]) ?? - env.EVENT_REPOSITORY_DEFAULT_STORE; - - const events = resourceLog.scopeLogs.flatMap((scopeLog) => { - return scopeLog.logRecords - .map((log) => { - const logLevel = logLevelToEventLevel(log.severityNumber); - - if (!log.traceId || !log.spanId) { - return; - } - - const logProperties = extractEventProperties( - log.attributes ?? [], - SemanticInternalAttributes.METADATA - ); - - const properties = - truncateAttributes( - convertKeyValueItemsToMap(log.attributes ?? [], [], undefined, [ - SemanticInternalAttributes.USAGE, - SemanticInternalAttributes.SPAN, - SemanticInternalAttributes.METADATA, - SemanticInternalAttributes.STYLE, - SemanticInternalAttributes.METRIC_EVENTS, - SemanticInternalAttributes.TRIGGER, - ]), - spanAttributeValueLengthLimit - ) ?? {}; - - return { - traceId: binaryToHex(log.traceId), - spanId: generateSpanId(), - parentId: binaryToHex(log.spanId), - message: isStringValue(log.body) - ? log.body.stringValue.slice(0, 4096) - : `${log.severityText} log`, - isPartial: false, - kind: "INTERNAL" as const, - level: logLevelToEventLevel(log.severityNumber), - isError: logLevel === "ERROR", - status: logLevelToEventStatus(log.severityNumber), - startTime: log.timeUnixNano, - properties, - resourceProperties: userDefinedResourceAttributes, - style: convertKeyValueItemsToMap( - pickAttributes(log.attributes ?? [], SemanticInternalAttributes.STYLE), - [] - ), - metadata: logProperties.metadata ?? resourceProperties.metadata ?? {}, - environmentId: - logProperties.environmentId ?? resourceProperties.environmentId ?? "unknown", - environmentType: "DEVELOPMENT" as const, // We've deprecated this but we need to keep it for backwards compatibility - organizationId: - logProperties.organizationId ?? resourceProperties.organizationId ?? "unknown", - projectId: logProperties.projectId ?? resourceProperties.projectId ?? "unknown", - runId: logProperties.runId ?? resourceProperties.runId ?? "unknown", - taskSlug: logProperties.taskSlug ?? resourceProperties.taskSlug ?? "unknown", - machineId: logProperties.machineId ?? resourceProperties.machineId, - attemptNumber: - extractNumberAttribute( - log.attributes ?? [], - [SemanticInternalAttributes.METADATA, SemanticInternalAttributes.ATTEMPT_NUMBER].join( - "." - ) - ) ?? resourceProperties.attemptNumber, - }; - }) - .filter(Boolean); - }); - - return { events, taskEventStore }; -} - -function convertSpansToCreateableEvents( - resourceSpan: ResourceSpans, - spanAttributeValueLengthLimit: number -): { events: Array; taskEventStore: string } { - const resourceAttributes = resourceSpan.resource?.attributes ?? []; - - const resourceProperties = extractEventProperties(resourceAttributes); - - const userDefinedResourceAttributes = truncateAttributes( - convertKeyValueItemsToMap(resourceAttributes ?? [], [], undefined, [ - SemanticInternalAttributes.USAGE, - SemanticInternalAttributes.SPAN, - SemanticInternalAttributes.METADATA, - SemanticInternalAttributes.STYLE, - SemanticInternalAttributes.METRIC_EVENTS, - SemanticInternalAttributes.TRIGGER, - "process", - "sdk", - "service", - "ctx", - "cli", - "cloud", - ]), - spanAttributeValueLengthLimit - ); - - const taskEventStore = - extractStringAttribute(resourceAttributes, [SemanticInternalAttributes.TASK_EVENT_STORE]) ?? - env.EVENT_REPOSITORY_DEFAULT_STORE; - - const events = resourceSpan.scopeSpans.flatMap((scopeSpan) => { - return scopeSpan.spans - .map((span) => { - const isPartial = isPartialSpan(span); - - if (!span.traceId || !span.spanId) { - return; - } - - const spanProperties = extractEventProperties( - span.attributes ?? [], - SemanticInternalAttributes.METADATA - ); - - const runTags = extractArrayAttribute( - span.attributes ?? [], - SemanticInternalAttributes.RUN_TAGS - ); - - const properties = - truncateAttributes( - convertKeyValueItemsToMap(span.attributes ?? [], [], undefined, [ - SemanticInternalAttributes.USAGE, - SemanticInternalAttributes.SPAN, - SemanticInternalAttributes.METADATA, - SemanticInternalAttributes.STYLE, - SemanticInternalAttributes.METRIC_EVENTS, - SemanticInternalAttributes.TRIGGER, - ]), - spanAttributeValueLengthLimit - ) ?? {}; - - return { - traceId: binaryToHex(span.traceId), - spanId: isPartial - ? extractStringAttribute( - span?.attributes ?? [], - SemanticInternalAttributes.SPAN_ID, - binaryToHex(span.spanId) - ) - : binaryToHex(span.spanId), - parentId: binaryToHex(span.parentSpanId), - message: span.name, - isPartial, - isError: span.status?.code === Status_StatusCode.ERROR, - kind: spanKindToEventKind(span.kind), - level: "TRACE" as const, - status: spanStatusToEventStatus(span.status), - startTime: span.startTimeUnixNano, - events: spanEventsToEventEvents(span.events ?? []), - duration: span.endTimeUnixNano - span.startTimeUnixNano, - properties, - resourceProperties: userDefinedResourceAttributes, - style: convertKeyValueItemsToMap( - pickAttributes(span.attributes ?? [], SemanticInternalAttributes.STYLE), - [] - ), - metadata: spanProperties.metadata ?? resourceProperties.metadata ?? {}, - environmentId: - spanProperties.environmentId ?? resourceProperties.environmentId ?? "unknown", - environmentType: "DEVELOPMENT" as const, - organizationId: - spanProperties.organizationId ?? resourceProperties.organizationId ?? "unknown", - projectId: spanProperties.projectId ?? resourceProperties.projectId ?? "unknown", - runId: spanProperties.runId ?? resourceProperties.runId ?? "unknown", - taskSlug: spanProperties.taskSlug ?? resourceProperties.taskSlug ?? "unknown", - machineId: spanProperties.machineId ?? resourceProperties.machineId, - runTags, - attemptNumber: - extractNumberAttribute( - span.attributes ?? [], - [SemanticInternalAttributes.METADATA, SemanticInternalAttributes.ATTEMPT_NUMBER].join( - "." - ) - ) ?? resourceProperties.attemptNumber, - }; - }) - .filter(Boolean); - }); - - return { events, taskEventStore }; -} - -function floorToTenSecondBucket(timeUnixNano: bigint | number): string { - const epochMs = Number(BigInt(timeUnixNano) / BigInt(1_000_000)); - const flooredMs = Math.floor(epochMs / 10_000) * 10_000; - const date = new Date(flooredMs); - // Format as ClickHouse DateTime: YYYY-MM-DD HH:MM:SS - return date - .toISOString() - .replace("T", " ") - .replace(/\.\d{3}Z$/, ""); -} - -function convertMetricsToClickhouseRows( - resourceMetrics: ResourceMetrics, - spanAttributeValueLengthLimit: number -): MetricsV1Input[] { - const resourceAttributes = resourceMetrics.resource?.attributes ?? []; - const resourceProperties = extractEventProperties(resourceAttributes); - - const organizationId = resourceProperties.organizationId ?? "unknown"; - const projectId = resourceProperties.projectId ?? "unknown"; - const environmentId = resourceProperties.environmentId ?? "unknown"; - const resourceCtx = { - taskSlug: resourceProperties.taskSlug, - runId: resourceProperties.runId, - attemptNumber: resourceProperties.attemptNumber, - machineId: extractStringAttribute(resourceAttributes, SemanticInternalAttributes.MACHINE_ID), - workerId: extractStringAttribute(resourceAttributes, SemanticInternalAttributes.WORKER_ID), - workerVersion: extractStringAttribute( - resourceAttributes, - SemanticInternalAttributes.WORKER_VERSION - ), - }; - - const rows: MetricsV1Input[] = []; - - for (const scopeMetrics of resourceMetrics.scopeMetrics) { - for (const metric of scopeMetrics.metrics) { - const metricName = metric.name; - - // Process gauge data points - if (metric.gauge) { - for (const dp of metric.gauge.dataPoints) { - const value: number = - dp.asDouble !== undefined ? dp.asDouble : dp.asInt !== undefined ? Number(dp.asInt) : 0; - const resolved = resolveDataPointContext(dp.attributes ?? [], resourceCtx); - - rows.push({ - organization_id: organizationId, - project_id: projectId, - environment_id: environmentId, - metric_name: metricName, - metric_type: "gauge", - metric_subject: resolved.machineId ?? "unknown", - bucket_start: floorToTenSecondBucket(dp.timeUnixNano), - value, - attributes: resolved.attributes, - }); - } - } - - // Process sum data points - if (metric.sum) { - for (const dp of metric.sum.dataPoints) { - const value: number = - dp.asDouble !== undefined ? dp.asDouble : dp.asInt !== undefined ? Number(dp.asInt) : 0; - const resolved = resolveDataPointContext(dp.attributes ?? [], resourceCtx); - - rows.push({ - organization_id: organizationId, - project_id: projectId, - environment_id: environmentId, - metric_name: metricName, - metric_type: "sum", - metric_subject: resolved.machineId ?? "unknown", - bucket_start: floorToTenSecondBucket(dp.timeUnixNano), - value, - attributes: resolved.attributes, - }); - } - } - - // Process histogram data points - if (metric.histogram) { - for (const dp of metric.histogram.dataPoints) { - const resolved = resolveDataPointContext(dp.attributes ?? [], resourceCtx); - const count = Number(dp.count); - const sum = dp.sum ?? 0; - - rows.push({ - organization_id: organizationId, - project_id: projectId, - environment_id: environmentId, - metric_name: metricName, - metric_type: "histogram", - metric_subject: resolved.machineId ?? "unknown", - bucket_start: floorToTenSecondBucket(dp.timeUnixNano), - value: count > 0 ? sum / count : 0, - attributes: resolved.attributes, - }); - } - } - } - } - - return rows; -} - -// Prefixes injected by TaskContextMetricExporter — these are extracted into -// the nested `trigger` key and should not appear as top-level user attributes. -const INTERNAL_METRIC_ATTRIBUTE_PREFIXES = ["ctx.", "worker."]; - -interface ResourceContext { - taskSlug: string | undefined; - runId: string | undefined; - attemptNumber: number | undefined; - machineId: string | undefined; - workerId: string | undefined; - workerVersion: string | undefined; -} - -function resolveDataPointContext( - dpAttributes: KeyValue[], - resourceCtx: ResourceContext -): { - machineId: string | undefined; - attributes: Record; -} { - const runId = - resourceCtx.runId ?? extractStringAttribute(dpAttributes, SemanticInternalAttributes.RUN_ID); - const taskSlug = - resourceCtx.taskSlug ?? - extractStringAttribute(dpAttributes, SemanticInternalAttributes.TASK_SLUG); - const attemptNumber = - resourceCtx.attemptNumber ?? - extractNumberAttribute(dpAttributes, SemanticInternalAttributes.ATTEMPT_NUMBER); - const machineId = - resourceCtx.machineId ?? - extractStringAttribute(dpAttributes, SemanticInternalAttributes.MACHINE_ID); - const workerId = - resourceCtx.workerId ?? - extractStringAttribute(dpAttributes, SemanticInternalAttributes.WORKER_ID); - const workerVersion = - resourceCtx.workerVersion ?? - extractStringAttribute(dpAttributes, SemanticInternalAttributes.WORKER_VERSION); - const machineName = extractStringAttribute( - dpAttributes, - SemanticInternalAttributes.MACHINE_PRESET_NAME - ); - const environmentType = extractStringAttribute( - dpAttributes, - SemanticInternalAttributes.ENVIRONMENT_TYPE - ); - - // Build the trigger context object with only defined values - const trigger: Record = {}; - if (runId) trigger.run_id = runId; - if (taskSlug) trigger.task_slug = taskSlug; - if (attemptNumber !== undefined) trigger.attempt_number = attemptNumber; - if (machineId) trigger.machine_id = machineId; - if (machineName) trigger.machine_name = machineName; - if (workerId) trigger.worker_id = workerId; - if (workerVersion) trigger.worker_version = workerVersion; - if (environmentType) trigger.environment_type = environmentType; - - // Build user attributes, filtering out internal ctx/worker keys - const result: Record = {}; - - if (Object.keys(trigger).length > 0) { - result.trigger = trigger; - } - - for (const attr of dpAttributes) { - if (INTERNAL_METRIC_ATTRIBUTE_PREFIXES.some((prefix) => attr.key.startsWith(prefix))) { - continue; - } - - if (isStringValue(attr.value)) { - result[attr.key] = attr.value.stringValue; - } else if (isIntValue(attr.value)) { - result[attr.key] = Number(attr.value.intValue); - } else if (isDoubleValue(attr.value)) { - result[attr.key] = attr.value.doubleValue; - } else if (isBoolValue(attr.value)) { - result[attr.key] = attr.value.boolValue; - } - } - - return { machineId, attributes: result }; -} - -function extractEventProperties(attributes: KeyValue[], prefix?: string) { - return { - metadata: convertSelectedKeyValueItemsToMap(attributes, [SemanticInternalAttributes.METADATA]), - environmentId: extractStringAttribute(attributes, [ - prefix, - SemanticInternalAttributes.ENVIRONMENT_ID, - ]), - organizationId: extractStringAttribute(attributes, [ - prefix, - SemanticInternalAttributes.ORGANIZATION_ID, - ]), - projectId: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.PROJECT_ID]), - runId: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.RUN_ID]), - attemptNumber: extractNumberAttribute(attributes, [ - prefix, - SemanticInternalAttributes.ATTEMPT_NUMBER, - ]), - taskSlug: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.TASK_SLUG]), - machineId: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.MACHINE_ID]), - }; -} - -function pickAttributes(attributes: KeyValue[], prefix: string): KeyValue[] { - return attributes - .filter((attribute) => attribute.key.startsWith(prefix)) - .map((attribute) => { - return { - key: attribute.key.replace(`${prefix}.`, ""), - value: attribute.value, - }; - }); -} - -function convertKeyValueItemsToMap( - attributes: KeyValue[], - filteredKeys: string[] = [], - prefix?: string, - filteredPrefixes: string[] = [] -): Record | undefined { - if (!attributes) return; - if (!attributes.length) return; - - let filteredAttributes = attributes.filter((attribute) => !filteredKeys.includes(attribute.key)); - - if (!filteredAttributes.length) return; - - if (filteredPrefixes.length) { - filteredAttributes = filteredAttributes.filter( - (attribute) => !filteredPrefixes.some((prefix) => attribute.key.startsWith(prefix)) - ); - } - - if (!filteredAttributes.length) return; - - const result = filteredAttributes.reduce( - (map: Record, attribute) => { - map[`${prefix ? `${prefix}.` : ""}${attribute.key}`] = isStringValue(attribute.value) - ? attribute.value.stringValue - : isIntValue(attribute.value) - ? Number(attribute.value.intValue) - : isDoubleValue(attribute.value) - ? attribute.value.doubleValue - : isBoolValue(attribute.value) - ? attribute.value.boolValue - : isBytesValue(attribute.value) - ? binaryToHex(attribute.value.bytesValue) - : isArrayValue(attribute.value) - ? serializeArrayValue(attribute.value.arrayValue!.values) - : undefined; - - return map; - }, - {} - ); - - return result; -} - -function convertSelectedKeyValueItemsToMap( - attributes: KeyValue[], - selectedPrefixes: string[] = [], - prefix?: string -): Record | undefined { - if (!attributes) return; - if (!attributes.length) return; - - let selectedAttributes = attributes.filter((attribute) => - selectedPrefixes.some((prefix) => attribute.key.startsWith(prefix)) - ); - - if (!selectedAttributes.length) return; - - const result = selectedAttributes.reduce( - (map: Record, attribute) => { - map[`${prefix ? `${prefix}.` : ""}${attribute.key}`] = isStringValue(attribute.value) - ? attribute.value.stringValue - : isIntValue(attribute.value) - ? Number(attribute.value.intValue) - : isDoubleValue(attribute.value) - ? attribute.value.doubleValue - : isBoolValue(attribute.value) - ? attribute.value.boolValue - : isBytesValue(attribute.value) - ? binaryToHex(attribute.value.bytesValue) - : isArrayValue(attribute.value) - ? serializeArrayValue(attribute.value.arrayValue!.values) - : undefined; - - return map; - }, - {} - ); - - return result; -} -function spanEventsToEventEvents(events: Span_Event[]): CreateEventInput["events"] { - return events.map((event) => { - return { - name: event.name, - time: convertUnixNanoToDate(event.timeUnixNano), - properties: convertKeyValueItemsToMap(event.attributes ?? []), - }; - }); -} - -function spanStatusToEventStatus(status: Span["status"]): CreatableEventStatus { - if (!status) return "UNSET"; - - switch (status.code) { - case Status_StatusCode.OK: { - return "OK"; - } - case Status_StatusCode.ERROR: { - return "ERROR"; - } - case Status_StatusCode.UNSET: { - return "UNSET"; - } - default: { - return "UNSET"; - } - } -} - -function spanKindToEventKind(kind: Span["kind"]): CreatableEventKind { - switch (kind) { - case Span_SpanKind.CLIENT: { - return "CLIENT"; - } - case Span_SpanKind.SERVER: { - return "SERVER"; - } - case Span_SpanKind.CONSUMER: { - return "CONSUMER"; - } - case Span_SpanKind.PRODUCER: { - return "PRODUCER"; - } - default: { - return "INTERNAL"; - } - } -} - -function logLevelToEventLevel(level: SeverityNumber): CreateEventInput["level"] { - switch (level) { - case SeverityNumber.TRACE: - case SeverityNumber.TRACE2: - case SeverityNumber.TRACE3: - case SeverityNumber.TRACE4: { - return "TRACE"; - } - case SeverityNumber.DEBUG: - case SeverityNumber.DEBUG2: - case SeverityNumber.DEBUG3: - case SeverityNumber.DEBUG4: { - return "DEBUG"; - } - case SeverityNumber.INFO: - case SeverityNumber.INFO2: - case SeverityNumber.INFO3: - case SeverityNumber.INFO4: { - return "INFO"; - } - case SeverityNumber.WARN: - case SeverityNumber.WARN2: - case SeverityNumber.WARN3: - case SeverityNumber.WARN4: { - return "WARN"; - } - case SeverityNumber.ERROR: - case SeverityNumber.ERROR2: - case SeverityNumber.ERROR3: - case SeverityNumber.ERROR4: { - return "ERROR"; - } - case SeverityNumber.FATAL: - case SeverityNumber.FATAL2: - case SeverityNumber.FATAL3: - case SeverityNumber.FATAL4: { - return "ERROR"; - } - default: { - return "INFO"; - } - } -} - -function logLevelToEventStatus(level: SeverityNumber): CreatableEventStatus { - switch (level) { - case SeverityNumber.TRACE: - case SeverityNumber.TRACE2: - case SeverityNumber.TRACE3: - case SeverityNumber.TRACE4: { - return "OK"; - } - case SeverityNumber.DEBUG: - case SeverityNumber.DEBUG2: - case SeverityNumber.DEBUG3: - case SeverityNumber.DEBUG4: { - return "OK"; - } - case SeverityNumber.INFO: - case SeverityNumber.INFO2: - case SeverityNumber.INFO3: - case SeverityNumber.INFO4: { - return "OK"; - } - case SeverityNumber.WARN: - case SeverityNumber.WARN2: - case SeverityNumber.WARN3: - case SeverityNumber.WARN4: { - return "OK"; - } - case SeverityNumber.ERROR: - case SeverityNumber.ERROR2: - case SeverityNumber.ERROR3: - case SeverityNumber.ERROR4: { - return "ERROR"; - } - case SeverityNumber.FATAL: - case SeverityNumber.FATAL2: - case SeverityNumber.FATAL3: - case SeverityNumber.FATAL4: { - return "ERROR"; - } - default: { - return "OK"; - } - } -} - -function convertUnixNanoToDate(unixNano: bigint | number): Date { - return new Date(Number(BigInt(unixNano) / BigInt(1_000_000))); -} - -function extractStringAttribute( - attributes: KeyValue[], - name: string | Array -): string | undefined; -function extractStringAttribute( - attributes: KeyValue[], - name: string | Array, - fallback: string -): string; -function extractStringAttribute( - attributes: KeyValue[], - name: string | Array, - fallback?: string -): string | undefined { - const key = Array.isArray(name) ? name.filter(Boolean).join(".") : name; - - const attribute = attributes.find((attribute) => attribute.key === key); - - if (!attribute) return fallback; - - return isStringValue(attribute?.value) ? attribute.value.stringValue : fallback; -} - -function extractNumberAttribute( - attributes: KeyValue[], - name: string | Array -): number | undefined; -function extractNumberAttribute( - attributes: KeyValue[], - name: string | Array, - fallback: number -): number; -function extractNumberAttribute( - attributes: KeyValue[], - name: string | Array, - fallback?: number -): number | undefined { - const key = Array.isArray(name) ? name.filter(Boolean).join(".") : name; - - const attribute = attributes.find((attribute) => attribute.key === key); - - if (!attribute) return fallback; - - return isIntValue(attribute?.value) ? Number(attribute.value.intValue) : fallback; -} - -// eslint-disable-next-line no-unused-vars -function extractDoubleAttribute( - attributes: KeyValue[], - name: string | Array -): number | undefined; -// eslint-disable-next-line no-unused-vars -function extractDoubleAttribute( - attributes: KeyValue[], - name: string | Array, - fallback: number -): number; -function extractDoubleAttribute( - attributes: KeyValue[], - name: string | Array, - fallback?: number -): number | undefined { - const key = Array.isArray(name) ? name.filter(Boolean).join(".") : name; - - const attribute = attributes.find((attribute) => attribute.key === key); - - if (!attribute) return fallback; - - return isDoubleValue(attribute?.value) ? Number(attribute.value.doubleValue) : fallback; -} - -// eslint-disable-next-line no-unused-vars -function extractBooleanAttribute( - attributes: KeyValue[], - name: string | Array -): boolean | undefined; -// eslint-disable-next-line no-unused-vars -function extractBooleanAttribute( - attributes: KeyValue[], - name: string | Array, - fallback: boolean -): boolean; -function extractBooleanAttribute( - attributes: KeyValue[], - name: string | Array, - fallback?: boolean -): boolean | undefined { - const key = Array.isArray(name) ? name.filter(Boolean).join(".") : name; - - const attribute = attributes.find((attribute) => attribute.key === key); - - if (!attribute) return fallback; - - return isBoolValue(attribute?.value) ? attribute.value.boolValue : fallback; -} - -function extractArrayAttribute( - attributes: KeyValue[], - name: string | Array -): string[] | undefined { - const key = Array.isArray(name) ? name.filter(Boolean).join(".") : name; - - const attribute = attributes.find((attribute) => attribute.key === key); - - if (!attribute?.value?.arrayValue?.values) return undefined; - - return attribute.value.arrayValue.values - .filter((v): v is { stringValue: string } => isStringValue(v)) - .map((v) => v.stringValue); -} - -function isPartialSpan(span: Span): boolean { - if (!span.attributes) return false; - - const attribute = span.attributes.find( - (attribute) => attribute.key === SemanticInternalAttributes.SPAN_PARTIAL - ); - - if (!attribute) return false; - - return isBoolValue(attribute.value) ? attribute.value.boolValue : false; -} - -function isBoolValue(value: AnyValue | undefined): value is { boolValue: boolean } { - if (!value) return false; - - return typeof value.boolValue === "boolean"; -} - -function isStringValue(value: AnyValue | undefined): value is { stringValue: string } { - if (!value) return false; - - return typeof value.stringValue === "string"; -} - -function isIntValue(value: AnyValue | undefined): value is { intValue: bigint } { - if (!value) return false; - - return typeof value.intValue === "number" || typeof value.intValue === "bigint"; -} - -function isDoubleValue(value: AnyValue | undefined): value is { doubleValue: number } { - if (!value) return false; - - return typeof value.doubleValue === "number"; -} - -function isBytesValue(value: AnyValue | undefined): value is { bytesValue: Buffer } { - if (!value) return false; - - return Buffer.isBuffer(value.bytesValue); -} - -function isArrayValue( - value: AnyValue | undefined -): value is { arrayValue: { values: AnyValue[] } } { - if (!value) return false; - - return value.arrayValue != null && Array.isArray(value.arrayValue.values); -} - -/** - * Serialize an OTEL array value into a JSON string. - * For arrays of strings, produces a JSON array: `["item1","item2"]` - * For mixed types, extracts primitives and serializes. - */ -function serializeArrayValue(values: AnyValue[]): string { - const items = values.map((v) => { - if (isStringValue(v)) return v.stringValue; - if (isIntValue(v)) return Number(v.intValue); - if (isDoubleValue(v)) return v.doubleValue; - if (isBoolValue(v)) return v.boolValue; - return null; - }); - - return JSON.stringify(items); -} - -function binaryToHex(buffer: Buffer | string): string; -function binaryToHex(buffer: Buffer | string | undefined): string | undefined; -function binaryToHex(buffer: Buffer | string | undefined): string | undefined { - if (!buffer) return undefined; - if (typeof buffer === "string") return buffer; - - return Buffer.from(Array.from(buffer)).toString("hex"); -} - -function truncateAttributes( - attributes: Record | undefined, - maximumLength: number = 1024 -): Record | undefined { - if (!attributes) return undefined; - - const truncatedAttributes: Record = {}; - - for (const [key, value] of Object.entries(attributes)) { - if (!key) continue; - - if (typeof value === "string") { - truncatedAttributes[key] = truncateAndDetectUnpairedSurrogate(value, maximumLength); - } else { - truncatedAttributes[key] = value; - } - } - - return truncatedAttributes; -} - -function truncateAndDetectUnpairedSurrogate(str: string, maximumLength: number): string { - const truncatedString = smartTruncateString(str, maximumLength); - - if (hasUnpairedSurrogateAtEnd(truncatedString)) { - return smartTruncateString(truncatedString, [...truncatedString].length - 1); - } - - return truncatedString; -} - -const ASCII_ONLY_REGEX = /^[\p{ASCII}]*$/u; - -function smartTruncateString(str: string, maximumLength: number): string { - if (!str) return ""; - if (str.length <= maximumLength) return str; - - const checkLength = Math.min(str.length, maximumLength * 2 + 2); - - if (ASCII_ONLY_REGEX.test(str.slice(0, checkLength))) { - return str.slice(0, maximumLength); - } - - return [...str.slice(0, checkLength)].slice(0, maximumLength).join(""); -} - -function hasUnpairedSurrogateAtEnd(str: string): boolean { - if (str.length === 0) return false; - - const lastCode = str.charCodeAt(str.length - 1); - - // Check if last character is an unpaired high surrogate - if (lastCode >= 0xd800 && lastCode <= 0xdbff) { - return true; // High surrogate at end = unpaired - } - - // Check if last character is an unpaired low surrogate - if (lastCode >= 0xdc00 && lastCode <= 0xdfff) { - // Low surrogate is only valid if preceded by high surrogate - if (str.length === 1) return true; // Single low surrogate - - const secondLastCode = str.charCodeAt(str.length - 2); - if (secondLastCode < 0xd800 || secondLastCode > 0xdbff) { - return true; // Low surrogate not preceded by high surrogate - } - } - - return false; -} - export const otlpExporter = singleton("otlpExporter", initializeOTLPExporter); async function initializeOTLPExporter() { diff --git a/apps/webapp/app/v3/otlpTransform.server.ts b/apps/webapp/app/v3/otlpTransform.server.ts new file mode 100644 index 0000000000..1e995253e3 --- /dev/null +++ b/apps/webapp/app/v3/otlpTransform.server.ts @@ -0,0 +1,888 @@ +// Worker-safe OTLP transform: no server singletons (env/clickhouse/repository/prisma). +import { SemanticInternalAttributes } from "@trigger.dev/core/v3"; +import type { + AnyValue, + KeyValue, + ResourceLogs, + ResourceMetrics, + ResourceSpans, + Span, + Span_Event, +} from "@trigger.dev/otlp-importer"; +import { SeverityNumber, Span_SpanKind, Status_StatusCode } from "@trigger.dev/otlp-importer"; +import type { MetricsV1Input } from "@internal/clickhouse"; +import { generateSpanId } from "./eventRepository/common.server"; +import type { + CreatableEventKind, + CreatableEventStatus, + CreateEventInput, +} from "./eventRepository/eventRepository.types"; + +export function convertLogsToCreateableEvents( + resourceLog: ResourceLogs, + spanAttributeValueLengthLimit: number, + defaultEventStore: string +): { events: Array; taskEventStore: string } { + const resourceAttributes = resourceLog.resource?.attributes ?? []; + + const resourceProperties = extractEventProperties(resourceAttributes); + + const userDefinedResourceAttributes = truncateAttributes( + convertKeyValueItemsToMap(resourceAttributes ?? [], [], undefined, [ + SemanticInternalAttributes.USAGE, + SemanticInternalAttributes.SPAN, + SemanticInternalAttributes.METADATA, + SemanticInternalAttributes.STYLE, + SemanticInternalAttributes.METRIC_EVENTS, + SemanticInternalAttributes.TRIGGER, + "process", + "sdk", + "service", + "ctx", + "cli", + "cloud", + ]), + spanAttributeValueLengthLimit + ); + + const taskEventStore = + extractStringAttribute(resourceAttributes, [SemanticInternalAttributes.TASK_EVENT_STORE]) ?? + defaultEventStore; + + const events = resourceLog.scopeLogs.flatMap((scopeLog) => { + return scopeLog.logRecords + .map((log) => { + const logLevel = logLevelToEventLevel(log.severityNumber); + + if (!log.traceId || !log.spanId) { + return; + } + + const logProperties = extractEventProperties( + log.attributes ?? [], + SemanticInternalAttributes.METADATA + ); + + const properties = + truncateAttributes( + convertKeyValueItemsToMap(log.attributes ?? [], [], undefined, [ + SemanticInternalAttributes.USAGE, + SemanticInternalAttributes.SPAN, + SemanticInternalAttributes.METADATA, + SemanticInternalAttributes.STYLE, + SemanticInternalAttributes.METRIC_EVENTS, + SemanticInternalAttributes.TRIGGER, + ]), + spanAttributeValueLengthLimit + ) ?? {}; + + return { + traceId: binaryToHex(log.traceId), + spanId: generateSpanId(), + parentId: binaryToHex(log.spanId), + message: isStringValue(log.body) + ? log.body.stringValue.slice(0, 4096) + : `${log.severityText} log`, + isPartial: false, + kind: "INTERNAL" as const, + level: logLevelToEventLevel(log.severityNumber), + isError: logLevel === "ERROR", + status: logLevelToEventStatus(log.severityNumber), + startTime: log.timeUnixNano, + properties, + resourceProperties: userDefinedResourceAttributes, + style: convertKeyValueItemsToMap( + pickAttributes(log.attributes ?? [], SemanticInternalAttributes.STYLE), + [] + ), + metadata: logProperties.metadata ?? resourceProperties.metadata ?? {}, + environmentId: + logProperties.environmentId ?? resourceProperties.environmentId ?? "unknown", + environmentType: "DEVELOPMENT" as const, // We've deprecated this but we need to keep it for backwards compatibility + organizationId: + logProperties.organizationId ?? resourceProperties.organizationId ?? "unknown", + projectId: logProperties.projectId ?? resourceProperties.projectId ?? "unknown", + runId: logProperties.runId ?? resourceProperties.runId ?? "unknown", + taskSlug: logProperties.taskSlug ?? resourceProperties.taskSlug ?? "unknown", + machineId: logProperties.machineId ?? resourceProperties.machineId, + attemptNumber: + extractNumberAttribute( + log.attributes ?? [], + [SemanticInternalAttributes.METADATA, SemanticInternalAttributes.ATTEMPT_NUMBER].join( + "." + ) + ) ?? resourceProperties.attemptNumber, + }; + }) + .filter(Boolean); + }); + + return { events, taskEventStore }; +} + +export function convertSpansToCreateableEvents( + resourceSpan: ResourceSpans, + spanAttributeValueLengthLimit: number, + defaultEventStore: string +): { events: Array; taskEventStore: string } { + const resourceAttributes = resourceSpan.resource?.attributes ?? []; + + const resourceProperties = extractEventProperties(resourceAttributes); + + const userDefinedResourceAttributes = truncateAttributes( + convertKeyValueItemsToMap(resourceAttributes ?? [], [], undefined, [ + SemanticInternalAttributes.USAGE, + SemanticInternalAttributes.SPAN, + SemanticInternalAttributes.METADATA, + SemanticInternalAttributes.STYLE, + SemanticInternalAttributes.METRIC_EVENTS, + SemanticInternalAttributes.TRIGGER, + "process", + "sdk", + "service", + "ctx", + "cli", + "cloud", + ]), + spanAttributeValueLengthLimit + ); + + const taskEventStore = + extractStringAttribute(resourceAttributes, [SemanticInternalAttributes.TASK_EVENT_STORE]) ?? + defaultEventStore; + + const events = resourceSpan.scopeSpans.flatMap((scopeSpan) => { + return scopeSpan.spans + .map((span) => { + const isPartial = isPartialSpan(span); + + if (!span.traceId || !span.spanId) { + return; + } + + const spanProperties = extractEventProperties( + span.attributes ?? [], + SemanticInternalAttributes.METADATA + ); + + const runTags = extractArrayAttribute( + span.attributes ?? [], + SemanticInternalAttributes.RUN_TAGS + ); + + const properties = + truncateAttributes( + convertKeyValueItemsToMap(span.attributes ?? [], [], undefined, [ + SemanticInternalAttributes.USAGE, + SemanticInternalAttributes.SPAN, + SemanticInternalAttributes.METADATA, + SemanticInternalAttributes.STYLE, + SemanticInternalAttributes.METRIC_EVENTS, + SemanticInternalAttributes.TRIGGER, + ]), + spanAttributeValueLengthLimit + ) ?? {}; + + return { + traceId: binaryToHex(span.traceId), + spanId: isPartial + ? extractStringAttribute( + span?.attributes ?? [], + SemanticInternalAttributes.SPAN_ID, + binaryToHex(span.spanId) + ) + : binaryToHex(span.spanId), + parentId: binaryToHex(span.parentSpanId), + message: span.name, + isPartial, + isError: span.status?.code === Status_StatusCode.ERROR, + kind: spanKindToEventKind(span.kind), + level: "TRACE" as const, + status: spanStatusToEventStatus(span.status), + startTime: span.startTimeUnixNano, + events: spanEventsToEventEvents(span.events ?? []), + duration: span.endTimeUnixNano - span.startTimeUnixNano, + properties, + resourceProperties: userDefinedResourceAttributes, + style: convertKeyValueItemsToMap( + pickAttributes(span.attributes ?? [], SemanticInternalAttributes.STYLE), + [] + ), + metadata: spanProperties.metadata ?? resourceProperties.metadata ?? {}, + environmentId: + spanProperties.environmentId ?? resourceProperties.environmentId ?? "unknown", + environmentType: "DEVELOPMENT" as const, + organizationId: + spanProperties.organizationId ?? resourceProperties.organizationId ?? "unknown", + projectId: spanProperties.projectId ?? resourceProperties.projectId ?? "unknown", + runId: spanProperties.runId ?? resourceProperties.runId ?? "unknown", + taskSlug: spanProperties.taskSlug ?? resourceProperties.taskSlug ?? "unknown", + machineId: spanProperties.machineId ?? resourceProperties.machineId, + runTags, + attemptNumber: + extractNumberAttribute( + span.attributes ?? [], + [SemanticInternalAttributes.METADATA, SemanticInternalAttributes.ATTEMPT_NUMBER].join( + "." + ) + ) ?? resourceProperties.attemptNumber, + }; + }) + .filter(Boolean); + }); + + return { events, taskEventStore }; +} + +function floorToTenSecondBucket(timeUnixNano: bigint | number): string { + const epochMs = Number(BigInt(timeUnixNano) / BigInt(1_000_000)); + const flooredMs = Math.floor(epochMs / 10_000) * 10_000; + const date = new Date(flooredMs); + // Format as ClickHouse DateTime: YYYY-MM-DD HH:MM:SS + return date + .toISOString() + .replace("T", " ") + .replace(/\.\d{3}Z$/, ""); +} + +export function convertMetricsToClickhouseRows( + resourceMetrics: ResourceMetrics, + spanAttributeValueLengthLimit: number +): MetricsV1Input[] { + const resourceAttributes = resourceMetrics.resource?.attributes ?? []; + const resourceProperties = extractEventProperties(resourceAttributes); + + const organizationId = resourceProperties.organizationId ?? "unknown"; + const projectId = resourceProperties.projectId ?? "unknown"; + const environmentId = resourceProperties.environmentId ?? "unknown"; + const resourceCtx = { + taskSlug: resourceProperties.taskSlug, + runId: resourceProperties.runId, + attemptNumber: resourceProperties.attemptNumber, + machineId: extractStringAttribute(resourceAttributes, SemanticInternalAttributes.MACHINE_ID), + workerId: extractStringAttribute(resourceAttributes, SemanticInternalAttributes.WORKER_ID), + workerVersion: extractStringAttribute( + resourceAttributes, + SemanticInternalAttributes.WORKER_VERSION + ), + }; + + const rows: MetricsV1Input[] = []; + + for (const scopeMetrics of resourceMetrics.scopeMetrics) { + for (const metric of scopeMetrics.metrics) { + const metricName = metric.name; + + // Process gauge data points + if (metric.gauge) { + for (const dp of metric.gauge.dataPoints) { + const value: number = + dp.asDouble !== undefined ? dp.asDouble : dp.asInt !== undefined ? Number(dp.asInt) : 0; + const resolved = resolveDataPointContext(dp.attributes ?? [], resourceCtx); + + rows.push({ + organization_id: organizationId, + project_id: projectId, + environment_id: environmentId, + metric_name: metricName, + metric_type: "gauge", + metric_subject: resolved.machineId ?? "unknown", + bucket_start: floorToTenSecondBucket(dp.timeUnixNano), + value, + attributes: resolved.attributes, + }); + } + } + + // Process sum data points + if (metric.sum) { + for (const dp of metric.sum.dataPoints) { + const value: number = + dp.asDouble !== undefined ? dp.asDouble : dp.asInt !== undefined ? Number(dp.asInt) : 0; + const resolved = resolveDataPointContext(dp.attributes ?? [], resourceCtx); + + rows.push({ + organization_id: organizationId, + project_id: projectId, + environment_id: environmentId, + metric_name: metricName, + metric_type: "sum", + metric_subject: resolved.machineId ?? "unknown", + bucket_start: floorToTenSecondBucket(dp.timeUnixNano), + value, + attributes: resolved.attributes, + }); + } + } + + // Process histogram data points + if (metric.histogram) { + for (const dp of metric.histogram.dataPoints) { + const resolved = resolveDataPointContext(dp.attributes ?? [], resourceCtx); + const count = Number(dp.count); + const sum = dp.sum ?? 0; + + rows.push({ + organization_id: organizationId, + project_id: projectId, + environment_id: environmentId, + metric_name: metricName, + metric_type: "histogram", + metric_subject: resolved.machineId ?? "unknown", + bucket_start: floorToTenSecondBucket(dp.timeUnixNano), + value: count > 0 ? sum / count : 0, + attributes: resolved.attributes, + }); + } + } + } + } + + return rows; +} + +// Prefixes injected by TaskContextMetricExporter — these are extracted into +// the nested `trigger` key and should not appear as top-level user attributes. +const INTERNAL_METRIC_ATTRIBUTE_PREFIXES = ["ctx.", "worker."]; + +interface ResourceContext { + taskSlug: string | undefined; + runId: string | undefined; + attemptNumber: number | undefined; + machineId: string | undefined; + workerId: string | undefined; + workerVersion: string | undefined; +} + +function resolveDataPointContext( + dpAttributes: KeyValue[], + resourceCtx: ResourceContext +): { + machineId: string | undefined; + attributes: Record; +} { + const runId = + resourceCtx.runId ?? extractStringAttribute(dpAttributes, SemanticInternalAttributes.RUN_ID); + const taskSlug = + resourceCtx.taskSlug ?? + extractStringAttribute(dpAttributes, SemanticInternalAttributes.TASK_SLUG); + const attemptNumber = + resourceCtx.attemptNumber ?? + extractNumberAttribute(dpAttributes, SemanticInternalAttributes.ATTEMPT_NUMBER); + const machineId = + resourceCtx.machineId ?? + extractStringAttribute(dpAttributes, SemanticInternalAttributes.MACHINE_ID); + const workerId = + resourceCtx.workerId ?? + extractStringAttribute(dpAttributes, SemanticInternalAttributes.WORKER_ID); + const workerVersion = + resourceCtx.workerVersion ?? + extractStringAttribute(dpAttributes, SemanticInternalAttributes.WORKER_VERSION); + const machineName = extractStringAttribute( + dpAttributes, + SemanticInternalAttributes.MACHINE_PRESET_NAME + ); + const environmentType = extractStringAttribute( + dpAttributes, + SemanticInternalAttributes.ENVIRONMENT_TYPE + ); + + // Build the trigger context object with only defined values + const trigger: Record = {}; + if (runId) trigger.run_id = runId; + if (taskSlug) trigger.task_slug = taskSlug; + if (attemptNumber !== undefined) trigger.attempt_number = attemptNumber; + if (machineId) trigger.machine_id = machineId; + if (machineName) trigger.machine_name = machineName; + if (workerId) trigger.worker_id = workerId; + if (workerVersion) trigger.worker_version = workerVersion; + if (environmentType) trigger.environment_type = environmentType; + + // Build user attributes, filtering out internal ctx/worker keys + const result: Record = {}; + + if (Object.keys(trigger).length > 0) { + result.trigger = trigger; + } + + for (const attr of dpAttributes) { + if (INTERNAL_METRIC_ATTRIBUTE_PREFIXES.some((prefix) => attr.key.startsWith(prefix))) { + continue; + } + + if (isStringValue(attr.value)) { + result[attr.key] = attr.value.stringValue; + } else if (isIntValue(attr.value)) { + result[attr.key] = Number(attr.value.intValue); + } else if (isDoubleValue(attr.value)) { + result[attr.key] = attr.value.doubleValue; + } else if (isBoolValue(attr.value)) { + result[attr.key] = attr.value.boolValue; + } + } + + return { machineId, attributes: result }; +} + +function extractEventProperties(attributes: KeyValue[], prefix?: string) { + return { + metadata: convertSelectedKeyValueItemsToMap(attributes, [SemanticInternalAttributes.METADATA]), + environmentId: extractStringAttribute(attributes, [ + prefix, + SemanticInternalAttributes.ENVIRONMENT_ID, + ]), + organizationId: extractStringAttribute(attributes, [ + prefix, + SemanticInternalAttributes.ORGANIZATION_ID, + ]), + projectId: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.PROJECT_ID]), + runId: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.RUN_ID]), + attemptNumber: extractNumberAttribute(attributes, [ + prefix, + SemanticInternalAttributes.ATTEMPT_NUMBER, + ]), + taskSlug: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.TASK_SLUG]), + machineId: extractStringAttribute(attributes, [prefix, SemanticInternalAttributes.MACHINE_ID]), + }; +} + +function pickAttributes(attributes: KeyValue[], prefix: string): KeyValue[] { + return attributes + .filter((attribute) => attribute.key.startsWith(prefix)) + .map((attribute) => { + return { + key: attribute.key.replace(`${prefix}.`, ""), + value: attribute.value, + }; + }); +} + +function convertKeyValueItemsToMap( + attributes: KeyValue[], + filteredKeys: string[] = [], + prefix?: string, + filteredPrefixes: string[] = [] +): Record | undefined { + if (!attributes) return; + if (!attributes.length) return; + + let filteredAttributes = attributes.filter((attribute) => !filteredKeys.includes(attribute.key)); + + if (!filteredAttributes.length) return; + + if (filteredPrefixes.length) { + filteredAttributes = filteredAttributes.filter( + (attribute) => !filteredPrefixes.some((prefix) => attribute.key.startsWith(prefix)) + ); + } + + if (!filteredAttributes.length) return; + + const result = filteredAttributes.reduce( + (map: Record, attribute) => { + map[`${prefix ? `${prefix}.` : ""}${attribute.key}`] = isStringValue(attribute.value) + ? attribute.value.stringValue + : isIntValue(attribute.value) + ? Number(attribute.value.intValue) + : isDoubleValue(attribute.value) + ? attribute.value.doubleValue + : isBoolValue(attribute.value) + ? attribute.value.boolValue + : isBytesValue(attribute.value) + ? binaryToHex(attribute.value.bytesValue) + : isArrayValue(attribute.value) + ? serializeArrayValue(attribute.value.arrayValue!.values) + : undefined; + + return map; + }, + {} + ); + + return result; +} + +function convertSelectedKeyValueItemsToMap( + attributes: KeyValue[], + selectedPrefixes: string[] = [], + prefix?: string +): Record | undefined { + if (!attributes) return; + if (!attributes.length) return; + + let selectedAttributes = attributes.filter((attribute) => + selectedPrefixes.some((prefix) => attribute.key.startsWith(prefix)) + ); + + if (!selectedAttributes.length) return; + + const result = selectedAttributes.reduce( + (map: Record, attribute) => { + map[`${prefix ? `${prefix}.` : ""}${attribute.key}`] = isStringValue(attribute.value) + ? attribute.value.stringValue + : isIntValue(attribute.value) + ? Number(attribute.value.intValue) + : isDoubleValue(attribute.value) + ? attribute.value.doubleValue + : isBoolValue(attribute.value) + ? attribute.value.boolValue + : isBytesValue(attribute.value) + ? binaryToHex(attribute.value.bytesValue) + : isArrayValue(attribute.value) + ? serializeArrayValue(attribute.value.arrayValue!.values) + : undefined; + + return map; + }, + {} + ); + + return result; +} +function spanEventsToEventEvents(events: Span_Event[]): CreateEventInput["events"] { + return events.map((event) => { + return { + name: event.name, + time: convertUnixNanoToDate(event.timeUnixNano), + properties: convertKeyValueItemsToMap(event.attributes ?? []), + }; + }); +} + +function spanStatusToEventStatus(status: Span["status"]): CreatableEventStatus { + if (!status) return "UNSET"; + + switch (status.code) { + case Status_StatusCode.OK: { + return "OK"; + } + case Status_StatusCode.ERROR: { + return "ERROR"; + } + case Status_StatusCode.UNSET: { + return "UNSET"; + } + default: { + return "UNSET"; + } + } +} + +function spanKindToEventKind(kind: Span["kind"]): CreatableEventKind { + switch (kind) { + case Span_SpanKind.CLIENT: { + return "CLIENT"; + } + case Span_SpanKind.SERVER: { + return "SERVER"; + } + case Span_SpanKind.CONSUMER: { + return "CONSUMER"; + } + case Span_SpanKind.PRODUCER: { + return "PRODUCER"; + } + default: { + return "INTERNAL"; + } + } +} + +function logLevelToEventLevel(level: SeverityNumber): CreateEventInput["level"] { + switch (level) { + case SeverityNumber.TRACE: + case SeverityNumber.TRACE2: + case SeverityNumber.TRACE3: + case SeverityNumber.TRACE4: { + return "TRACE"; + } + case SeverityNumber.DEBUG: + case SeverityNumber.DEBUG2: + case SeverityNumber.DEBUG3: + case SeverityNumber.DEBUG4: { + return "DEBUG"; + } + case SeverityNumber.INFO: + case SeverityNumber.INFO2: + case SeverityNumber.INFO3: + case SeverityNumber.INFO4: { + return "INFO"; + } + case SeverityNumber.WARN: + case SeverityNumber.WARN2: + case SeverityNumber.WARN3: + case SeverityNumber.WARN4: { + return "WARN"; + } + case SeverityNumber.ERROR: + case SeverityNumber.ERROR2: + case SeverityNumber.ERROR3: + case SeverityNumber.ERROR4: { + return "ERROR"; + } + case SeverityNumber.FATAL: + case SeverityNumber.FATAL2: + case SeverityNumber.FATAL3: + case SeverityNumber.FATAL4: { + return "ERROR"; + } + default: { + return "INFO"; + } + } +} + +function logLevelToEventStatus(level: SeverityNumber): CreatableEventStatus { + switch (level) { + case SeverityNumber.TRACE: + case SeverityNumber.TRACE2: + case SeverityNumber.TRACE3: + case SeverityNumber.TRACE4: { + return "OK"; + } + case SeverityNumber.DEBUG: + case SeverityNumber.DEBUG2: + case SeverityNumber.DEBUG3: + case SeverityNumber.DEBUG4: { + return "OK"; + } + case SeverityNumber.INFO: + case SeverityNumber.INFO2: + case SeverityNumber.INFO3: + case SeverityNumber.INFO4: { + return "OK"; + } + case SeverityNumber.WARN: + case SeverityNumber.WARN2: + case SeverityNumber.WARN3: + case SeverityNumber.WARN4: { + return "OK"; + } + case SeverityNumber.ERROR: + case SeverityNumber.ERROR2: + case SeverityNumber.ERROR3: + case SeverityNumber.ERROR4: { + return "ERROR"; + } + case SeverityNumber.FATAL: + case SeverityNumber.FATAL2: + case SeverityNumber.FATAL3: + case SeverityNumber.FATAL4: { + return "ERROR"; + } + default: { + return "OK"; + } + } +} + +function convertUnixNanoToDate(unixNano: bigint | number): Date { + return new Date(Number(BigInt(unixNano) / BigInt(1_000_000))); +} + +function extractStringAttribute( + attributes: KeyValue[], + name: string | Array +): string | undefined; +function extractStringAttribute( + attributes: KeyValue[], + name: string | Array, + fallback: string +): string; +function extractStringAttribute( + attributes: KeyValue[], + name: string | Array, + fallback?: string +): string | undefined { + const key = Array.isArray(name) ? name.filter(Boolean).join(".") : name; + + const attribute = attributes.find((attribute) => attribute.key === key); + + if (!attribute) return fallback; + + return isStringValue(attribute?.value) ? attribute.value.stringValue : fallback; +} + +function extractNumberAttribute( + attributes: KeyValue[], + name: string | Array +): number | undefined; +function extractNumberAttribute( + attributes: KeyValue[], + name: string | Array, + fallback: number +): number; +function extractNumberAttribute( + attributes: KeyValue[], + name: string | Array, + fallback?: number +): number | undefined { + const key = Array.isArray(name) ? name.filter(Boolean).join(".") : name; + + const attribute = attributes.find((attribute) => attribute.key === key); + + if (!attribute) return fallback; + + return isIntValue(attribute?.value) ? Number(attribute.value.intValue) : fallback; +} + +function extractArrayAttribute( + attributes: KeyValue[], + name: string | Array +): string[] | undefined { + const key = Array.isArray(name) ? name.filter(Boolean).join(".") : name; + + const attribute = attributes.find((attribute) => attribute.key === key); + + if (!attribute?.value?.arrayValue?.values) return undefined; + + return attribute.value.arrayValue.values + .filter((v): v is { stringValue: string } => isStringValue(v)) + .map((v) => v.stringValue); +} + +function isPartialSpan(span: Span): boolean { + if (!span.attributes) return false; + + const attribute = span.attributes.find( + (attribute) => attribute.key === SemanticInternalAttributes.SPAN_PARTIAL + ); + + if (!attribute) return false; + + return isBoolValue(attribute.value) ? attribute.value.boolValue : false; +} + +export function isBoolValue(value: AnyValue | undefined): value is { boolValue: boolean } { + if (!value) return false; + + return typeof value.boolValue === "boolean"; +} + +export function isStringValue(value: AnyValue | undefined): value is { stringValue: string } { + if (!value) return false; + + return typeof value.stringValue === "string"; +} + +function isIntValue(value: AnyValue | undefined): value is { intValue: bigint } { + if (!value) return false; + + return typeof value.intValue === "number" || typeof value.intValue === "bigint"; +} + +function isDoubleValue(value: AnyValue | undefined): value is { doubleValue: number } { + if (!value) return false; + + return typeof value.doubleValue === "number"; +} + +function isBytesValue(value: AnyValue | undefined): value is { bytesValue: Buffer } { + if (!value) return false; + + return Buffer.isBuffer(value.bytesValue); +} + +function isArrayValue( + value: AnyValue | undefined +): value is { arrayValue: { values: AnyValue[] } } { + if (!value) return false; + + return value.arrayValue != null && Array.isArray(value.arrayValue.values); +} + +/** + * Serialize an OTEL array value into a JSON string. + * For arrays of strings, produces a JSON array: `["item1","item2"]` + * For mixed types, extracts primitives and serializes. + */ +function serializeArrayValue(values: AnyValue[]): string { + const items = values.map((v) => { + if (isStringValue(v)) return v.stringValue; + if (isIntValue(v)) return Number(v.intValue); + if (isDoubleValue(v)) return v.doubleValue; + if (isBoolValue(v)) return v.boolValue; + return null; + }); + + return JSON.stringify(items); +} + +function binaryToHex(buffer: Buffer | string): string; +function binaryToHex(buffer: Buffer | string | undefined): string | undefined; +function binaryToHex(buffer: Buffer | string | undefined): string | undefined { + if (!buffer) return undefined; + if (typeof buffer === "string") return buffer; + + return Buffer.from(Array.from(buffer)).toString("hex"); +} + +function truncateAttributes( + attributes: Record | undefined, + maximumLength: number = 1024 +): Record | undefined { + if (!attributes) return undefined; + + const truncatedAttributes: Record = {}; + + for (const [key, value] of Object.entries(attributes)) { + if (!key) continue; + + if (typeof value === "string") { + truncatedAttributes[key] = truncateAndDetectUnpairedSurrogate(value, maximumLength); + } else { + truncatedAttributes[key] = value; + } + } + + return truncatedAttributes; +} + +function truncateAndDetectUnpairedSurrogate(str: string, maximumLength: number): string { + const truncatedString = smartTruncateString(str, maximumLength); + + if (hasUnpairedSurrogateAtEnd(truncatedString)) { + return smartTruncateString(truncatedString, [...truncatedString].length - 1); + } + + return truncatedString; +} + +const ASCII_ONLY_REGEX = /^[\p{ASCII}]*$/u; + +function smartTruncateString(str: string, maximumLength: number): string { + if (!str) return ""; + if (str.length <= maximumLength) return str; + + const checkLength = Math.min(str.length, maximumLength * 2 + 2); + + if (ASCII_ONLY_REGEX.test(str.slice(0, checkLength))) { + return str.slice(0, maximumLength); + } + + return [...str.slice(0, checkLength)].slice(0, maximumLength).join(""); +} + +function hasUnpairedSurrogateAtEnd(str: string): boolean { + if (str.length === 0) return false; + + const lastCode = str.charCodeAt(str.length - 1); + + // Check if last character is an unpaired high surrogate + if (lastCode >= 0xd800 && lastCode <= 0xdbff) { + return true; // High surrogate at end = unpaired + } + + // Check if last character is an unpaired low surrogate + if (lastCode >= 0xdc00 && lastCode <= 0xdfff) { + // Low surrogate is only valid if preceded by high surrogate + if (str.length === 1) return true; // Single low surrogate + + const secondLastCode = str.charCodeAt(str.length - 2); + if (secondLastCode < 0xd800 || secondLastCode > 0xdbff) { + return true; // Low surrogate not preceded by high surrogate + } + } + + return false; +} From 507bdf55011dd2342b19f31ae44e2cc414c1ef3a Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Fri, 10 Jul 2026 22:52:31 +0100 Subject: [PATCH 02/10] 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. --- apps/webapp/app/routes/otel.v1.logs.ts | 11 +- apps/webapp/app/routes/otel.v1.metrics.ts | 11 +- apps/webapp/app/routes/otel.v1.traces.ts | 11 +- apps/webapp/app/v3/otlpExporter.server.ts | 63 ++++++- apps/webapp/app/v3/otlpTransform.server.ts | 42 +++++ apps/webapp/app/v3/otlpTransformWorker.ts | 102 ++++++++++++ apps/webapp/app/v3/otlpWorkerPool.server.ts | 157 ++++++++++++++++++ apps/webapp/package.json | 1 + .../llm-model-catalog/src/registry.test.ts | 34 ++++ .../llm-model-catalog/src/registry.ts | 32 +++- 10 files changed, 456 insertions(+), 8 deletions(-) create mode 100644 apps/webapp/app/v3/otlpTransformWorker.ts create mode 100644 apps/webapp/app/v3/otlpWorkerPool.server.ts diff --git a/apps/webapp/app/routes/otel.v1.logs.ts b/apps/webapp/app/routes/otel.v1.logs.ts index 6e75de0a5c..8b177f524c 100644 --- a/apps/webapp/app/routes/otel.v1.logs.ts +++ b/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); diff --git a/apps/webapp/app/routes/otel.v1.metrics.ts b/apps/webapp/app/routes/otel.v1.metrics.ts index 78c067348f..8b39649562 100644 --- a/apps/webapp/app/routes/otel.v1.metrics.ts +++ b/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); diff --git a/apps/webapp/app/routes/otel.v1.traces.ts b/apps/webapp/app/routes/otel.v1.traces.ts index daf106f7f3..313077ac05 100644 --- a/apps/webapp/app/routes/otel.v1.traces.ts +++ b/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); diff --git a/apps/webapp/app/v3/otlpExporter.server.ts b/apps/webapp/app/v3/otlpExporter.server.ts index 4700ab8a3b..9f42a697d8 100644 --- a/apps/webapp/app/v3/otlpExporter.server.ts +++ b/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 { + await this.#exportRawEvents("traces", payload); + } + + async exportLogsRaw(payload: Uint8Array): Promise { + await this.#exportRawEvents("logs", payload); + } + + async exportMetricsRaw(payload: Uint8Array): Promise { + 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 { + 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; taskEventStore: string }[] + eventsWithStores: { events: Array; taskEventStore: string }[], + alreadyEnriched = false ) { - await waitForLlmPricingReady(); + if (!alreadyEnriched) { + await waitForLlmPricingReady(); + } // Group by unique event repositories const routeCache = new Map(); @@ -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}`); diff --git a/apps/webapp/app/v3/otlpTransform.server.ts b/apps/webapp/app/v3/otlpTransform.server.ts index 1e995253e3..d66f1a8b82 100644 --- a/apps/webapp/app/v3/otlpTransform.server.ts +++ b/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, diff --git a/apps/webapp/app/v3/otlpTransformWorker.ts b/apps/webapp/app/v3/otlpTransformWorker.ts new file mode 100644 index 0000000000..c158e10ce8 --- /dev/null +++ b/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), + }); + } +}); diff --git a/apps/webapp/app/v3/otlpWorkerPool.server.ts b/apps/webapp/app/v3/otlpWorkerPool.server.ts new file mode 100644 index 0000000000..80eaf6be25 --- /dev/null +++ b/apps/webapp/app/v3/otlpWorkerPool.server.ts @@ -0,0 +1,157 @@ +import { Worker } from "node:worker_threads"; +import path from "node:path"; +import { logger } from "~/services/logger.server"; + +export type TransformKind = "traces" | "logs" | "metrics"; + +type Pending = { resolve: (r: any) => void; reject: (e: Error) => void }; + +type QueueItem = { + message: { + id: number; + kind: TransformKind; + payload: Uint8Array; + spanAttributeValueLengthLimit: number; + defaultEventStore: string; + }; + transfer: ArrayBuffer[]; + pending: Pending; +}; + +// Hand-rolled worker_threads pool: one in-flight task per worker so CPU-bound transforms run +// fully in parallel. The main thread stays the only DB reader and broadcasts pricing to workers. +export class OtlpWorkerPool { + private readonly workers: Worker[] = []; + private readonly idle: Worker[] = []; + private readonly queue: QueueItem[] = []; + private readonly pending = new Map(); + private readonly busyByWorker = new Map(); + private nextId = 1; + private latestPricingModels: unknown[]; + + constructor( + private readonly size: number, + private readonly workerPath: string, + pricingModels: unknown[] + ) { + this.latestPricingModels = pricingModels; + for (let i = 0; i < size; i++) { + this.spawn(); + } + logger.info("OtlpWorkerPool started", { size, workerPath }); + } + + private spawn() { + const worker = new Worker(this.workerPath, { + workerData: { pricingModels: this.latestPricingModels }, + }); + + worker.on("message", (msg: { id: number; ok: boolean; result?: any; error?: string }) => { + const pending = this.pending.get(msg.id); + this.pending.delete(msg.id); + this.busyByWorker.delete(worker); + if (pending) { + if (msg.ok) pending.resolve(msg.result); + else pending.reject(new Error(msg.error ?? "otlp worker error")); + } + this.release(worker); + }); + + worker.on("error", (error) => { + logger.error("OtlpWorkerPool worker error", { error: error.message }); + this.reap(worker, error); + }); + + worker.on("exit", (code) => { + if (code !== 0) { + this.reap(worker, new Error(`otlp worker exited with code ${code}`)); + } + }); + + this.workers.push(worker); + this.idle.push(worker); + } + + // On crash: fail its in-flight task, drop it, and spawn a replacement. + private reap(worker: Worker, error: Error) { + const inFlightId = this.busyByWorker.get(worker); + if (inFlightId !== undefined) { + const pending = this.pending.get(inFlightId); + this.pending.delete(inFlightId); + this.busyByWorker.delete(worker); + pending?.reject(error); + } + const wi = this.workers.indexOf(worker); + if (wi !== -1) this.workers.splice(wi, 1); + const ii = this.idle.indexOf(worker); + if (ii !== -1) this.idle.splice(ii, 1); + void worker.terminate().catch(() => {}); + if (this.workers.length < this.size) this.spawn(); + this.drain(); + } + + private release(worker: Worker) { + this.idle.push(worker); + this.drain(); + } + + private drain() { + while (this.queue.length > 0 && this.idle.length > 0) { + const worker = this.idle.pop()!; + const item = this.queue.shift()!; + this.pending.set(item.message.id, item.pending); + this.busyByWorker.set(worker, item.message.id); + worker.postMessage(item.message, item.transfer); + } + } + + runTransform( + kind: TransformKind, + payload: Uint8Array, + config: { spanAttributeValueLengthLimit: number; defaultEventStore: string } + ): Promise { + const id = this.nextId++; + return new Promise((resolve, reject) => { + this.queue.push({ + message: { + id, + kind, + payload, + spanAttributeValueLengthLimit: config.spanAttributeValueLengthLimit, + defaultEventStore: config.defaultEventStore, + }, + // Zero-copy the payload into the worker; the request owns a fresh ArrayBuffer. + transfer: [payload.buffer as ArrayBuffer], + pending: { resolve, reject }, + }); + this.drain(); + }); + } + + broadcastPricing(models: unknown[]) { + this.latestPricingModels = models; + for (const worker of this.workers) { + worker.postMessage({ type: "pricing", models }); + } + } + + get queueDepth() { + return this.queue.length; + } +} + +let pool: OtlpWorkerPool | undefined; + +export function getOtlpWorkerPool(size: number, pricingModels: unknown[]): OtlpWorkerPool { + if (!pool) { + const workerPath = + process.env.OTEL_TRANSFORM_WORKER_PATH ?? + path.join(process.cwd(), "build", "otlpTransformWorker.cjs"); + pool = new OtlpWorkerPool(size, workerPath, pricingModels); + } + return pool; +} + +export function broadcastPricingToPool(models: unknown[]) { + pool?.broadcastPricing(models); +} diff --git a/apps/webapp/package.json b/apps/webapp/package.json index db75c65b27..5392a418f1 100644 --- a/apps/webapp/package.json +++ b/apps/webapp/package.json @@ -7,6 +7,7 @@ "build": "run-s build:** && pnpm run upload:sourcemaps", "build:remix": "remix build --sourcemap", "build:server": "esbuild --platform=node --format=cjs ./server.ts --outdir=build --sourcemap", + "build:otlpworker": "esbuild --platform=node --format=cjs --bundle ./app/v3/otlpTransformWorker.ts --outfile=build/otlpTransformWorker.cjs --sourcemap", "build:sentry": "esbuild --platform=node --format=cjs --outbase=. ./sentry.server.ts ./app/utils/sentryTraceContext.server.ts --outdir=build --sourcemap", "dev": "cross-env PORT=3030 remix dev -c \"node ./build/server.js\"", "dev:worker": "cross-env NODE_PATH=../../node_modules/.pnpm/node_modules node ./build/server.js", diff --git a/internal-packages/llm-model-catalog/src/registry.test.ts b/internal-packages/llm-model-catalog/src/registry.test.ts index cfda717727..37c5800e63 100644 --- a/internal-packages/llm-model-catalog/src/registry.test.ts +++ b/internal-packages/llm-model-catalog/src/registry.test.ts @@ -485,3 +485,37 @@ describe("ModelPricingRegistry", () => { }); }); }); + +describe("loadFromModels / toSerializable (worker in-memory load)", () => { + it("matches and prices from in-memory models without a DB", () => { + const reg = new ModelPricingRegistry(); + reg.loadFromModels([gpt4o, claudeSonnet]); + + expect(reg.isLoaded).toBe(true); + expect(reg.match("gpt-4o")).not.toBeNull(); + expect(reg.match("gpt-4o-2024-08-06")).not.toBeNull(); + + const cost = reg.calculateCost("gpt-4o", { input: 1000, output: 500 }); + expect(cost).not.toBeNull(); + expect(cost!.totalCost).toBeGreaterThan(0); + }); + + it("round-trips through toSerializable (the main->worker broadcast shape)", () => { + const source = new ModelPricingRegistry(); + source.loadFromModels([gpt4o, claudeSonnet]); + + const worker = new ModelPricingRegistry(); + worker.loadFromModels(source.toSerializable()); + + for (const m of ["gpt-4o", "gpt-4o-2024-08-06", "claude-sonnet-4-0", "unknown-model"]) { + expect(worker.match(m)).toEqual(source.match(m)); + } + const usage = { input: 1234, output: 567 }; + expect(worker.calculateCost("gpt-4o", usage)).toEqual(source.calculateCost("gpt-4o", usage)); + }); + + it("throws if loadFromDatabase is called without a prisma client", async () => { + const reg = new ModelPricingRegistry(); + await expect(reg.loadFromDatabase()).rejects.toThrow(/requires a prisma client/); + }); +}); diff --git a/internal-packages/llm-model-catalog/src/registry.ts b/internal-packages/llm-model-catalog/src/registry.ts index 880bed3951..3164efdc5e 100644 --- a/internal-packages/llm-model-catalog/src/registry.ts +++ b/internal-packages/llm-model-catalog/src/registry.ts @@ -20,7 +20,7 @@ function compilePattern(pattern: string): RegExp { } export class ModelPricingRegistry { - private _prisma: PrismaClient | PrismaReplicaClient; + private _prisma?: PrismaClient | PrismaReplicaClient; private _patterns: CompiledPattern[] = []; // TODO: When we add project-based models (users adding their own), this cache grows unbounded // between reloads. Fine-tuned model IDs (e.g. "ft:gpt-3.5-turbo:org:name:id") create unique @@ -32,7 +32,7 @@ export class ModelPricingRegistry { /** Resolves once the initial `loadFromDatabase()` completes successfully. */ readonly isReady: Promise; - constructor(prisma: PrismaClient | PrismaReplicaClient) { + constructor(prisma?: PrismaClient | PrismaReplicaClient) { this._prisma = prisma; this.isReady = new Promise((resolve) => { this._readyResolve = resolve; @@ -44,6 +44,10 @@ export class ModelPricingRegistry { } async loadFromDatabase(): Promise { + if (!this._prisma) { + throw new Error("loadFromDatabase requires a prisma client; use loadFromModels in a worker"); + } + const models = await this._prisma.llmModel.findMany({ where: { projectId: null }, include: { @@ -89,6 +93,30 @@ export class ModelPricingRegistry { } } + this.applyPatterns(compiled); + } + + // Build the registry from already-loaded model rows (e.g. broadcast to a worker), no DB. + loadFromModels(models: LlmModelWithPricing[]): void { + const compiled: CompiledPattern[] = []; + + for (const model of models) { + try { + compiled.push({ regex: compilePattern(model.matchPattern), model }); + } catch { + console.warn(`Invalid regex pattern for model ${model.modelName}: ${model.matchPattern}`); + } + } + + this.applyPatterns(compiled); + } + + // Serializable snapshot of the loaded models (safe to postMessage to a worker). + toSerializable(): LlmModelWithPricing[] { + return this._patterns.map((p) => p.model); + } + + private applyPatterns(compiled: CompiledPattern[]): void { this._patterns = compiled; this._exactMatchCache.clear(); From f94c11418d2bf1c0918cbfba738105aeda44ac39 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Fri, 10 Jul 2026 23:02:21 +0100 Subject: [PATCH 03/10] feat(webapp): re-broadcast LLM pricing to OTLP workers on registry reload The pricing registry now notifies subscribers after every load/reload (interval + pub/sub); the OTLP worker pool subscribes and re-broadcasts the compiled model rows to all workers, so their in-memory pricing stays fresh instead of being a static snapshot from pool creation. Respawned workers also inherit the latest snapshot via workerData. Verified on the isolated stack: broadcast fires on init then every reload interval to all workers; registry-sourced cost enrichment kept landing in ClickHouse across multiple reload cycles under load (no break). --- .../app/v3/llmPricingRegistry.server.ts | 60 +++++++++++++++---- apps/webapp/app/v3/otlpExporter.server.ts | 11 +++- apps/webapp/app/v3/otlpWorkerPool.server.ts | 4 ++ 3 files changed, 63 insertions(+), 12 deletions(-) diff --git a/apps/webapp/app/v3/llmPricingRegistry.server.ts b/apps/webapp/app/v3/llmPricingRegistry.server.ts index eb186e1521..20ee116061 100644 --- a/apps/webapp/app/v3/llmPricingRegistry.server.ts +++ b/apps/webapp/app/v3/llmPricingRegistry.server.ts @@ -1,4 +1,5 @@ import { ModelPricingRegistry, seedLlmPricing } from "@internal/llm-model-catalog"; +import type { LlmModelWithPricing } from "@internal/llm-model-catalog"; import { prisma, $replica } from "~/db.server"; import { env } from "~/env.server"; import { logger } from "~/services/logger.server"; @@ -7,6 +8,37 @@ import { createRedisClient } from "~/redis.server"; import { singleton } from "~/utils/singleton"; import { setLlmPricingRegistry } from "./utils/enrichCreatableEvents.server"; +type PricingReloadListener = (models: LlmModelWithPricing[]) => void; +const pricingReloadListeners = new Set(); + +// Notify subscribers (e.g. the OTLP worker pool) after each load/reload so they can rebuild +// their own in-memory copy. No-op until something subscribes. +function emitPricingReload() { + if (!llmPricingRegistry || !llmPricingRegistry.isLoaded || pricingReloadListeners.size === 0) { + return; + } + const models = llmPricingRegistry.toSerializable(); + for (const listener of pricingReloadListeners) { + try { + listener(models); + } catch (err) { + logger.warn("LLM pricing reload listener failed", { + error: err instanceof Error ? err.message : String(err), + }); + } + } +} + +export function subscribeToPricingReload(listener: PricingReloadListener): () => void { + pricingReloadListeners.add(listener); + if (llmPricingRegistry?.isLoaded) { + listener(llmPricingRegistry.toSerializable()); + } + return () => { + pricingReloadListeners.delete(listener); + }; +} + async function initRegistry(registry: ModelPricingRegistry) { if (env.LLM_PRICING_SEED_ON_STARTUP) { await seedLlmPricing(prisma); @@ -25,16 +57,21 @@ export const llmPricingRegistry = singleton("llmPricingRegistry", () => { // Wire up the registry so enrichCreatableEvents can use it setLlmPricingRegistry(registry); - initRegistry(registry).catch((err) => { - console.error("Failed to initialize LLM pricing registry", err); - }); + initRegistry(registry) + .then(() => emitPricingReload()) + .catch((err) => { + console.error("Failed to initialize LLM pricing registry", err); + }); // Periodic reload (backstop for the pub/sub path below) const reloadInterval = env.LLM_PRICING_RELOAD_INTERVAL_MS; const interval = setInterval(() => { - registry.reload().catch((err) => { - console.error("Failed to reload LLM pricing registry", err); - }); + registry + .reload() + .then(() => emitPricingReload()) + .catch((err) => { + console.error("Failed to reload LLM pricing registry", err); + }); }, reloadInterval); // Pub/sub reload is opt-in per process (default off). Without it, the @@ -73,11 +110,14 @@ export const llmPricingRegistry = singleton("llmPricingRegistry", () => { if (pendingReloadTimer) return; pendingReloadTimer = setTimeout(() => { pendingReloadTimer = null; - registry.reload().catch((err) => { - logger.warn("Failed to reload LLM pricing registry from pub/sub", { - error: err instanceof Error ? err.message : String(err), + registry + .reload() + .then(() => emitPricingReload()) + .catch((err) => { + logger.warn("Failed to reload LLM pricing registry from pub/sub", { + error: err instanceof Error ? err.message : String(err), + }); }); - }); }, debounceMs); } diff --git a/apps/webapp/app/v3/otlpExporter.server.ts b/apps/webapp/app/v3/otlpExporter.server.ts index 9f42a697d8..2ff8e4b029 100644 --- a/apps/webapp/app/v3/otlpExporter.server.ts +++ b/apps/webapp/app/v3/otlpExporter.server.ts @@ -32,7 +32,7 @@ import { } from "./otlpTransform.server"; import os from "node:os"; import { getOtlpWorkerPool } from "./otlpWorkerPool.server"; -import { llmPricingRegistry } from "./llmPricingRegistry.server"; +import { llmPricingRegistry, subscribeToPricingReload } 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. @@ -54,6 +54,7 @@ class OTLPExporter { private readonly _clickhouseFactory: ClickhouseFactory; private readonly _verbose: boolean; private readonly _spanAttributeValueLengthLimit: number; + #pricingSubscribed = false; constructor(config: OTLPExporterConfig) { this._tracer = trace.getTracer("otlp-exporter"); @@ -161,7 +162,13 @@ class OTLPExporter { await waitForLlmPricingReady(); const models = llmPricingRegistry && llmPricingRegistry.isLoaded ? llmPricingRegistry.toSerializable() : []; - return getOtlpWorkerPool(OTEL_TRANSFORM_WORKER_POOL_SIZE, models); + const pool = getOtlpWorkerPool(OTEL_TRANSFORM_WORKER_POOL_SIZE, models); + if (!this.#pricingSubscribed) { + this.#pricingSubscribed = true; + // Re-broadcast pricing to workers on every registry reload so their cost math stays fresh. + subscribeToPricingReload((updated) => pool.broadcastPricing(updated)); + } + return pool; } async #exportEvents( diff --git a/apps/webapp/app/v3/otlpWorkerPool.server.ts b/apps/webapp/app/v3/otlpWorkerPool.server.ts index 80eaf6be25..172131965b 100644 --- a/apps/webapp/app/v3/otlpWorkerPool.server.ts +++ b/apps/webapp/app/v3/otlpWorkerPool.server.ts @@ -133,6 +133,10 @@ export class OtlpWorkerPool { for (const worker of this.workers) { worker.postMessage({ type: "pricing", models }); } + logger.info("OtlpWorkerPool broadcast pricing", { + models: models.length, + workers: this.workers.length, + }); } get queueDepth() { From a20709f5f77d76037fdad6d7b98d2af0617a2863 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Fri, 10 Jul 2026 23:24:58 +0100 Subject: [PATCH 04/10] chore(webapp): add server-changes note for the OTLP transform worker pool --- .server-changes/otlp-transform-worker-pool.md | 6 ++++++ 1 file changed, 6 insertions(+) create mode 100644 .server-changes/otlp-transform-worker-pool.md diff --git a/.server-changes/otlp-transform-worker-pool.md b/.server-changes/otlp-transform-worker-pool.md new file mode 100644 index 0000000000..f5c23a40da --- /dev/null +++ b/.server-changes/otlp-transform-worker-pool.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: improvement +--- + +Optionally offload OTLP ingest processing (decode, transform, and LLM-cost enrichment) to a worker pool, freeing the main event loop under high telemetry volume. Off by default; opt in with the OTEL_TRANSFORM_WORKER_POOL_ENABLED environment variable. From 5c6287734217ddbe86c86156cdcd26c0230491af Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Sat, 11 Jul 2026 06:46:19 +0100 Subject: [PATCH 05/10] fix(webapp): harden the OTLP transform worker pool Address review feedback on the opt-in worker pool: - Validate the pool config through env.server and clamp the worker count to at least 1, so a zero or non-numeric OTEL_TRANSFORM_WORKER_POOL_SIZE can no longer leave an empty pool that hangs every request. - Add a per-task timeout (reject and reap a stuck worker) and a bounded queue (shed when full), so a wedged worker or overload can't hang requests forever or grow memory without limit. - Respawn crashed workers with exponential backoff instead of a tight loop. - Guard the immediate pricing-reload listener replay so a throwing listener can't escape subscription setup. --- .server-changes/otlp-transform-worker-pool.md | 2 +- apps/webapp/app/env.server.ts | 7 + .../app/v3/llmPricingRegistry.server.ts | 8 +- apps/webapp/app/v3/otlpExporter.server.ts | 45 +++++-- apps/webapp/app/v3/otlpWorkerPool.server.ts | 127 ++++++++++++------ 5 files changed, 134 insertions(+), 55 deletions(-) diff --git a/.server-changes/otlp-transform-worker-pool.md b/.server-changes/otlp-transform-worker-pool.md index f5c23a40da..a88d15020c 100644 --- a/.server-changes/otlp-transform-worker-pool.md +++ b/.server-changes/otlp-transform-worker-pool.md @@ -3,4 +3,4 @@ area: webapp type: improvement --- -Optionally offload OTLP ingest processing (decode, transform, and LLM-cost enrichment) to a worker pool, freeing the main event loop under high telemetry volume. Off by default; opt in with the OTEL_TRANSFORM_WORKER_POOL_ENABLED environment variable. +Optionally process high-volume telemetry ingestion in parallel for higher throughput under heavy load by setting `OTEL_TRANSFORM_WORKER_POOL_ENABLED=1`. Off by default. diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index e8bb102760..ca3eba44d5 100644 --- a/apps/webapp/app/env.server.ts +++ b/apps/webapp/app/env.server.ts @@ -1937,6 +1937,13 @@ const EnvironmentSchema = z EVENTS_CLICKHOUSE_MAX_TRACE_DETAILED_SUMMARY_VIEW_COUNT: z.coerce.number().int().default(5_000), EVENTS_CLICKHOUSE_MAX_LIVE_RELOADING_SETTING: z.coerce.number().int().default(2000), + // OTLP ingest transform worker pool (opt-in). When enabled, decode/convert/enrich run in a + // worker_threads pool instead of the request event loop; the single consolidated insert path + // is unchanged. + OTEL_TRANSFORM_WORKER_POOL_ENABLED: BoolEnv.default(false), + OTEL_TRANSFORM_WORKER_POOL_SIZE: z.coerce.number().int().optional(), + OTEL_TRANSFORM_WORKER_PATH: z.string().optional(), + // Organization data stores registry ORGANIZATION_DATA_STORES_RELOAD_INTERVAL_MS: z.coerce .number() diff --git a/apps/webapp/app/v3/llmPricingRegistry.server.ts b/apps/webapp/app/v3/llmPricingRegistry.server.ts index 20ee116061..31baf37059 100644 --- a/apps/webapp/app/v3/llmPricingRegistry.server.ts +++ b/apps/webapp/app/v3/llmPricingRegistry.server.ts @@ -32,7 +32,13 @@ function emitPricingReload() { export function subscribeToPricingReload(listener: PricingReloadListener): () => void { pricingReloadListeners.add(listener); if (llmPricingRegistry?.isLoaded) { - listener(llmPricingRegistry.toSerializable()); + try { + listener(llmPricingRegistry.toSerializable()); + } catch (err) { + logger.warn("LLM pricing reload listener failed", { + error: err instanceof Error ? err.message : String(err), + }); + } } return () => { pricingReloadListeners.delete(listener); diff --git a/apps/webapp/app/v3/otlpExporter.server.ts b/apps/webapp/app/v3/otlpExporter.server.ts index 2ff8e4b029..6b51315ed5 100644 --- a/apps/webapp/app/v3/otlpExporter.server.ts +++ b/apps/webapp/app/v3/otlpExporter.server.ts @@ -20,7 +20,6 @@ import { eventRepository } from "./eventRepository/eventRepository.server"; import type { CreateEventInput, IEventRepository } from "./eventRepository/eventRepository.types"; import { startSpan } from "./tracing.server"; import { enrichCreatableEvents } from "./utils/enrichCreatableEvents.server"; -import { waitForLlmPricingReady } from "./llmPricingRegistry.server"; import { env } from "~/env.server"; import { singleton } from "~/utils/singleton"; import { @@ -32,16 +31,22 @@ import { } from "./otlpTransform.server"; import os from "node:os"; import { getOtlpWorkerPool } from "./otlpWorkerPool.server"; -import { llmPricingRegistry, subscribeToPricingReload } from "./llmPricingRegistry.server"; +import { + llmPricingRegistry, + subscribeToPricingReload, + waitForLlmPricingReady, +} 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"; +export const otlpTransformWorkerPoolEnabled = env.OTEL_TRANSFORM_WORKER_POOL_ENABLED; -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); +// Always at least 1 worker: a 0/negative override must not silently disable the pool while the +// flag is on (that would hang every raw-export request on an empty pool). +const OTEL_TRANSFORM_WORKER_POOL_SIZE = Math.max( + 1, + env.OTEL_TRANSFORM_WORKER_POOL_SIZE ?? (os.cpus()?.length ?? 2) - 2 +); type OTLPExporterConfig = { clickhouseFactory: ClickhouseFactory; @@ -143,12 +148,20 @@ class OTLPExporter { } async #exportRawEvents(kind: "traces" | "logs", payload: Uint8Array): Promise { - 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); - }); + 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() { @@ -162,7 +175,11 @@ class OTLPExporter { await waitForLlmPricingReady(); const models = llmPricingRegistry && llmPricingRegistry.isLoaded ? llmPricingRegistry.toSerializable() : []; - const pool = getOtlpWorkerPool(OTEL_TRANSFORM_WORKER_POOL_SIZE, models); + const pool = getOtlpWorkerPool( + OTEL_TRANSFORM_WORKER_POOL_SIZE, + models, + env.OTEL_TRANSFORM_WORKER_PATH + ); if (!this.#pricingSubscribed) { this.#pricingSubscribed = true; // Re-broadcast pricing to workers on every registry reload so their cost math stays fresh. diff --git a/apps/webapp/app/v3/otlpWorkerPool.server.ts b/apps/webapp/app/v3/otlpWorkerPool.server.ts index 172131965b..db3b214aed 100644 --- a/apps/webapp/app/v3/otlpWorkerPool.server.ts +++ b/apps/webapp/app/v3/otlpWorkerPool.server.ts @@ -4,29 +4,38 @@ import { logger } from "~/services/logger.server"; export type TransformKind = "traces" | "logs" | "metrics"; -type Pending = { resolve: (r: any) => void; reject: (e: Error) => void }; - -type QueueItem = { - message: { - id: number; - kind: TransformKind; - payload: Uint8Array; - spanAttributeValueLengthLimit: number; - defaultEventStore: string; - }; +type TaskMessage = { + id: number; + kind: TransformKind; + payload: Uint8Array; + spanAttributeValueLengthLimit: number; + defaultEventStore: string; +}; + +type Task = { + message: TaskMessage; transfer: ArrayBuffer[]; - pending: Pending; + resolve: (r: any) => void; + reject: (e: Error) => void; + timer: NodeJS.Timeout; + worker?: Worker; }; +const TASK_TIMEOUT_MS = 30_000; +const MAX_QUEUE_DEPTH = 2_000; +const RESPAWN_BASE_MS = 500; +const RESPAWN_MAX_MS = 30_000; + // Hand-rolled worker_threads pool: one in-flight task per worker so CPU-bound transforms run // fully in parallel. The main thread stays the only DB reader and broadcasts pricing to workers. export class OtlpWorkerPool { private readonly workers: Worker[] = []; private readonly idle: Worker[] = []; - private readonly queue: QueueItem[] = []; - private readonly pending = new Map(); + private readonly queue: number[] = []; + private readonly tasks = new Map(); private readonly busyByWorker = new Map(); private nextId = 1; + private consecutiveFailures = 0; private latestPricingModels: unknown[]; constructor( @@ -35,9 +44,7 @@ export class OtlpWorkerPool { pricingModels: unknown[] ) { this.latestPricingModels = pricingModels; - for (let i = 0; i < size; i++) { - this.spawn(); - } + for (let i = 0; i < size; i++) this.spawn(); logger.info("OtlpWorkerPool started", { size, workerPath }); } @@ -47,12 +54,14 @@ export class OtlpWorkerPool { }); worker.on("message", (msg: { id: number; ok: boolean; result?: any; error?: string }) => { - const pending = this.pending.get(msg.id); - this.pending.delete(msg.id); + this.consecutiveFailures = 0; this.busyByWorker.delete(worker); - if (pending) { - if (msg.ok) pending.resolve(msg.result); - else pending.reject(new Error(msg.error ?? "otlp worker error")); + const task = this.tasks.get(msg.id); + if (task) { + clearTimeout(task.timer); + this.tasks.delete(msg.id); + if (msg.ok) task.resolve(msg.result); + else task.reject(new Error(msg.error ?? "otlp worker error")); } this.release(worker); }); @@ -72,22 +81,35 @@ export class OtlpWorkerPool { this.idle.push(worker); } - // On crash: fail its in-flight task, drop it, and spawn a replacement. + // On crash/timeout: fail the worker's in-flight task (if still pending), drop the worker, and + // respawn with exponential backoff so a persistently failing worker can't tight-loop. private reap(worker: Worker, error: Error) { const inFlightId = this.busyByWorker.get(worker); + this.busyByWorker.delete(worker); if (inFlightId !== undefined) { - const pending = this.pending.get(inFlightId); - this.pending.delete(inFlightId); - this.busyByWorker.delete(worker); - pending?.reject(error); + const task = this.tasks.get(inFlightId); + if (task) { + clearTimeout(task.timer); + this.tasks.delete(inFlightId); + task.reject(error); + } } const wi = this.workers.indexOf(worker); if (wi !== -1) this.workers.splice(wi, 1); const ii = this.idle.indexOf(worker); if (ii !== -1) this.idle.splice(ii, 1); void worker.terminate().catch(() => {}); - if (this.workers.length < this.size) this.spawn(); - this.drain(); + this.scheduleRespawn(); + } + + private scheduleRespawn() { + if (this.workers.length >= this.size) return; + const delay = Math.min(RESPAWN_BASE_MS * 2 ** this.consecutiveFailures, RESPAWN_MAX_MS); + this.consecutiveFailures++; + setTimeout(() => { + if (this.workers.length < this.size) this.spawn(); + this.drain(); + }, delay); } private release(worker: Worker) { @@ -98,21 +120,43 @@ export class OtlpWorkerPool { private drain() { while (this.queue.length > 0 && this.idle.length > 0) { const worker = this.idle.pop()!; - const item = this.queue.shift()!; - this.pending.set(item.message.id, item.pending); - this.busyByWorker.set(worker, item.message.id); - worker.postMessage(item.message, item.transfer); + const id = this.queue.shift()!; + const task = this.tasks.get(id); + if (!task) continue; + task.worker = worker; + this.busyByWorker.set(worker, id); + worker.postMessage(task.message, task.transfer); } } + private onTimeout(id: number) { + const task = this.tasks.get(id); + if (!task) return; + this.tasks.delete(id); + const err = new Error(`otlp worker task timed out after ${TASK_TIMEOUT_MS}ms`); + if (task.worker) { + // Dispatched to a stuck worker: reap it. The task is already removed, so reap won't + // double-reject. + this.reap(task.worker, err); + } else { + const qi = this.queue.indexOf(id); + if (qi !== -1) this.queue.splice(qi, 1); + } + task.reject(err); + } + runTransform( kind: TransformKind, payload: Uint8Array, config: { spanAttributeValueLengthLimit: number; defaultEventStore: string } ): Promise { + if (this.queue.length >= MAX_QUEUE_DEPTH) { + return Promise.reject(new Error("otlp worker pool queue is full")); + } const id = this.nextId++; return new Promise((resolve, reject) => { - this.queue.push({ + const timer = setTimeout(() => this.onTimeout(id), TASK_TIMEOUT_MS); + this.tasks.set(id, { message: { id, kind, @@ -122,8 +166,11 @@ export class OtlpWorkerPool { }, // Zero-copy the payload into the worker; the request owns a fresh ArrayBuffer. transfer: [payload.buffer as ArrayBuffer], - pending: { resolve, reject }, + resolve, + reject, + timer, }); + this.queue.push(id); this.drain(); }); } @@ -146,12 +193,14 @@ export class OtlpWorkerPool { let pool: OtlpWorkerPool | undefined; -export function getOtlpWorkerPool(size: number, pricingModels: unknown[]): OtlpWorkerPool { +export function getOtlpWorkerPool( + size: number, + pricingModels: unknown[], + workerPath?: string +): OtlpWorkerPool { if (!pool) { - const workerPath = - process.env.OTEL_TRANSFORM_WORKER_PATH ?? - path.join(process.cwd(), "build", "otlpTransformWorker.cjs"); - pool = new OtlpWorkerPool(size, workerPath, pricingModels); + const resolvedPath = workerPath ?? path.join(process.cwd(), "build", "otlpTransformWorker.cjs"); + pool = new OtlpWorkerPool(size, resolvedPath, pricingModels); } return pool; } From 18d1daae10eb4f43a58ba4d5cbd6c57e85137a90 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Sat, 11 Jul 2026 07:13:54 +0100 Subject: [PATCH 06/10] fix(webapp): guard the OTLP worker pool against double-reap and late messages A crashing worker fires both error and exit, so reap() ran twice (double respawn and double backoff increment). reap() now returns early once the worker has been removed. The message handler also ignores late messages from an already-reaped worker, which otherwise pushed a dead worker back into the idle set. --- apps/webapp/app/v3/otlpWorkerPool.server.ts | 13 +++++++++---- 1 file changed, 9 insertions(+), 4 deletions(-) diff --git a/apps/webapp/app/v3/otlpWorkerPool.server.ts b/apps/webapp/app/v3/otlpWorkerPool.server.ts index db3b214aed..06b19ce57d 100644 --- a/apps/webapp/app/v3/otlpWorkerPool.server.ts +++ b/apps/webapp/app/v3/otlpWorkerPool.server.ts @@ -54,6 +54,7 @@ export class OtlpWorkerPool { }); worker.on("message", (msg: { id: number; ok: boolean; result?: any; error?: string }) => { + if (this.workers.indexOf(worker) === -1) return; // late message from an already-reaped worker this.consecutiveFailures = 0; this.busyByWorker.delete(worker); const task = this.tasks.get(msg.id); @@ -84,6 +85,13 @@ export class OtlpWorkerPool { // On crash/timeout: fail the worker's in-flight task (if still pending), drop the worker, and // respawn with exponential backoff so a persistently failing worker can't tight-loop. private reap(worker: Worker, error: Error) { + const wi = this.workers.indexOf(worker); + if (wi === -1) return; // already reaped (error + exit can both fire for one crash) + this.workers.splice(wi, 1); + + const ii = this.idle.indexOf(worker); + if (ii !== -1) this.idle.splice(ii, 1); + const inFlightId = this.busyByWorker.get(worker); this.busyByWorker.delete(worker); if (inFlightId !== undefined) { @@ -94,10 +102,7 @@ export class OtlpWorkerPool { task.reject(error); } } - const wi = this.workers.indexOf(worker); - if (wi !== -1) this.workers.splice(wi, 1); - const ii = this.idle.indexOf(worker); - if (ii !== -1) this.idle.splice(ii, 1); + void worker.terminate().catch(() => {}); this.scheduleRespawn(); } From a4ce5fe6a495f7aa13f51fb4ed2eb7e61958fc16 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Sat, 11 Jul 2026 07:52:49 +0100 Subject: [PATCH 07/10] fix(webapp): reap OTLP workers on any exit, not just non-zero A worker that exits cleanly (code 0) while holding a task previously left the request hanging until the task timeout and the pool one worker short. Reap on any exit; the double-reap guard makes this a no-op when error already reaped it. --- apps/webapp/app/v3/otlpWorkerPool.server.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/apps/webapp/app/v3/otlpWorkerPool.server.ts b/apps/webapp/app/v3/otlpWorkerPool.server.ts index 06b19ce57d..7c780bca0f 100644 --- a/apps/webapp/app/v3/otlpWorkerPool.server.ts +++ b/apps/webapp/app/v3/otlpWorkerPool.server.ts @@ -73,9 +73,9 @@ export class OtlpWorkerPool { }); worker.on("exit", (code) => { - if (code !== 0) { - this.reap(worker, new Error(`otlp worker exited with code ${code}`)); - } + // Any exit means this worker is gone, including a clean exit while it held a task; reap() + // no-ops if the worker was already removed (e.g. error fired first). + this.reap(worker, new Error(`otlp worker exited with code ${code}`)); }); this.workers.push(worker); From b0ecf59c90915b0292c02c76e4e22bd0c1a23149 Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Sat, 11 Jul 2026 09:44:02 +0100 Subject: [PATCH 08/10] feat(webapp): add ingest.* metrics for the telemetry ingestion pipeline Instruments the OTLP ingest path, the transform worker pool, and the shared flush scheduler with OpenTelemetry metrics under a single ingest.* prefix: request and byte throughput with per-signal produced counts, worker task vs compute durations plus queue depth and worker health, and flush batch size, duration, queue depth and dropped-batch counters. Reuses the existing MeterProvider and is gated by the existing INTERNAL_OTEL_METRIC_EXPORTER_ENABLED, so every instrument is a no-op when metrics are disabled. Recording happens in bulk once per request or flush (never per span) with pull-based gauges, so it stays off the hot path. --- .../app/v3/dynamicFlushScheduler.server.ts | 75 +++++++ .../clickhouseEventRepository.server.ts | 18 +- .../eventRepository/eventRepository.server.ts | 1 + apps/webapp/app/v3/otlpExporter.server.ts | 184 +++++++++++++----- apps/webapp/app/v3/otlpTransformWorker.ts | 6 +- apps/webapp/app/v3/otlpWorkerPool.server.ts | 148 ++++++++++++-- .../test/dynamicFlushSchedulerMetrics.test.ts | 133 +++++++++++++ apps/webapp/test/fixtures/otlpEchoWorker.cjs | 7 + apps/webapp/test/fixtures/otlpErrorWorker.cjs | 7 + apps/webapp/test/otlpMetrics.helpers.ts | 67 +++++++ .../webapp/test/otlpWorkerPoolMetrics.test.ts | 75 +++++++ 11 files changed, 649 insertions(+), 72 deletions(-) create mode 100644 apps/webapp/test/dynamicFlushSchedulerMetrics.test.ts create mode 100644 apps/webapp/test/fixtures/otlpEchoWorker.cjs create mode 100644 apps/webapp/test/fixtures/otlpErrorWorker.cjs create mode 100644 apps/webapp/test/otlpMetrics.helpers.ts create mode 100644 apps/webapp/test/otlpWorkerPoolMetrics.test.ts diff --git a/apps/webapp/app/v3/dynamicFlushScheduler.server.ts b/apps/webapp/app/v3/dynamicFlushScheduler.server.ts index 79db87de9d..850ad0e375 100644 --- a/apps/webapp/app/v3/dynamicFlushScheduler.server.ts +++ b/apps/webapp/app/v3/dynamicFlushScheduler.server.ts @@ -1,5 +1,6 @@ import { Logger } from "@trigger.dev/core/logger"; import { tryCatch } from "@trigger.dev/core/utils"; +import { getMeter, type Counter, type Histogram, type Meter } from "@internal/tracing"; import { nanoid } from "nanoid"; import pLimit from "p-limit"; import { signalsEmitter } from "~/services/signals.server"; @@ -16,6 +17,11 @@ export type DynamicFlushSchedulerConfig = { loadSheddingThreshold?: number; // Number of items that triggers load shedding loadSheddingEnabled?: boolean; isDroppableEvent?: (item: T) => boolean; // Function to determine if an event can be dropped + // Self-observability. `name` is the low-cardinality `scheduler` label that separates the + // task_events / llm_metrics / otlp_metrics instances in the same process. `meter` defaults to + // the global provider; inject one in tests. Instruments are no-op unless metrics are enabled. + meter?: Meter; + name?: string; }; export class DynamicFlushScheduler { @@ -54,7 +60,21 @@ export class DynamicFlushScheduler { private readonly logger: Logger = new Logger("EventRepo.DynamicFlushScheduler", "info"); + // Pre-allocated attribute objects (closed label sets) so the hot flush path never allocates. + private readonly _metricAttrs: { scheduler: string }; + private readonly _batchOkAttrs: { scheduler: string; outcome: string }; + private readonly _batchFailedAttrs: { scheduler: string; outcome: string }; + private _batchesCounter?: Counter; + private _itemsCounter?: Counter; + private _flushDurationHistogram?: Histogram; + private _batchSizeHistogram?: Histogram; + private _droppedEventsCounter?: Counter; + constructor(config: DynamicFlushSchedulerConfig) { + const schedulerName = config.name ?? "unknown"; + this._metricAttrs = { scheduler: schedulerName }; + this._batchOkAttrs = { scheduler: schedulerName, outcome: "ok" }; + this._batchFailedAttrs = { scheduler: schedulerName, outcome: "failed" }; this.batchQueue = []; this.currentBatch = []; this.BATCH_SIZE = config.batchSize; @@ -80,6 +100,54 @@ export class DynamicFlushScheduler { this.startFlushTimer(); this.startMetricsReporter(); this.setupShutdownHandlers(); + this.#setupOtelMetrics(config.meter, schedulerName); + } + + #setupOtelMetrics(meterOverride: Meter | undefined, name: string): void { + const meter = meterOverride ?? getMeter("ingest-flush"); + + this._batchesCounter = meter.createCounter("ingest.flush.batches", { + description: "Batches flushed to the sink, by outcome", + unit: "batches", + }); + this._itemsCounter = meter.createCounter("ingest.flush.items", { + description: "Items successfully flushed to the sink", + unit: "items", + }); + this._flushDurationHistogram = meter.createHistogram("ingest.flush.duration", { + description: "Wall-clock duration of a single batch flush", + unit: "ms", + }); + this._batchSizeHistogram = meter.createHistogram("ingest.flush.batch_size", { + description: "Number of items in a flushed batch", + unit: "items", + }); + this._droppedEventsCounter = meter.createCounter("ingest.flush.dropped_events", { + description: "Events dropped by load shedding before they reached the sink", + unit: "events", + }); + + // Pull-based gauges: read at export time only, so they add zero hot-path cost. + const queueDepthGauge = meter.createObservableGauge("ingest.flush.queue_depth", { + description: "Items queued and awaiting flush", + unit: "items", + }); + const concurrencyGauge = meter.createObservableGauge("ingest.flush.concurrency", { + description: "Current concurrent-flush limit", + unit: "flushes", + }); + const loadSheddingGauge = meter.createObservableGauge("ingest.flush.load_shedding", { + description: "1 while actively shedding load, otherwise 0", + }); + + meter.addBatchObservableCallback( + (result) => { + result.observe(queueDepthGauge, this.totalQueuedItems, this._metricAttrs); + result.observe(concurrencyGauge, this.limiter.concurrency, this._metricAttrs); + result.observe(loadSheddingGauge, this.isLoadShedding ? 1 : 0, this._metricAttrs); + }, + [queueDepthGauge, concurrencyGauge, loadSheddingGauge] + ); } addToBatch(items: T[]): void { @@ -92,6 +160,7 @@ export class DynamicFlushScheduler { if (dropped.length > 0) { this.metrics.droppedEvents += dropped.length; + this._droppedEventsCounter?.add(dropped.length, this._metricAttrs); // Track dropped events by kind if possible dropped.forEach((item) => { @@ -213,6 +282,11 @@ export class DynamicFlushScheduler { self.metrics.flushedBatches++; self.metrics.totalItemsFlushed += itemCount; + self._flushDurationHistogram?.record(duration, self._metricAttrs); + self._batchSizeHistogram?.record(itemCount, self._metricAttrs); + self._itemsCounter?.add(itemCount, self._metricAttrs); + self._batchesCounter?.add(1, self._batchOkAttrs); + self.logger.debug("Batch flushed successfully", { flushId, itemCount, @@ -253,6 +327,7 @@ export class DynamicFlushScheduler { this.logger.error("Error flushing batch", { error: flushError, }); + this._batchesCounter?.add(1, this._batchFailedAttrs); } }) ); diff --git a/apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts b/apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts index 25f2bac347..573097005e 100644 --- a/apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts +++ b/apps/webapp/app/v3/eventRepository/clickhouseEventRepository.server.ts @@ -8,8 +8,8 @@ import type { TaskEventV1Input, TaskEventV2Input, } from "@internal/clickhouse"; -import type { Attributes, Tracer } from "@internal/tracing"; -import { startSpan, trace } from "@internal/tracing"; +import type { Attributes, Counter, Meter, Tracer } from "@internal/tracing"; +import { getMeter, startSpan, trace } from "@internal/tracing"; import { createJsonErrorObject } from "@trigger.dev/core/v3/errors"; import { serializeTraceparent } from "@trigger.dev/core/v3/isomorphic"; @@ -104,6 +104,8 @@ export type ClickhouseEventRepositoryConfig = { otlpMetricsBatchSize?: number; otlpMetricsFlushInterval?: number; otlpMetricsMaxConcurrency?: number; + /** Inject a meter for self-observability; defaults to the global provider. */ + meter?: Meter; }; /** @@ -125,6 +127,7 @@ export class ClickhouseEventRepository implements IEventRepository { * track the drop count for observability. */ private _permanentlyDroppedBatches = 0; + private readonly _droppedBatchesCounter: Counter; constructor(config: ClickhouseEventRepositoryConfig) { this._clickhouse = config.clickhouse; @@ -132,7 +135,14 @@ export class ClickhouseEventRepository implements IEventRepository { this._tracer = config.tracer ?? trace.getTracer("clickhouseEventRepo", "0.0.1"); this._version = config.version ?? "v1"; + const meter = config.meter ?? getMeter("ingest-flush"); + this._droppedBatchesCounter = meter.createCounter("ingest.flush.batches_dropped", { + description: "Batches permanently dropped after an unrecoverable ClickHouse JSON parse error", + unit: "batches", + }); + this._flushScheduler = new DynamicFlushScheduler({ + name: `task_events_${this._version}`, batchSize: config.batchSize ?? 1000, flushInterval: config.flushInterval ?? 1000, callback: this.#flushBatch.bind(this), @@ -149,6 +159,7 @@ export class ClickhouseEventRepository implements IEventRepository { }); this._llmMetricsFlushScheduler = new DynamicFlushScheduler({ + name: "llm_metrics", batchSize: config.llmMetricsBatchSize ?? 5000, flushInterval: config.llmMetricsFlushInterval ?? 2000, callback: this.#flushLlmMetricsBatch.bind(this), @@ -160,6 +171,7 @@ export class ClickhouseEventRepository implements IEventRepository { }); this._otlpMetricsFlushScheduler = new DynamicFlushScheduler({ + name: "otlp_metrics", batchSize: config.otlpMetricsBatchSize ?? 10000, flushInterval: config.otlpMetricsFlushInterval ?? 1000, callback: this.#flushOtelMetricsBatch.bind(this), @@ -359,6 +371,7 @@ export class ClickhouseEventRepository implements IEventRepository { // exactly the retry storm this wrapper is designed to avoid. if (fieldsSanitized === 0) { this._permanentlyDroppedBatches += 1; + this._droppedBatchesCounter.add(1, { table: contextLabel }); logger.error( "Dropped batch — ClickHouse JSON parse error but sanitizer found nothing to fix", { @@ -390,6 +403,7 @@ export class ClickhouseEventRepository implements IEventRepository { if (!isClickHouseJsonParseError(retryError)) throw retryError; this._permanentlyDroppedBatches += 1; + this._droppedBatchesCounter.add(1, { table: contextLabel }); const retryMessage = typeof retryError === "object" && retryError !== null && "message" in retryError ? String((retryError as { message?: unknown }).message ?? "") diff --git a/apps/webapp/app/v3/eventRepository/eventRepository.server.ts b/apps/webapp/app/v3/eventRepository/eventRepository.server.ts index 1d1f5350b6..2de5ce8b25 100644 --- a/apps/webapp/app/v3/eventRepository/eventRepository.server.ts +++ b/apps/webapp/app/v3/eventRepository/eventRepository.server.ts @@ -101,6 +101,7 @@ export class EventRepository implements IEventRepository { private readonly _config: EventRepoConfig ) { this._flushScheduler = new DynamicFlushScheduler({ + name: "postgres_events", batchSize: _config.batchSize, flushInterval: _config.batchInterval, callback: this.#flushBatch.bind(this), diff --git a/apps/webapp/app/v3/otlpExporter.server.ts b/apps/webapp/app/v3/otlpExporter.server.ts index 6b51315ed5..da4ca4605e 100644 --- a/apps/webapp/app/v3/otlpExporter.server.ts +++ b/apps/webapp/app/v3/otlpExporter.server.ts @@ -13,6 +13,7 @@ import { ExportTraceServiceResponse, } from "@trigger.dev/otlp-importer"; import type { MetricsV1Input } from "@internal/clickhouse"; +import { getMeter, type Counter, type Histogram, type Meter } from "@internal/tracing"; import { logger } from "~/services/logger.server"; import type { ClickhouseFactory } from "~/services/clickhouse/clickhouseFactory.server"; import { clickhouseFactory } from "~/services/clickhouse/clickhouseFactoryInstance.server"; @@ -52,6 +53,8 @@ type OTLPExporterConfig = { clickhouseFactory: ClickhouseFactory; verbose: boolean; spanAttributeValueLengthLimit: number; + // Inject in tests; defaults to the global provider. Instruments are no-op when metrics are off. + meter?: Meter; }; class OTLPExporter { @@ -61,70 +64,132 @@ class OTLPExporter { private readonly _spanAttributeValueLengthLimit: number; #pricingSubscribed = false; + private readonly _meter: Meter; + private readonly _ingestRequests: Counter; + private readonly _ingestBytes: Counter; + private readonly _ingestDuration: Histogram; + private readonly _eventsProduced: Counter; + private readonly _metricRowsProduced: Counter; + constructor(config: OTLPExporterConfig) { this._tracer = trace.getTracer("otlp-exporter"); this._clickhouseFactory = config.clickhouseFactory; this._verbose = config.verbose; this._spanAttributeValueLengthLimit = config.spanAttributeValueLengthLimit; + + this._meter = config.meter ?? getMeter("ingest"); + this._ingestRequests = this._meter.createCounter("ingest.requests", { + description: "OTLP export calls received, by signal / path / outcome", + unit: "requests", + }); + this._ingestBytes = this._meter.createCounter("ingest.bytes", { + description: "Compressed protobuf payload bytes received", + unit: "By", + }); + this._ingestDuration = this._meter.createHistogram("ingest.duration", { + description: "End-to-end time to handle an OTLP export call", + unit: "ms", + }); + this._eventsProduced = this._meter.createCounter("ingest.events.produced", { + description: "Task events produced from spans/logs after filter + convert", + unit: "events", + }); + this._metricRowsProduced = this._meter.createCounter("ingest.metric_rows.produced", { + description: "ClickHouse metric rows produced from OTLP metrics", + unit: "rows", + }); + } + + // One helper for both the worker (mode="worker") and inline (mode="inline") paths. Called once + // per export call, so building the small attribute objects here is not a hot-path allocation. + #recordIngest(kind: string, mode: string, outcome: string, startedAt: number): void { + this._ingestRequests.add(1, { kind, mode, outcome }); + this._ingestDuration.record(Date.now() - startedAt, { kind, mode }); } async exportTraces(request: ExportTraceServiceRequest): Promise { return await startSpan(this._tracer, "exportTraces", async (span) => { - this.#logExportTracesVerbose(request); - - const eventsWithStores = this.#filterResourceSpans(request.resourceSpans).flatMap( - (resourceSpan) => { - return convertSpansToCreateableEvents( - resourceSpan, - this._spanAttributeValueLengthLimit, - env.EVENT_REPOSITORY_DEFAULT_STORE - ); - } - ); + const startedAt = Date.now(); + try { + this.#logExportTracesVerbose(request); + + const eventsWithStores = this.#filterResourceSpans(request.resourceSpans).flatMap( + (resourceSpan) => { + return convertSpansToCreateableEvents( + resourceSpan, + this._spanAttributeValueLengthLimit, + env.EVENT_REPOSITORY_DEFAULT_STORE + ); + } + ); - const eventCount = await this.#exportEvents(eventsWithStores); + const eventCount = await this.#exportEvents(eventsWithStores); - span.setAttribute("event_count", eventCount); + span.setAttribute("event_count", eventCount); + this._eventsProduced.add(eventCount, { kind: "traces" }); + this.#recordIngest("traces", "inline", "ok", startedAt); - return ExportTraceServiceResponse.create(); + return ExportTraceServiceResponse.create(); + } catch (error) { + this.#recordIngest("traces", "inline", "error", startedAt); + throw error; + } }); } async exportMetrics(request: ExportMetricsServiceRequest): Promise { return await startSpan(this._tracer, "exportMetrics", async (span) => { - const rows = this.#filterResourceMetrics(request.resourceMetrics).flatMap((resourceMetrics) => - convertMetricsToClickhouseRows(resourceMetrics, this._spanAttributeValueLengthLimit) - ); + const startedAt = Date.now(); + try { + const rows = this.#filterResourceMetrics(request.resourceMetrics).flatMap( + (resourceMetrics) => + convertMetricsToClickhouseRows(resourceMetrics, this._spanAttributeValueLengthLimit) + ); - span.setAttribute("metric_row_count", rows.length); + span.setAttribute("metric_row_count", rows.length); - if (rows.length > 0) { - await this.#exportMetricRows(rows); - } + if (rows.length > 0) { + await this.#exportMetricRows(rows); + } + + this._metricRowsProduced.add(rows.length, { kind: "metrics" }); + this.#recordIngest("metrics", "inline", "ok", startedAt); - return ExportMetricsServiceResponse.create(); + return ExportMetricsServiceResponse.create(); + } catch (error) { + this.#recordIngest("metrics", "inline", "error", startedAt); + throw error; + } }); } async exportLogs(request: ExportLogsServiceRequest): Promise { return await startSpan(this._tracer, "exportLogs", async (span) => { - this.#logExportLogsVerbose(request); - - const eventsWithStores = this.#filterResourceLogs(request.resourceLogs).flatMap( - (resourceLog) => { - return convertLogsToCreateableEvents( - resourceLog, - this._spanAttributeValueLengthLimit, - env.EVENT_REPOSITORY_DEFAULT_STORE - ); - } - ); + const startedAt = Date.now(); + try { + this.#logExportLogsVerbose(request); + + const eventsWithStores = this.#filterResourceLogs(request.resourceLogs).flatMap( + (resourceLog) => { + return convertLogsToCreateableEvents( + resourceLog, + this._spanAttributeValueLengthLimit, + env.EVENT_REPOSITORY_DEFAULT_STORE + ); + } + ); - const eventCount = await this.#exportEvents(eventsWithStores); + const eventCount = await this.#exportEvents(eventsWithStores); - span.setAttribute("event_count", eventCount); + span.setAttribute("event_count", eventCount); + this._eventsProduced.add(eventCount, { kind: "logs" }); + this.#recordIngest("logs", "inline", "ok", startedAt); - return ExportLogsServiceResponse.create(); + return ExportLogsServiceResponse.create(); + } catch (error) { + this.#recordIngest("logs", "inline", "error", startedAt); + throw error; + } }); } @@ -138,11 +203,20 @@ class OTLPExporter { async exportMetricsRaw(payload: Uint8Array): Promise { 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); + const startedAt = Date.now(); + this._ingestBytes.add(payload.byteLength, { kind: "metrics" }); + try { + 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); + } + this._metricRowsProduced.add(rows.length, { kind: "metrics" }); + this.#recordIngest("metrics", "worker", "ok", startedAt); + } catch (error) { + this.#recordIngest("metrics", "worker", "error", startedAt); + throw error; } }); } @@ -152,14 +226,23 @@ class OTLPExporter { 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); + const startedAt = Date.now(); + this._ingestBytes.add(payload.byteLength, { kind }); + try { + 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); + this._eventsProduced.add(eventCount, { kind }); + this.#recordIngest(kind, "worker", "ok", startedAt); + } catch (error) { + this.#recordIngest(kind, "worker", "error", startedAt); + throw error; + } } ); } @@ -178,7 +261,8 @@ class OTLPExporter { const pool = getOtlpWorkerPool( OTEL_TRANSFORM_WORKER_POOL_SIZE, models, - env.OTEL_TRANSFORM_WORKER_PATH + env.OTEL_TRANSFORM_WORKER_PATH, + this._meter ); if (!this.#pricingSubscribed) { this.#pricingSubscribed = true; diff --git a/apps/webapp/app/v3/otlpTransformWorker.ts b/apps/webapp/app/v3/otlpTransformWorker.ts index c158e10ce8..a8f3d3dfb5 100644 --- a/apps/webapp/app/v3/otlpTransformWorker.ts +++ b/apps/webapp/app/v3/otlpTransformWorker.ts @@ -90,8 +90,12 @@ parentPort.on("message", (message: TransformTask | PricingUpdate) => { const task = message as TransformTask; try { + // The worker has no MeterProvider, so it can't emit metrics itself. It measures its own + // compute time (decode + convert + enrich) and hands it back for the main thread to record. + const startedAt = performance.now(); const result = runTask(task); - parentPort!.postMessage({ id: task.id, ok: true, result }); + const computeMs = performance.now() - startedAt; + parentPort!.postMessage({ id: task.id, ok: true, result, computeMs }); } catch (error) { parentPort!.postMessage({ id: task.id, diff --git a/apps/webapp/app/v3/otlpWorkerPool.server.ts b/apps/webapp/app/v3/otlpWorkerPool.server.ts index 7c780bca0f..e8fb2d5b2b 100644 --- a/apps/webapp/app/v3/otlpWorkerPool.server.ts +++ b/apps/webapp/app/v3/otlpWorkerPool.server.ts @@ -1,5 +1,12 @@ import { Worker } from "node:worker_threads"; import path from "node:path"; +import { + getMeter, + type Counter, + type Histogram, + type Meter, + type ObservableGauge, +} from "@internal/tracing"; import { logger } from "~/services/logger.server"; export type TransformKind = "traces" | "logs" | "metrics"; @@ -19,8 +26,13 @@ type Task = { reject: (e: Error) => void; timer: NodeJS.Timeout; worker?: Worker; + // Wall-clock stamp at enqueue; the task-duration histogram measures enqueue -> terminal state + // (queue wait + worker compute), so the gap from the worker-reported compute time is queue wait. + enqueuedAt: number; }; +type ReapReason = "error" | "exit" | "timeout"; + const TASK_TIMEOUT_MS = 30_000; const MAX_QUEUE_DEPTH = 2_000; const RESPAWN_BASE_MS = 500; @@ -38,44 +50,119 @@ export class OtlpWorkerPool { private consecutiveFailures = 0; private latestPricingModels: unknown[]; + // Pre-allocated per-kind {kind} attribute objects so the per-task record path never allocates. + private readonly _kindAttrs: Record = { + traces: { kind: "traces" }, + logs: { kind: "logs" }, + metrics: { kind: "metrics" }, + }; + private _taskDurationHistogram?: Histogram; + private _computeDurationHistogram?: Histogram; + private _tasksCounter?: Counter; + private _respawnsCounter?: Counter; + constructor( private readonly size: number, private readonly workerPath: string, - pricingModels: unknown[] + pricingModels: unknown[], + meter?: Meter ) { this.latestPricingModels = pricingModels; + this.#setupOtelMetrics(meter); for (let i = 0; i < size; i++) this.spawn(); logger.info("OtlpWorkerPool started", { size, workerPath }); } + #setupOtelMetrics(meterOverride: Meter | undefined): void { + const meter = meterOverride ?? getMeter("ingest"); + + this._taskDurationHistogram = meter.createHistogram("ingest.worker_pool.task.duration", { + description: "Enqueue-to-completion time for a transform task (queue wait + worker compute)", + unit: "ms", + }); + this._computeDurationHistogram = meter.createHistogram("ingest.worker_pool.compute.duration", { + description: "Worker-reported compute time (decode + convert + enrich)", + unit: "ms", + }); + this._tasksCounter = meter.createCounter("ingest.worker_pool.tasks", { + description: "Transform tasks by terminal outcome", + unit: "tasks", + }); + this._respawnsCounter = meter.createCounter("ingest.worker_pool.respawns", { + description: "Worker respawns by reason", + unit: "respawns", + }); + + // Pull-based gauges: read at export time only, zero hot-path cost. + const queueDepthGauge: ObservableGauge = meter.createObservableGauge( + "ingest.worker_pool.queue_depth", + { description: "Tasks queued and awaiting a free worker", unit: "tasks" } + ); + const workersGauge: ObservableGauge = meter.createObservableGauge( + "ingest.worker_pool.workers", + { + description: "Pool workers by state (alive workers, idle workers)", + unit: "workers", + } + ); + + meter.addBatchObservableCallback( + (result) => { + result.observe(queueDepthGauge, this.queue.length); + result.observe(workersGauge, this.workers.length, { state: "alive" }); + result.observe(workersGauge, this.idle.length, { state: "idle" }); + }, + [queueDepthGauge, workersGauge] + ); + } + + #recordTaskEnd(task: Task, outcome: string, computeMs?: number): void { + this._taskDurationHistogram?.record( + Date.now() - task.enqueuedAt, + this._kindAttrs[task.message.kind] + ); + this._tasksCounter?.add(1, { kind: task.message.kind, outcome }); + if (computeMs !== undefined) { + this._computeDurationHistogram?.record(computeMs, this._kindAttrs[task.message.kind]); + } + } + private spawn() { const worker = new Worker(this.workerPath, { workerData: { pricingModels: this.latestPricingModels }, }); - worker.on("message", (msg: { id: number; ok: boolean; result?: any; error?: string }) => { - if (this.workers.indexOf(worker) === -1) return; // late message from an already-reaped worker - this.consecutiveFailures = 0; - this.busyByWorker.delete(worker); - const task = this.tasks.get(msg.id); - if (task) { - clearTimeout(task.timer); - this.tasks.delete(msg.id); - if (msg.ok) task.resolve(msg.result); - else task.reject(new Error(msg.error ?? "otlp worker error")); + worker.on( + "message", + (msg: { id: number; ok: boolean; result?: any; error?: string; computeMs?: number }) => { + if (this.workers.indexOf(worker) === -1) return; // late message from an already-reaped worker + this.consecutiveFailures = 0; + this.busyByWorker.delete(worker); + const task = this.tasks.get(msg.id); + if (task) { + clearTimeout(task.timer); + this.tasks.delete(msg.id); + if (msg.ok) { + this.#recordTaskEnd(task, "ok", msg.computeMs); + task.resolve(msg.result); + } else { + this.#recordTaskEnd(task, "error", msg.computeMs); + task.reject(new Error(msg.error ?? "otlp worker error")); + } + } + this.release(worker); } - this.release(worker); - }); + ); worker.on("error", (error) => { logger.error("OtlpWorkerPool worker error", { error: error.message }); - this.reap(worker, error); + this.reap(worker, error, "error"); }); worker.on("exit", (code) => { // Any exit means this worker is gone, including a clean exit while it held a task; reap() // no-ops if the worker was already removed (e.g. error fired first). - this.reap(worker, new Error(`otlp worker exited with code ${code}`)); + this.reap(worker, new Error(`otlp worker exited with code ${code}`), "exit"); }); this.workers.push(worker); @@ -84,7 +171,7 @@ export class OtlpWorkerPool { // On crash/timeout: fail the worker's in-flight task (if still pending), drop the worker, and // respawn with exponential backoff so a persistently failing worker can't tight-loop. - private reap(worker: Worker, error: Error) { + private reap(worker: Worker, error: Error, reason: ReapReason) { const wi = this.workers.indexOf(worker); if (wi === -1) return; // already reaped (error + exit can both fire for one crash) this.workers.splice(wi, 1); @@ -99,10 +186,14 @@ export class OtlpWorkerPool { if (task) { clearTimeout(task.timer); this.tasks.delete(inFlightId); + // A timed-out task already recorded its own end + was removed from the map, so this only + // fires for a crash that killed a task mid-flight. + this.#recordTaskEnd(task, "crash"); task.reject(error); } } + this._respawnsCounter?.add(1, { reason }); void worker.terminate().catch(() => {}); this.scheduleRespawn(); } @@ -138,11 +229,12 @@ export class OtlpWorkerPool { const task = this.tasks.get(id); if (!task) return; this.tasks.delete(id); + this.#recordTaskEnd(task, "timeout"); const err = new Error(`otlp worker task timed out after ${TASK_TIMEOUT_MS}ms`); if (task.worker) { // Dispatched to a stuck worker: reap it. The task is already removed, so reap won't // double-reject. - this.reap(task.worker, err); + this.reap(task.worker, err, "timeout"); } else { const qi = this.queue.indexOf(id); if (qi !== -1) this.queue.splice(qi, 1); @@ -156,6 +248,7 @@ export class OtlpWorkerPool { config: { spanAttributeValueLengthLimit: number; defaultEventStore: string } ): Promise { if (this.queue.length >= MAX_QUEUE_DEPTH) { + this._tasksCounter?.add(1, { kind, outcome: "rejected" }); return Promise.reject(new Error("otlp worker pool queue is full")); } const id = this.nextId++; @@ -174,6 +267,7 @@ export class OtlpWorkerPool { resolve, reject, timer, + enqueuedAt: Date.now(), }); this.queue.push(id); this.drain(); @@ -194,6 +288,21 @@ export class OtlpWorkerPool { get queueDepth() { return this.queue.length; } + + // Terminate all workers and reject anything in flight. Terminated workers fire "exit", but reap() + // no-ops on an already-removed worker, so no respawn is scheduled. + async shutdown(): Promise { + const workers = this.workers.splice(0); + this.idle.length = 0; + this.queue.length = 0; + this.busyByWorker.clear(); + for (const [, task] of this.tasks) { + clearTimeout(task.timer); + task.reject(new Error("otlp worker pool shutting down")); + } + this.tasks.clear(); + await Promise.all(workers.map((worker) => worker.terminate().catch(() => {}))); + } } let pool: OtlpWorkerPool | undefined; @@ -201,11 +310,12 @@ let pool: OtlpWorkerPool | undefined; export function getOtlpWorkerPool( size: number, pricingModels: unknown[], - workerPath?: string + workerPath?: string, + meter?: Meter ): OtlpWorkerPool { if (!pool) { const resolvedPath = workerPath ?? path.join(process.cwd(), "build", "otlpTransformWorker.cjs"); - pool = new OtlpWorkerPool(size, resolvedPath, pricingModels); + pool = new OtlpWorkerPool(size, resolvedPath, pricingModels, meter); } return pool; } diff --git a/apps/webapp/test/dynamicFlushSchedulerMetrics.test.ts b/apps/webapp/test/dynamicFlushSchedulerMetrics.test.ts new file mode 100644 index 0000000000..b969675ead --- /dev/null +++ b/apps/webapp/test/dynamicFlushSchedulerMetrics.test.ts @@ -0,0 +1,133 @@ +import { afterEach, describe, expect, it, vi } from "vitest"; +import { DynamicFlushScheduler } from "~/v3/dynamicFlushScheduler.server"; +import { createInMemoryMetrics } from "./utils/tracing"; +import { gaugeValue, latestMetrics, metricSum } from "./otlpMetrics.helpers"; + +type Item = { id: number }; + +describe("DynamicFlushScheduler self-observability", () => { + const cleanups: Array<() => Promise> = []; + + afterEach(async () => { + for (const cleanup of cleanups.splice(0)) { + await cleanup(); + } + }); + + it("records flush counters, histograms and gauges on a successful flush", async () => { + const metrics = createInMemoryMetrics(); + const flushed: number[] = []; + + const scheduler = new DynamicFlushScheduler({ + name: "test_events", + batchSize: 5, + flushInterval: 50, + meter: metrics.meter, + loadSheddingEnabled: false, + callback: async (_flushId, batch) => { + flushed.push(batch.length); + }, + }); + cleanups.push(async () => { + await scheduler.shutdown(); + await metrics.shutdown(); + }); + + // Reaching batchSize triggers an immediate flush. + scheduler.addToBatch([{ id: 1 }, { id: 2 }, { id: 3 }, { id: 4 }, { id: 5 }]); + + await vi.waitFor( + async () => { + const rm = await latestMetrics(metrics); + expect(metricSum(rm, "ingest.flush.items", { scheduler: "test_events" })).toBe(5); + }, + { timeout: 4000, interval: 50 } + ); + + expect(flushed).toEqual([5]); + + const rm = await latestMetrics(metrics); + expect(metricSum(rm, "ingest.flush.batches", { scheduler: "test_events", outcome: "ok" })).toBe( + 1 + ); + expect(metricSum(rm, "ingest.flush.batch_size", { scheduler: "test_events" })).toBe(5); + // Gauges are pull-based; the export we just collected observed the current state. + expect(gaugeValue(rm, "ingest.flush.queue_depth", { scheduler: "test_events" })).toBeDefined(); + expect( + gaugeValue(rm, "ingest.flush.concurrency", { scheduler: "test_events" }) + ).toBeGreaterThanOrEqual(1); + }); + + it("labels each scheduler instance separately", async () => { + const metrics = createInMemoryMetrics(); + + const makeScheduler = (name: string) => { + const s = new DynamicFlushScheduler({ + name, + batchSize: 2, + flushInterval: 50, + meter: metrics.meter, + loadSheddingEnabled: false, + callback: async () => {}, + }); + cleanups.push(async () => s.shutdown()); + return s; + }; + + const a = makeScheduler("task_events_v2"); + const b = makeScheduler("llm_metrics"); + cleanups.push(async () => metrics.shutdown()); + + a.addToBatch([{ id: 1 }, { id: 2 }]); + b.addToBatch([{ id: 3 }, { id: 4 }, { id: 5 }, { id: 6 }]); + + await vi.waitFor( + async () => { + const rm = await latestMetrics(metrics); + expect(metricSum(rm, "ingest.flush.items", { scheduler: "task_events_v2" })).toBe(2); + expect(metricSum(rm, "ingest.flush.items", { scheduler: "llm_metrics" })).toBe(4); + }, + { timeout: 4000, interval: 50 } + ); + }); + + it("counts a permanently failing flush as a failed batch", async () => { + const metrics = createInMemoryMetrics(); + + const scheduler = new DynamicFlushScheduler({ + name: "failing_events", + batchSize: 1, + flushInterval: 50, + meter: metrics.meter, + loadSheddingEnabled: false, + callback: async () => { + throw new Error("insert failed"); + }, + }); + cleanups.push(async () => { + await scheduler.shutdown(); + await metrics.shutdown(); + }); + + scheduler.addToBatch([{ id: 1 }]); + + // The scheduler retries 3x with a 500ms backoff before giving up, so allow ~2s. + await vi.waitFor( + async () => { + const rm = await latestMetrics(metrics); + expect( + metricSum(rm, "ingest.flush.batches", { + scheduler: "failing_events", + outcome: "failed", + }) + ).toBeGreaterThanOrEqual(1); + }, + { timeout: 8000, interval: 100 } + ); + + const rm = await latestMetrics(metrics); + expect( + metricSum(rm, "ingest.flush.batches", { scheduler: "failing_events", outcome: "ok" }) + ).toBe(0); + }); +}); diff --git a/apps/webapp/test/fixtures/otlpEchoWorker.cjs b/apps/webapp/test/fixtures/otlpEchoWorker.cjs new file mode 100644 index 0000000000..9ba28ffa83 --- /dev/null +++ b/apps/webapp/test/fixtures/otlpEchoWorker.cjs @@ -0,0 +1,7 @@ +// Minimal real worker for OtlpWorkerPool metric tests: echoes an ok result with a compute time. +const { parentPort } = require("node:worker_threads"); + +parentPort.on("message", (message) => { + if (message && message.type === "pricing") return; + parentPort.postMessage({ id: message.id, ok: true, result: { rows: [] }, computeMs: 2 }); +}); diff --git a/apps/webapp/test/fixtures/otlpErrorWorker.cjs b/apps/webapp/test/fixtures/otlpErrorWorker.cjs new file mode 100644 index 0000000000..6b37c40b52 --- /dev/null +++ b/apps/webapp/test/fixtures/otlpErrorWorker.cjs @@ -0,0 +1,7 @@ +// Real worker that always reports a failed task, to exercise the pool's error-outcome metric. +const { parentPort } = require("node:worker_threads"); + +parentPort.on("message", (message) => { + if (message && message.type === "pricing") return; + parentPort.postMessage({ id: message.id, ok: false, error: "boom" }); +}); diff --git a/apps/webapp/test/otlpMetrics.helpers.ts b/apps/webapp/test/otlpMetrics.helpers.ts new file mode 100644 index 0000000000..7141d85e7e --- /dev/null +++ b/apps/webapp/test/otlpMetrics.helpers.ts @@ -0,0 +1,67 @@ +import type { createInMemoryMetrics } from "./utils/tracing"; + +type MetricsHelper = ReturnType; + +// With cumulative temporality the latest export carries running totals for every instrument. +export async function latestMetrics(helper: MetricsHelper) { + const all = await helper.getMetrics(); + return all[all.length - 1]; +} + +export function findMetric(resourceMetrics: any, name: string): any | undefined { + if (!resourceMetrics) return undefined; + for (const scopeMetrics of resourceMetrics.scopeMetrics) { + for (const metric of scopeMetrics.metrics) { + if (metric.descriptor.name === name) return metric; + } + } + return undefined; +} + +function pointValue(dp: any): number { + const value = dp.value; + if (typeof value === "number") return value; + // Histogram / ExponentialHistogram data point + return value?.sum ?? 0; +} + +function matches(dp: any, attrs?: Record): boolean { + if (!attrs) return true; + return Object.entries(attrs).every(([k, v]) => String(dp.attributes?.[k]) === v); +} + +export function metricSum( + resourceMetrics: any, + name: string, + attrs?: Record +): number { + const metric = findMetric(resourceMetrics, name); + if (!metric) return 0; + return metric.dataPoints + .filter((dp: any) => matches(dp, attrs)) + .reduce((acc: number, dp: any) => acc + pointValue(dp), 0); +} + +export function histogramCount( + resourceMetrics: any, + name: string, + attrs?: Record +): number { + const metric = findMetric(resourceMetrics, name); + if (!metric) return 0; + return metric.dataPoints + .filter((dp: any) => matches(dp, attrs)) + .reduce((acc: number, dp: any) => acc + (dp.value?.count ?? 0), 0); +} + +export function gaugeValue( + resourceMetrics: any, + name: string, + attrs?: Record +): number | undefined { + const metric = findMetric(resourceMetrics, name); + if (!metric) return undefined; + const points = metric.dataPoints.filter((dp: any) => matches(dp, attrs)); + if (points.length === 0) return undefined; + return Math.max(...points.map((dp: any) => pointValue(dp))); +} diff --git a/apps/webapp/test/otlpWorkerPoolMetrics.test.ts b/apps/webapp/test/otlpWorkerPoolMetrics.test.ts new file mode 100644 index 0000000000..b15ea509d1 --- /dev/null +++ b/apps/webapp/test/otlpWorkerPoolMetrics.test.ts @@ -0,0 +1,75 @@ +import { fileURLToPath } from "node:url"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import { OtlpWorkerPool } from "~/v3/otlpWorkerPool.server"; +import { createInMemoryMetrics } from "./utils/tracing"; +import { gaugeValue, histogramCount, latestMetrics, metricSum } from "./otlpMetrics.helpers"; + +const echoWorker = fileURLToPath(new URL("./fixtures/otlpEchoWorker.cjs", import.meta.url)); +const errorWorker = fileURLToPath(new URL("./fixtures/otlpErrorWorker.cjs", import.meta.url)); + +const config = { spanAttributeValueLengthLimit: 8192, defaultEventStore: "clickhouse" }; +const payload = () => new Uint8Array([1, 2, 3, 4]); + +describe("OtlpWorkerPool self-observability", () => { + const cleanups: Array<() => Promise> = []; + + afterEach(async () => { + for (const cleanup of cleanups.splice(0)) { + await cleanup(); + } + }); + + it("records task/compute duration, outcome counts and gauges for successful tasks", async () => { + const metrics = createInMemoryMetrics(); + const pool = new OtlpWorkerPool(2, echoWorker, [], metrics.meter); + cleanups.push(async () => { + await pool.shutdown(); + await metrics.shutdown(); + }); + + await Promise.all([ + pool.runTransform("traces", payload(), config), + pool.runTransform("logs", payload(), config), + pool.runTransform("traces", payload(), config), + ]); + + await vi.waitFor( + async () => { + const rm = await latestMetrics(metrics); + expect(metricSum(rm, "ingest.worker_pool.tasks", { kind: "traces", outcome: "ok" })).toBe( + 2 + ); + expect(metricSum(rm, "ingest.worker_pool.tasks", { kind: "logs", outcome: "ok" })).toBe(1); + }, + { timeout: 5000, interval: 50 } + ); + + const rm = await latestMetrics(metrics); + expect(histogramCount(rm, "ingest.worker_pool.task.duration")).toBeGreaterThanOrEqual(3); + // The worker reports its own compute time; the pool records it on the main thread. + expect(histogramCount(rm, "ingest.worker_pool.compute.duration")).toBeGreaterThanOrEqual(3); + expect(gaugeValue(rm, "ingest.worker_pool.workers", { state: "alive" })).toBe(2); + expect(gaugeValue(rm, "ingest.worker_pool.queue_depth")).toBeDefined(); + }); + + it("records a failed-task outcome when the worker reports an error", async () => { + const metrics = createInMemoryMetrics(); + const pool = new OtlpWorkerPool(1, errorWorker, [], metrics.meter); + cleanups.push(async () => { + await pool.shutdown(); + await metrics.shutdown(); + }); + + await expect(pool.runTransform("metrics", payload(), config)).rejects.toThrow(); + + await vi.waitFor( + async () => { + const rm = await latestMetrics(metrics); + expect( + metricSum(rm, "ingest.worker_pool.tasks", { kind: "metrics", outcome: "error" }) + ).toBe(1); + }, + { timeout: 5000, interval: 50 } + ); + }); +}); From 522105704c6cf2520df0fa6baac36b0aaaa12d9f Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Sat, 11 Jul 2026 10:03:54 +0100 Subject: [PATCH 09/10] fix(webapp): drain and terminate OTLP transform workers on shutdown Register SIGTERM/SIGINT handlers for the OTLP transform worker pool so that on shutdown it stops accepting new work, lets in-flight transforms finish (bounded), and terminates its workers, instead of leaving them to be force-killed. This avoids respawn churn and error-log noise during a graceful shutdown; the main thread stays the only DB writer, so inserts are unaffected. --- apps/webapp/app/v3/otlpWorkerPool.server.ts | 31 +++++++++++++++++-- .../webapp/test/otlpWorkerPoolMetrics.test.ts | 12 +++++++ 2 files changed, 41 insertions(+), 2 deletions(-) diff --git a/apps/webapp/app/v3/otlpWorkerPool.server.ts b/apps/webapp/app/v3/otlpWorkerPool.server.ts index e8fb2d5b2b..a7844f9174 100644 --- a/apps/webapp/app/v3/otlpWorkerPool.server.ts +++ b/apps/webapp/app/v3/otlpWorkerPool.server.ts @@ -8,6 +8,7 @@ import { type ObservableGauge, } from "@internal/tracing"; import { logger } from "~/services/logger.server"; +import { signalsEmitter } from "~/services/signals.server"; export type TransformKind = "traces" | "logs" | "metrics"; @@ -37,6 +38,7 @@ const TASK_TIMEOUT_MS = 30_000; const MAX_QUEUE_DEPTH = 2_000; const RESPAWN_BASE_MS = 500; const RESPAWN_MAX_MS = 30_000; +const SHUTDOWN_DRAIN_MS = 5_000; // Hand-rolled worker_threads pool: one in-flight task per worker so CPU-bound transforms run // fully in parallel. The main thread stays the only DB reader and broadcasts pricing to workers. @@ -48,6 +50,7 @@ export class OtlpWorkerPool { private readonly busyByWorker = new Map(); private nextId = 1; private consecutiveFailures = 0; + private isShuttingDown = false; private latestPricingModels: unknown[]; // Pre-allocated per-kind {kind} attribute objects so the per-task record path never allocates. @@ -199,10 +202,12 @@ export class OtlpWorkerPool { } private scheduleRespawn() { + if (this.isShuttingDown) return; if (this.workers.length >= this.size) return; const delay = Math.min(RESPAWN_BASE_MS * 2 ** this.consecutiveFailures, RESPAWN_MAX_MS); this.consecutiveFailures++; setTimeout(() => { + if (this.isShuttingDown) return; if (this.workers.length < this.size) this.spawn(); this.drain(); }, delay); @@ -247,6 +252,9 @@ export class OtlpWorkerPool { payload: Uint8Array, config: { spanAttributeValueLengthLimit: number; defaultEventStore: string } ): Promise { + if (this.isShuttingDown) { + return Promise.reject(new Error("otlp worker pool is shutting down")); + } if (this.queue.length >= MAX_QUEUE_DEPTH) { this._tasksCounter?.add(1, { kind, outcome: "rejected" }); return Promise.reject(new Error("otlp worker pool queue is full")); @@ -289,13 +297,28 @@ export class OtlpWorkerPool { return this.queue.length; } - // Terminate all workers and reject anything in flight. Terminated workers fire "exit", but reap() - // no-ops on an already-removed worker, so no respawn is scheduled. + // Stop taking new work, let in-flight tasks finish (bounded), then terminate every worker. + // Terminated workers fire "exit", but reap() no-ops on an already-removed worker, and the + // isShuttingDown guard stops any pending respawn, so shutdown is quiet. async shutdown(): Promise { + if (this.isShuttingDown) return; + this.isShuttingDown = true; + + logger.info("OtlpWorkerPool shutting down", { + workers: this.workers.length, + inFlight: this.tasks.size, + }); + + const deadline = Date.now() + SHUTDOWN_DRAIN_MS; + while (this.tasks.size > 0 && Date.now() < deadline) { + await new Promise((resolve) => setTimeout(resolve, 50)); + } + const workers = this.workers.splice(0); this.idle.length = 0; this.queue.length = 0; this.busyByWorker.clear(); + // Reject anything that didn't drain within the deadline. for (const [, task] of this.tasks) { clearTimeout(task.timer); task.reject(new Error("otlp worker pool shutting down")); @@ -316,6 +339,10 @@ export function getOtlpWorkerPool( if (!pool) { const resolvedPath = workerPath ?? path.join(process.cwd(), "build", "otlpTransformWorker.cjs"); pool = new OtlpWorkerPool(size, resolvedPath, pricingModels, meter); + // Drain + terminate workers on shutdown so they aren't force-killed mid-task (which would + // churn respawns). The main thread stays the only DB writer, so inserts are unaffected. + signalsEmitter.on("SIGTERM", () => void pool!.shutdown()); + signalsEmitter.on("SIGINT", () => void pool!.shutdown()); } return pool; } diff --git a/apps/webapp/test/otlpWorkerPoolMetrics.test.ts b/apps/webapp/test/otlpWorkerPoolMetrics.test.ts index b15ea509d1..99238064e8 100644 --- a/apps/webapp/test/otlpWorkerPoolMetrics.test.ts +++ b/apps/webapp/test/otlpWorkerPoolMetrics.test.ts @@ -72,4 +72,16 @@ describe("OtlpWorkerPool self-observability", () => { { timeout: 5000, interval: 50 } ); }); + + it("rejects new work once shutdown has started", async () => { + const pool = new OtlpWorkerPool(2, echoWorker, []); + // A task before shutdown resolves normally. + await expect(pool.runTransform("traces", payload(), config)).resolves.toBeDefined(); + + await pool.shutdown(); + + await expect(pool.runTransform("traces", payload(), config)).rejects.toThrow(/shutting down/); + // Shutting down twice is a no-op. + await expect(pool.shutdown()).resolves.toBeUndefined(); + }); }); From 982a17028016b6d9103f80058bfb234c89c0d9fb Mon Sep 17 00:00:00 2001 From: Eric Allam Date: Sat, 11 Jul 2026 12:46:30 +0100 Subject: [PATCH 10/10] refactor(webapp): make the OTLP transform worker pool an HMR-safe singleton Use the shared singleton() helper (stored on globalThis) for the OTLP transform worker pool, like otlpExporter and llmPricingRegistry, so a single pool and its worker threads survive Remix hot-reloads in dev instead of orphaning workers on every reload. Production is unaffected. Also drops the unused broadcastPricingToPool helper. --- apps/webapp/app/v3/otlpWorkerPool.server.ts | 21 +++++++++------------ 1 file changed, 9 insertions(+), 12 deletions(-) diff --git a/apps/webapp/app/v3/otlpWorkerPool.server.ts b/apps/webapp/app/v3/otlpWorkerPool.server.ts index a7844f9174..51bc0bca83 100644 --- a/apps/webapp/app/v3/otlpWorkerPool.server.ts +++ b/apps/webapp/app/v3/otlpWorkerPool.server.ts @@ -9,6 +9,7 @@ import { } from "@internal/tracing"; import { logger } from "~/services/logger.server"; import { signalsEmitter } from "~/services/signals.server"; +import { singleton } from "~/utils/singleton"; export type TransformKind = "traces" | "logs" | "metrics"; @@ -328,25 +329,21 @@ export class OtlpWorkerPool { } } -let pool: OtlpWorkerPool | undefined; - export function getOtlpWorkerPool( size: number, pricingModels: unknown[], workerPath?: string, meter?: Meter ): OtlpWorkerPool { - if (!pool) { + // singleton() stores on globalThis so the pool (and its worker threads) survive Remix HMR in dev + // rather than leaking an orphaned pool + workers on every reload. + return singleton("otlpWorkerPool", () => { const resolvedPath = workerPath ?? path.join(process.cwd(), "build", "otlpTransformWorker.cjs"); - pool = new OtlpWorkerPool(size, resolvedPath, pricingModels, meter); + const created = new OtlpWorkerPool(size, resolvedPath, pricingModels, meter); // Drain + terminate workers on shutdown so they aren't force-killed mid-task (which would // churn respawns). The main thread stays the only DB writer, so inserts are unaffected. - signalsEmitter.on("SIGTERM", () => void pool!.shutdown()); - signalsEmitter.on("SIGINT", () => void pool!.shutdown()); - } - return pool; -} - -export function broadcastPricingToPool(models: unknown[]) { - pool?.broadcastPricing(models); + signalsEmitter.on("SIGTERM", () => void created.shutdown()); + signalsEmitter.on("SIGINT", () => void created.shutdown()); + return created; + }); }