keep sending otel metrics in between runs, dev now acts more like prod, decouple flushing metrics with metric bucket intervals

Author: ericallam

apps/webapp/app/env.server.ts

@@ -384,6 +384,7 @@ const EnvironmentSchema = z
     DEV_OTEL_LOG_EXPORT_TIMEOUT_MILLIS: z.string().default("30000"),
     DEV_OTEL_LOG_MAX_QUEUE_SIZE: z.string().default("512"),
     DEV_OTEL_METRICS_EXPORT_INTERVAL_MILLIS: z.string().optional(),
+    DEV_OTEL_METRICS_COLLECTION_INTERVAL_MILLIS: z.string().optional(),
 
     PROD_OTEL_BATCH_PROCESSING_ENABLED: z.string().default("0"),
     PROD_OTEL_SPAN_MAX_EXPORT_BATCH_SIZE: z.string().default("64"),
@@ -395,6 +396,7 @@ const EnvironmentSchema = z
     PROD_OTEL_LOG_EXPORT_TIMEOUT_MILLIS: z.string().default("30000"),
     PROD_OTEL_LOG_MAX_QUEUE_SIZE: z.string().default("512"),
     PROD_OTEL_METRICS_EXPORT_INTERVAL_MILLIS: z.string().optional(),
+    PROD_OTEL_METRICS_COLLECTION_INTERVAL_MILLIS: z.string().optional(),
 
     TRIGGER_OTEL_SPAN_ATTRIBUTE_COUNT_LIMIT: z.string().default("1024"),
     TRIGGER_OTEL_LOG_ATTRIBUTE_COUNT_LIMIT: z.string().default("1024"),

apps/webapp/app/v3/environmentVariables/environmentVariablesRepository.server.ts

@@ -976,6 +976,13 @@ async function resolveBuiltInDevVariables(runtimeEnvironment: RuntimeEnvironment
     );
   }
 
+  if (env.DEV_OTEL_METRICS_COLLECTION_INTERVAL_MILLIS) {
+    result.push({
+      key: "TRIGGER_OTEL_METRICS_COLLECTION_INTERVAL_MILLIS",
+      value: env.DEV_OTEL_METRICS_COLLECTION_INTERVAL_MILLIS,
+    });
+  }
+
   if (env.DEV_OTEL_BATCH_PROCESSING_ENABLED === "1") {
     result = result.concat([
       {
@@ -1120,6 +1127,13 @@ async function resolveBuiltInProdVariables(
     );
   }
 
+  if (env.PROD_OTEL_METRICS_COLLECTION_INTERVAL_MILLIS) {
+    result.push({
+      key: "TRIGGER_OTEL_METRICS_COLLECTION_INTERVAL_MILLIS",
+      value: env.PROD_OTEL_METRICS_COLLECTION_INTERVAL_MILLIS,
+    });
+  }
+
   if (env.PROD_OTEL_BATCH_PROCESSING_ENABLED === "1") {
     result = result.concat([
       {

packages/cli-v3/src/dev/taskRunProcessPool.ts

@@ -24,6 +24,8 @@ export class TaskRunProcessPool {
   private readonly maxExecutionsPerProcess: number;
   private readonly executionCountsPerProcess: Map<number, number> = new Map();
   private readonly deprecatedVersions: Set<string> = new Set();
+  private readonly idleTimers: Map<TaskRunProcess, NodeJS.Timeout> = new Map();
+  private static readonly IDLE_TIMEOUT_MS = 30_000;
 
   constructor(options: TaskRunProcessPoolOptions) {
     this.options = options;
@@ -39,6 +41,7 @@ export class TaskRunProcessPool {
     const versionProcesses = this.availableProcessesByVersion.get(version) || [];
 
     const processesToKill = versionProcesses.filter((process) => !process.isExecuting());
+    processesToKill.forEach((process) => this.clearIdleTimer(process));
     Promise.all(processesToKill.map((process) => this.killProcess(process))).then(() => {
       this.availableProcessesByVersion.delete(version);
     });
@@ -72,6 +75,7 @@ export class TaskRunProcessPool {
           version,
           availableProcesses.filter((p) => p !== reusableProcess)
         );
+        this.clearIdleTimer(reusableProcess);
 
         if (!this.busyProcessesByVersion.has(version)) {
           this.busyProcessesByVersion.set(version, new Set());
@@ -156,6 +160,7 @@ export class TaskRunProcessPool {
           this.availableProcessesByVersion.set(version, []);
         }
         this.availableProcessesByVersion.get(version)!.push(process);
+        this.startIdleTimer(process, version);
       } catch (error) {
         logger.debug("[TaskRunProcessPool] Failed to cleanup process for reuse, killing it", {
           error,
@@ -215,7 +220,42 @@ export class TaskRunProcessPool {
     return process.isHealthy;
   }
 
+  private startIdleTimer(process: TaskRunProcess, version: string): void {
+    this.clearIdleTimer(process);
+
+    const timer = setTimeout(() => {
+      // Synchronously remove from available pool before async kill to prevent race with getProcess()
+      const available = this.availableProcessesByVersion.get(version);
+      if (available) {
+        const index = available.indexOf(process);
+        if (index !== -1) {
+          available.splice(index, 1);
+        }
+      }
+      this.idleTimers.delete(process);
+
+      logger.debug("[TaskRunProcessPool] Idle timeout reached, killing process", {
+        pid: process.pid,
+        version,
+      });
+
+      this.killProcess(process);
+    }, TaskRunProcessPool.IDLE_TIMEOUT_MS);
+
+    this.idleTimers.set(process, timer);
+  }
+
+  private clearIdleTimer(process: TaskRunProcess): void {
+    const timer = this.idleTimers.get(process);
+    if (timer) {
+      clearTimeout(timer);
+      this.idleTimers.delete(process);
+    }
+  }
+
   private async killProcess(process: TaskRunProcess): Promise<void> {
+    this.clearIdleTimer(process);
+
     if (!process.isHealthy) {
       logger.debug("[TaskRunProcessPool] Process is not healthy, skipping cleanup", {
         processId: process.pid,
@@ -247,6 +287,12 @@ export class TaskRunProcessPool {
       versions: Array.from(this.availableProcessesByVersion.keys()),
     });
 
+    // Clear all idle timers
+    for (const timer of this.idleTimers.values()) {
+      clearTimeout(timer);
+    }
+    this.idleTimers.clear();
+
     // Kill all available processes across all versions
     const allAvailableProcesses = Array.from(this.availableProcessesByVersion.values()).flat();
     await Promise.all(allAvailableProcesses.map((process) => this.killProcess(process)));

packages/core/src/v3/otel/tracingSDK.ts

@@ -55,6 +55,7 @@ import {
 import { SemanticInternalAttributes } from "../semanticInternalAttributes.js";
 import { taskContext } from "../task-context-api.js";
 import {
+  BufferingMetricExporter,
   TaskContextLogProcessor,
   TaskContextMetricExporter,
   TaskContextSpanProcessor,
@@ -285,14 +286,22 @@ export class TracingSDK {
       url: metricsUrl,
       timeoutMillis: config.forceFlushTimeoutMillis,
     });
-    const metricExporter = new TaskContextMetricExporter(rawMetricExporter);
+
+    const collectionIntervalMs = parseInt(
+      getEnvVar("TRIGGER_OTEL_METRICS_COLLECTION_INTERVAL_MILLIS") ?? "10000"
+    );
+    const exportIntervalMs = parseInt(
+      getEnvVar("TRIGGER_OTEL_METRICS_EXPORT_INTERVAL_MILLIS") ?? "30000"
+    );
+
+    // Chain: PeriodicReader(10s) → TaskContextMetricExporter → BufferingMetricExporter(30s) → OTLP
+    const bufferingExporter = new BufferingMetricExporter(rawMetricExporter, exportIntervalMs);
+    const metricExporter = new TaskContextMetricExporter(bufferingExporter);
 
     const metricReaders: MetricReader[] = [
       new PeriodicExportingMetricReader({
         exporter: metricExporter,
-        exportIntervalMillis: parseInt(
-          getEnvVar("TRIGGER_OTEL_METRICS_EXPORT_INTERVAL_MILLIS") ?? "60000"
-        ),
+        exportIntervalMillis: collectionIntervalMs,
         exportTimeoutMillis: parseInt(
           getEnvVar("TRIGGER_OTEL_METRICS_EXPORT_TIMEOUT_MILLIS") ?? "30000"
         ),

packages/core/src/v3/taskContext/index.ts

@@ -1,13 +1,14 @@
 import { Attributes } from "@opentelemetry/api";
 import { ServerBackgroundWorker, TaskRunContext } from "../schemas/index.js";
 import { SemanticInternalAttributes } from "../semanticInternalAttributes.js";
-import { getGlobal, registerGlobal, unregisterGlobal } from "../utils/globals.js";
+import { getGlobal, registerGlobal } from "../utils/globals.js";
 import { TaskContext } from "./types.js";
 
 const API_NAME = "task-context";
 
 export class TaskContextAPI {
   private static _instance?: TaskContextAPI;
+  private _runDisabled = false;
 
   private constructor() {}
 
@@ -23,6 +24,10 @@ export class TaskContextAPI {
     return this.#getTaskContext() !== undefined;
   }
 
+  get isRunDisabled(): boolean {
+    return this._runDisabled;
+  }
+
   get ctx(): TaskRunContext | undefined {
     return this.#getTaskContext()?.ctx;
   }
@@ -98,11 +103,12 @@ export class TaskContextAPI {
   }
 
   public disable() {
-    unregisterGlobal(API_NAME);
+    this._runDisabled = true;
   }
 
   public setGlobalTaskContext(taskContext: TaskContext): boolean {
-    return registerGlobal(API_NAME, taskContext);
+    this._runDisabled = false;
+    return registerGlobal(API_NAME, taskContext, true);
   }
 
   #getTaskContext(): TaskContext | undefined {

packages/core/src/v3/taskContext/otelProcessors.ts

@@ -8,6 +8,7 @@ import type {
   MetricData,
   PushMetricExporter,
   ResourceMetrics,
+  ScopeMetrics,
 } from "@opentelemetry/sdk-metrics";
 import { Span, SpanProcessor } from "@opentelemetry/sdk-trace-base";
 import { SemanticInternalAttributes } from "../semanticInternalAttributes.js";
@@ -130,29 +131,44 @@ export class TaskContextMetricExporter implements PushMetricExporter {
 
   export(metrics: ResourceMetrics, resultCallback: (result: ExportResult) => void): void {
     if (!taskContext.ctx) {
-      // No active run — drop metrics (between-run noise)
+      // No context at all — drop metrics
       resultCallback({ code: ExportResultCode.SUCCESS });
       return;
     }
 
     const ctx = taskContext.ctx;
-    const contextAttrs: Attributes = {
-      [SemanticInternalAttributes.RUN_ID]: ctx.run.id,
-      [SemanticInternalAttributes.TASK_SLUG]: ctx.task.id,
-      [SemanticInternalAttributes.ATTEMPT_NUMBER]: ctx.attempt.number,
-      [SemanticInternalAttributes.ENVIRONMENT_ID]: ctx.environment.id,
-      [SemanticInternalAttributes.ORGANIZATION_ID]: ctx.organization.id,
-      [SemanticInternalAttributes.PROJECT_ID]: ctx.project.id,
-      [SemanticInternalAttributes.MACHINE_PRESET_NAME]: ctx.machine?.name,
-      [SemanticInternalAttributes.ENVIRONMENT_TYPE]: ctx.environment.type,
-    };
+
+    let contextAttrs: Attributes;
+
+    if (taskContext.isRunDisabled) {
+      // Between runs: keep environment/project/org/machine attrs, strip run-specific ones
+      contextAttrs = {
+        [SemanticInternalAttributes.ENVIRONMENT_ID]: ctx.environment.id,
+        [SemanticInternalAttributes.ENVIRONMENT_TYPE]: ctx.environment.type,
+        [SemanticInternalAttributes.ORGANIZATION_ID]: ctx.organization.id,
+        [SemanticInternalAttributes.PROJECT_ID]: ctx.project.id,
+        [SemanticInternalAttributes.MACHINE_PRESET_NAME]: ctx.machine?.name,
+      };
+    } else {
+      // During a run: full context attrs
+      contextAttrs = {
+        [SemanticInternalAttributes.RUN_ID]: ctx.run.id,
+        [SemanticInternalAttributes.TASK_SLUG]: ctx.task.id,
+        [SemanticInternalAttributes.ATTEMPT_NUMBER]: ctx.attempt.number,
+        [SemanticInternalAttributes.ENVIRONMENT_ID]: ctx.environment.id,
+        [SemanticInternalAttributes.ORGANIZATION_ID]: ctx.organization.id,
+        [SemanticInternalAttributes.PROJECT_ID]: ctx.project.id,
+        [SemanticInternalAttributes.MACHINE_PRESET_NAME]: ctx.machine?.name,
+        [SemanticInternalAttributes.ENVIRONMENT_TYPE]: ctx.environment.type,
+      };
+    }
 
     if (taskContext.worker) {
       contextAttrs[SemanticInternalAttributes.WORKER_ID] = taskContext.worker.id;
       contextAttrs[SemanticInternalAttributes.WORKER_VERSION] = taskContext.worker.version;
     }
 
-    if (ctx.run.tags?.length) {
+    if (!taskContext.isRunDisabled && ctx.run.tags?.length) {
       contextAttrs[SemanticInternalAttributes.RUN_TAGS] = ctx.run.tags;
     }
 
@@ -184,3 +200,108 @@ export class TaskContextMetricExporter implements PushMetricExporter {
     return this._innerExporter.shutdown();
   }
 }
+
+export class BufferingMetricExporter implements PushMetricExporter {
+  selectAggregationTemporality?: (instrumentType: InstrumentType) => AggregationTemporality;
+  selectAggregation?: (instrumentType: InstrumentType) => AggregationOption;
+
+  private _buffer: ResourceMetrics[] = [];
+  private _lastFlushTime = Date.now();
+
+  constructor(
+    private _innerExporter: PushMetricExporter,
+    private _flushIntervalMs: number
+  ) {
+    if (_innerExporter.selectAggregationTemporality) {
+      this.selectAggregationTemporality =
+        _innerExporter.selectAggregationTemporality.bind(_innerExporter);
+    }
+    if (_innerExporter.selectAggregation) {
+      this.selectAggregation = _innerExporter.selectAggregation.bind(_innerExporter);
+    }
+  }
+
+  export(metrics: ResourceMetrics, resultCallback: (result: ExportResult) => void): void {
+    this._buffer.push(metrics);
+
+    const now = Date.now();
+    if (now - this._lastFlushTime >= this._flushIntervalMs) {
+      this._lastFlushTime = now;
+      const merged = this._mergeBuffer();
+      this._innerExporter.export(merged, resultCallback);
+    } else {
+      resultCallback({ code: ExportResultCode.SUCCESS });
+    }
+  }
+
+  forceFlush(): Promise<void> {
+    if (this._buffer.length > 0) {
+      this._lastFlushTime = Date.now();
+      const merged = this._mergeBuffer();
+      return new Promise<void>((resolve, reject) => {
+        this._innerExporter.export(merged, (result) => {
+          if (result.code === ExportResultCode.SUCCESS) {
+            resolve();
+          } else {
+            reject(result.error ?? new Error("Export failed"));
+          }
+        });
+      }).then(() => this._innerExporter.forceFlush());
+    }
+    return this._innerExporter.forceFlush();
+  }
+
+  shutdown(): Promise<void> {
+    return this.forceFlush().then(() => this._innerExporter.shutdown());
+  }
+
+  private _mergeBuffer(): ResourceMetrics {
+    const batch = this._buffer;
+    this._buffer = [];
+
+    if (batch.length === 1) {
+      return batch[0]!;
+    }
+
+    const base = batch[0]!;
+
+    // Merge all scopeMetrics by scope name, then metrics by descriptor name
+    const scopeMap = new Map<string, { scope: ScopeMetrics["scope"]; metricsMap: Map<string, MetricData> }>();
+
+    for (const rm of batch) {
+      for (const sm of rm.scopeMetrics) {
+        const scopeKey = sm.scope.name;
+        let scopeEntry = scopeMap.get(scopeKey);
+        if (!scopeEntry) {
+          scopeEntry = { scope: sm.scope, metricsMap: new Map() };
+          scopeMap.set(scopeKey, scopeEntry);
+        }
+
+        for (const metric of sm.metrics) {
+          const metricKey = metric.descriptor.name;
+          const existing = scopeEntry.metricsMap.get(metricKey);
+          if (existing) {
+            // Append data points from this collection to the existing metric
+            scopeEntry.metricsMap.set(metricKey, {
+              ...existing,
+              dataPoints: [...existing.dataPoints, ...metric.dataPoints],
+            } as MetricData);
+          } else {
+            scopeEntry.metricsMap.set(metricKey, {
+              ...metric,
+              dataPoints: [...metric.dataPoints],
+            } as MetricData);
+          }
+        }
+      }
+    }
+
+    return {
+      resource: base.resource,
+      scopeMetrics: Array.from(scopeMap.values()).map(({ scope, metricsMap }) => ({
+        scope,
+        metrics: Array.from(metricsMap.values()),
+      })),
+    };
+  }
+}