durable clock WIP

Author: ericallam

apps/webapp/app/env.server.ts

@@ -97,13 +97,25 @@ const EnvironmentSchema = z.object({
   CONTAINER_REGISTRY_USERNAME: z.string().optional(),
   CONTAINER_REGISTRY_PASSWORD: z.string().optional(),
   DEPLOY_REGISTRY_HOST: z.string().optional(),
-  DEV_OTEL_EXPORTER_OTLP_ENDPOINT: z.string().optional(),
   OBJECT_STORE_BASE_URL: z.string().optional(),
   OBJECT_STORE_ACCESS_KEY_ID: z.string().optional(),
   OBJECT_STORE_SECRET_ACCESS_KEY: z.string().optional(),
   EVENTS_BATCH_SIZE: z.coerce.number().int().default(100),
   EVENTS_BATCH_INTERVAL: z.coerce.number().int().default(1000),
   EVENTS_DEFAULT_LOG_RETENTION: z.coerce.number().int().default(7),
+
+  // Development OTEL environment variables
+  DEV_OTEL_EXPORTER_OTLP_ENDPOINT: z.string().optional(),
+  // If this is set to 1, then the below variables are used to configure the batch processor for spans and logs
+  DEV_OTEL_BATCH_PROCESSING_ENABLED: z.string().default("0"),
+  DEV_OTEL_SPAN_MAX_EXPORT_BATCH_SIZE: z.string().default("64"),
+  DEV_OTEL_SPAN_SCHEDULED_DELAY_MILLIS: z.string().default("200"),
+  DEV_OTEL_SPAN_EXPORT_TIMEOUT_MILLIS: z.string().default("30000"),
+  DEV_OTEL_SPAN_MAX_QUEUE_SIZE: z.string().default("512"),
+  DEV_OTEL_LOG_MAX_EXPORT_BATCH_SIZE: z.string().default("64"),
+  DEV_OTEL_LOG_SCHEDULED_DELAY_MILLIS: z.string().default("200"),
+  DEV_OTEL_LOG_EXPORT_TIMEOUT_MILLIS: z.string().default("30000"),
+  DEV_OTEL_LOG_MAX_QUEUE_SIZE: z.string().default("512"),
 });
 
 export type Environment = z.infer<typeof EnvironmentSchema>;

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

@@ -30,7 +30,7 @@ function parseSecretKey(key: string) {
 const SecretValue = z.object({ secret: z.string() });
 
 export class EnvironmentVariablesRepository implements Repository {
-  constructor(private prismaClient: PrismaClient = prisma) { }
+  constructor(private prismaClient: PrismaClient = prisma) {}
 
   async create(
     projectId: string,
@@ -419,8 +419,49 @@ export class EnvironmentVariablesRepository implements Repository {
         {
           key: "OTEL_EXPORTER_OTLP_ENDPOINT",
           value: env.DEV_OTEL_EXPORTER_OTLP_ENDPOINT ?? env.APP_ORIGIN,
-        }
-      ];
+        },
+      ].concat(
+        env.DEV_OTEL_BATCH_PROCESSING_ENABLED === "1"
+          ? [
+              {
+                key: "OTEL_BATCH_PROCESSING_ENABLED",
+                value: "1",
+              },
+              {
+                key: "OTEL_SPAN_MAX_EXPORT_BATCH_SIZE",
+                value: env.DEV_OTEL_SPAN_MAX_EXPORT_BATCH_SIZE,
+              },
+              {
+                key: "OTEL_SPAN_SCHEDULED_DELAY_MILLIS",
+                value: env.DEV_OTEL_SPAN_SCHEDULED_DELAY_MILLIS,
+              },
+              {
+                key: "OTEL_SPAN_EXPORT_TIMEOUT_MILLIS",
+                value: env.DEV_OTEL_SPAN_EXPORT_TIMEOUT_MILLIS,
+              },
+              {
+                key: "OTEL_SPAN_MAX_QUEUE_SIZE",
+                value: env.DEV_OTEL_SPAN_MAX_QUEUE_SIZE,
+              },
+              {
+                key: "OTEL_LOG_MAX_EXPORT_BATCH_SIZE",
+                value: env.DEV_OTEL_LOG_MAX_EXPORT_BATCH_SIZE,
+              },
+              {
+                key: "OTEL_LOG_SCHEDULED_DELAY_MILLIS",
+                value: env.DEV_OTEL_LOG_SCHEDULED_DELAY_MILLIS,
+              },
+              {
+                key: "OTEL_LOG_EXPORT_TIMEOUT_MILLIS",
+                value: env.DEV_OTEL_LOG_EXPORT_TIMEOUT_MILLIS,
+              },
+              {
+                key: "OTEL_LOG_MAX_QUEUE_SIZE",
+                value: env.DEV_OTEL_LOG_MAX_QUEUE_SIZE,
+              },
+            ]
+          : []
+      );
     }
 
     return [

packages/cli-v3/src/workers/dev/worker-facade.ts

@@ -2,9 +2,10 @@ import {
   Config,
   ProjectConfig,
   TaskExecutor,
-  preciseDateOriginNow,
   type TracingSDK,
   type HandleErrorFunction,
+  DurableClock,
+  clock,
 } from "@trigger.dev/core/v3";
 
 __WORKER_SETUP__;
@@ -42,10 +43,11 @@ import { TaskMetadataWithFunctions } from "../../types.js";
 
 declare const sender: ZodMessageSender<typeof childToWorkerMessages>;
 
-const preciseDateOrigin = preciseDateOriginNow();
+const durableClock = new DurableClock();
+clock.setGlobalClock(durableClock);
 
 const tracer = new TriggerTracer({ tracer: otelTracer, logger: otelLogger });
-const consoleInterceptor = new ConsoleInterceptor(otelLogger, preciseDateOrigin);
+const consoleInterceptor = new ConsoleInterceptor(otelLogger);
 
 const devRuntimeManager = new DevRuntimeManager();
 
@@ -55,7 +57,6 @@ const otelTaskLogger = new OtelTaskLogger({
   logger: otelLogger,
   tracer: tracer,
   level: "info",
-  preciseDateOrigin,
 });
 
 logger.setGlobalTaskLogger(otelTaskLogger);

packages/cli-v3/src/workers/prod/worker-facade.ts

@@ -6,8 +6,9 @@ import {
   TaskExecutor,
   ZodIpcConnection,
   type TracingSDK,
-  preciseDateOriginNow,
   HandleErrorFunction,
+  DurableClock,
+  clock,
 } from "@trigger.dev/core/v3";
 import "source-map-support/register.js";
 
@@ -40,16 +41,16 @@ import * as packageJson from "../../../package.json";
 
 import { TaskMetadataWithFunctions } from "../../types";
 
-const preciseDateOrigin = preciseDateOriginNow();
+const durableClock = new DurableClock();
+clock.setGlobalClock(durableClock);
 
 const tracer = new TriggerTracer({ tracer: otelTracer, logger: otelLogger });
-const consoleInterceptor = new ConsoleInterceptor(otelLogger, preciseDateOrigin);
+const consoleInterceptor = new ConsoleInterceptor(otelLogger);
 
 const otelTaskLogger = new OtelTaskLogger({
   logger: otelLogger,
   tracer: tracer,
   level: "info",
-  preciseDateOrigin,
 });
 
 logger.setGlobalTaskLogger(otelTaskLogger);

packages/core/src/v3/clock-api.ts

@@ -0,0 +1,5 @@
+// Split module-level variable definition into separate files to allow
+// tree-shaking on each api instance.
+import { ClockAPI } from "./clock";
+/** Entrypoint for clock API */
+export const clock = ClockAPI.getInstance();

packages/core/src/v3/clock/clock.ts

@@ -0,0 +1,5 @@
+export type ClockTime = [number, number];
+
+export interface Clock {
+  preciseNow(): ClockTime;
+}

packages/core/src/v3/clock/durableClock.ts

@@ -0,0 +1,27 @@
+import { PreciseDate } from "@google-cloud/precise-date";
+import { Clock, ClockTime } from "./clock";
+
+export type DurableClockOptions = {
+  origin?: ClockTime;
+  now?: PreciseDate;
+};
+
+export class DurableClock implements Clock {
+  private _originClockTime: ClockTime;
+  private _originPreciseDate: PreciseDate;
+
+  constructor(options: DurableClockOptions = {}) {
+    this._originClockTime = options.origin ?? process.hrtime();
+    this._originPreciseDate = options.now ?? new PreciseDate();
+  }
+
+  preciseNow(): [number, number] {
+    const elapsedHrTime = process.hrtime(this._originClockTime);
+    const elapsedNanoseconds = BigInt(elapsedHrTime[0]) * BigInt(1e9) + BigInt(elapsedHrTime[1]);
+
+    const preciseDate = new PreciseDate(this._originPreciseDate.getFullTime() + elapsedNanoseconds);
+    const dateStruct = preciseDate.toStruct();
+
+    return [dateStruct.seconds, dateStruct.nanos];
+  }
+}

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

@@ -0,0 +1,33 @@
+const API_NAME = "clock";
+
+import { getGlobal, registerGlobal } from "../utils/globals";
+import type { Clock, ClockTime } from "./clock";
+import { SimpleClock } from "./simpleClock";
+
+const SIMPLE_CLOCK = new SimpleClock();
+
+export class ClockAPI {
+  private static _instance?: ClockAPI;
+
+  private constructor() {}
+
+  public static getInstance(): ClockAPI {
+    if (!this._instance) {
+      this._instance = new ClockAPI();
+    }
+
+    return this._instance;
+  }
+
+  public setGlobalClock(clock: Clock): boolean {
+    return registerGlobal(API_NAME, clock);
+  }
+
+  public preciseNow(): ClockTime {
+    return this.#getClock().preciseNow();
+  }
+
+  #getClock(): Clock {
+    return getGlobal(API_NAME) ?? SIMPLE_CLOCK;
+  }
+}

packages/core/src/v3/clock/simpleClock.ts

@@ -0,0 +1,11 @@
+import { PreciseDate } from "@google-cloud/precise-date";
+import { Clock } from "./clock";
+
+export class SimpleClock implements Clock {
+  preciseNow(): [number, number] {
+    const now = new PreciseDate();
+    const nowStruct = now.toStruct();
+
+    return [nowStruct.seconds, nowStruct.nanos];
+  }
+}

packages/core/src/v3/consoleInterceptor.ts

@@ -4,11 +4,11 @@ import util from "node:util";
 import { iconStringForSeverity } from "./icons";
 import { SemanticInternalAttributes } from "./semanticInternalAttributes";
 import { flattenAttributes } from "./utils/flattenAttributes";
-import { type PreciseDateOrigin, calculatePreciseDateHrTime } from "./utils/preciseDate";
-
+import { ClockTime } from "./clock/clock";
+import { clock } from "./clock-api";
 
 export class ConsoleInterceptor {
-  constructor(private readonly logger: logsAPI.Logger, private readonly preciseDateOrigin: PreciseDateOrigin) { }
+  constructor(private readonly logger: logsAPI.Logger) {}
 
   // Intercept the console and send logs to the OpenTelemetry logger
   // during the execution of the callback
@@ -39,24 +39,28 @@ export class ConsoleInterceptor {
   }
 
   log(...args: unknown[]): void {
-    this.#handleLog(SeverityNumber.INFO, "Log", ...args);
+    this.#handleLog(SeverityNumber.INFO, this.#getTimestampInHrTime(), "Log", ...args);
   }
 
   info(...args: unknown[]): void {
-    this.#handleLog(SeverityNumber.INFO, "Info", ...args);
+    this.#handleLog(SeverityNumber.INFO, this.#getTimestampInHrTime(), "Info", ...args);
   }
 
   warn(...args: unknown[]): void {
-    this.#handleLog(SeverityNumber.WARN, "Warn", ...args);
+    this.#handleLog(SeverityNumber.WARN, this.#getTimestampInHrTime(), "Warn", ...args);
   }
 
   error(...args: unknown[]): void {
-    this.#handleLog(SeverityNumber.ERROR, "Error", ...args);
+    this.#handleLog(SeverityNumber.ERROR, this.#getTimestampInHrTime(), "Error", ...args);
   }
 
-  #handleLog(severityNumber: SeverityNumber, severityText: string, ...args: unknown[]): void {
+  #handleLog(
+    severityNumber: SeverityNumber,
+    timestamp: ClockTime,
+    severityText: string,
+    ...args: unknown[]
+  ): void {
     const body = util.format(...args);
-    const timestamp = this.#getTimestampInHrTime();
 
     const parsed = tryParseJSON(body);
 
@@ -81,8 +85,8 @@ export class ConsoleInterceptor {
     });
   }
 
-  #getTimestampInHrTime(): [number, number] {
-    return calculatePreciseDateHrTime(this.preciseDateOrigin);
+  #getTimestampInHrTime(): ClockTime {
+    return clock.preciseNow();
   }
 
   #getAttributes(severityNumber: SeverityNumber): logsAPI.LogAttributes {

packages/core/src/v3/index.ts

@@ -9,6 +9,7 @@ export * from "./zodIpc";
 export * from "./errors";
 export * from "./runtime-api";
 export * from "./logger-api";
+export * from "./clock-api";
 export * from "./types";
 export * from "./limits";
 export { SemanticInternalAttributes } from "./semanticInternalAttributes";
@@ -35,6 +36,7 @@ export { taskContextManager, TaskContextSpanProcessor } from "./tasks/taskContex
 export type { RuntimeManager } from "./runtime/manager";
 export { DevRuntimeManager } from "./runtime/devRuntimeManager";
 export { ProdRuntimeManager } from "./runtime/prodRuntimeManager";
+export { DurableClock } from "./clock/durableClock";
 export { TriggerTracer } from "./tracer";
 
 export type { TaskLogger } from "./logger/taskLogger";
@@ -57,11 +59,6 @@ export { omit } from "./utils/omit";
 export { TracingSDK, type TracingDiagnosticLogLevel, recordSpanException } from "./otel";
 export { TaskExecutor, type TaskExecutorOptions } from "./workers/taskExecutor";
 export { detectDependencyVersion } from "./utils/detectDependencyVersion";
-export {
-  type PreciseDateOrigin,
-  calculatePreciseDateHrTime,
-  preciseDateOriginNow,
-} from "./utils/preciseDate";
 export {
   parsePacket,
   stringifyIO,