feat(webapp): add ingest.* metrics for the telemetry ingestion pipeline

Instruments the OTLP ingest path, the transform worker pool, and the shared flush scheduler with OpenTelemetry metrics under a single ingest.* prefix: request and byte throughput with per-signal produced counts, worker task vs compute durations plus queue depth and worker health, and flush batch size, duration, queue depth and dropped-batch counters.

Reuses the existing MeterProvider and is gated by the existing INTERNAL_OTEL_METRIC_EXPORTER_ENABLED, so every instrument is a no-op when metrics are disabled. Recording happens in bulk once per request or flush (never per span) with pull-based gauges, so it stays off the hot path.

Author: ericallam

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<T> = {
   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<T> {
@@ -54,7 +60,21 @@ export class DynamicFlushScheduler<T> {
 
   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<T>) {
+    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<T> {
     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<T> {
 
       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<T> {
             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<T> {
           this.logger.error("Error flushing batch", {
             error: flushError,
           });
+          this._batchesCounter?.add(1, this._batchFailedAttrs);
         }
       })
     );

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,14 +127,22 @@ 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;
     this._config = config;
     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 ?? "")

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),