diff --git a/.server-changes/otlp-transform-worker-pool.md b/.server-changes/otlp-transform-worker-pool.md new file mode 100644 index 0000000000..a88d15020c --- /dev/null +++ b/.server-changes/otlp-transform-worker-pool.md @@ -0,0 +1,6 @@ +--- +area: webapp +type: improvement +--- + +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/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/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/llmPricingRegistry.server.ts b/apps/webapp/app/v3/llmPricingRegistry.server.ts index eb186e1521..31baf37059 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,43 @@ 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) { + 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); + }; +} + async function initRegistry(registry: ModelPricingRegistry) { if (env.LLM_PRICING_SEED_ON_STARTUP) { await seedLlmPricing(prisma); @@ -25,16 +63,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 +116,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 65680bbd2a..da4ca4605e 100644 --- a/apps/webapp/app/v3/otlpExporter.server.ts +++ b/apps/webapp/app/v3/otlpExporter.server.ts @@ -2,48 +2,59 @@ 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 { 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"; - -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"; +import os from "node:os"; +import { getOtlpWorkerPool } from "./otlpWorkerPool.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 = env.OTEL_TRANSFORM_WORKER_POOL_ENABLED; + +// 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; verbose: boolean; spanAttributeValueLengthLimit: number; + // Inject in tests; defaults to the global provider. Instruments are no-op when metrics are off. + meter?: Meter; }; class OTLPExporter { @@ -51,70 +62,223 @@ class OTLPExporter { private readonly _clickhouseFactory: ClickhouseFactory; private readonly _verbose: boolean; 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); - } - ); + 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 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 eventsWithStores = this.#filterResourceLogs(request.resourceLogs).flatMap( - (resourceLog) => { - return convertLogsToCreateableEvents(resourceLog, this._spanAttributeValueLengthLimit); - } - ); + const eventCount = await this.#exportEvents(eventsWithStores); - const eventCount = await this.#exportEvents(eventsWithStores); + span.setAttribute("event_count", eventCount); + this._eventsProduced.add(eventCount, { kind: "logs" }); + this.#recordIngest("logs", "inline", "ok", startedAt); - span.setAttribute("event_count", eventCount); + return ExportLogsServiceResponse.create(); + } catch (error) { + this.#recordIngest("logs", "inline", "error", startedAt); + throw error; + } + }); + } - return ExportLogsServiceResponse.create(); + 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 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; + } }); } + async #exportRawEvents(kind: "traces" | "logs", payload: Uint8Array): Promise { + await startSpan( + this._tracer, + kind === "traces" ? "exportTracesRaw" : "exportLogsRaw", + async (span) => { + 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; + } + } + ); + } + + #transformConfig() { + return { + spanAttributeValueLengthLimit: this._spanAttributeValueLengthLimit, + defaultEventStore: env.EVENT_REPOSITORY_DEFAULT_STORE, + }; + } + + async #pool() { + await waitForLlmPricingReady(); + const models = + llmPricingRegistry && llmPricingRegistry.isLoaded ? llmPricingRegistry.toSerializable() : []; + const pool = getOtlpWorkerPool( + OTEL_TRANSFORM_WORKER_POOL_SIZE, + models, + env.OTEL_TRANSFORM_WORKER_PATH, + this._meter + ); + 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( - 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(); @@ -151,7 +315,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}`); @@ -285,923 +449,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..d66f1a8b82 --- /dev/null +++ b/apps/webapp/app/v3/otlpTransform.server.ts @@ -0,0 +1,930 @@ +// 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"; + +// 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, + 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; +} diff --git a/apps/webapp/app/v3/otlpTransformWorker.ts b/apps/webapp/app/v3/otlpTransformWorker.ts new file mode 100644 index 0000000000..a8f3d3dfb5 --- /dev/null +++ b/apps/webapp/app/v3/otlpTransformWorker.ts @@ -0,0 +1,106 @@ +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 { + // 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); + const computeMs = performance.now() - startedAt; + parentPort!.postMessage({ id: task.id, ok: true, result, computeMs }); + } 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..51bc0bca83 --- /dev/null +++ b/apps/webapp/app/v3/otlpWorkerPool.server.ts @@ -0,0 +1,349 @@ +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"; +import { signalsEmitter } from "~/services/signals.server"; +import { singleton } from "~/utils/singleton"; + +export type TransformKind = "traces" | "logs" | "metrics"; + +type TaskMessage = { + id: number; + kind: TransformKind; + payload: Uint8Array; + spanAttributeValueLengthLimit: number; + defaultEventStore: string; +}; + +type Task = { + message: TaskMessage; + transfer: ArrayBuffer[]; + resolve: (r: any) => void; + 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; +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. +export class OtlpWorkerPool { + private readonly workers: Worker[] = []; + private readonly idle: Worker[] = []; + private readonly queue: number[] = []; + private readonly tasks = new Map(); + 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. + 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[], + 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; 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); + } + ); + + worker.on("error", (error) => { + logger.error("OtlpWorkerPool worker error", { error: error.message }); + 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}`), "exit"); + }); + + this.workers.push(worker); + this.idle.push(worker); + } + + // 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, 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); + + 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) { + const task = this.tasks.get(inFlightId); + 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(); + } + + 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); + } + + 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 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); + 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, "timeout"); + } 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.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")); + } + const id = this.nextId++; + return new Promise((resolve, reject) => { + const timer = setTimeout(() => this.onTimeout(id), TASK_TIMEOUT_MS); + this.tasks.set(id, { + 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], + resolve, + reject, + timer, + enqueuedAt: Date.now(), + }); + this.queue.push(id); + this.drain(); + }); + } + + broadcastPricing(models: unknown[]) { + this.latestPricingModels = models; + 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() { + return this.queue.length; + } + + // 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")); + } + this.tasks.clear(); + await Promise.all(workers.map((worker) => worker.terminate().catch(() => {}))); + } +} + +export function getOtlpWorkerPool( + size: number, + pricingModels: unknown[], + workerPath?: string, + meter?: Meter +): OtlpWorkerPool { + // 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"); + 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 created.shutdown()); + signalsEmitter.on("SIGINT", () => void created.shutdown()); + return created; + }); +} 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/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..99238064e8 --- /dev/null +++ b/apps/webapp/test/otlpWorkerPoolMetrics.test.ts @@ -0,0 +1,87 @@ +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 } + ); + }); + + 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(); + }); +}); 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();