diff --git a/.changeset/real-planets-stare.md b/.changeset/real-planets-stare.md new file mode 100644 index 000000000..6549f06bd --- /dev/null +++ b/.changeset/real-planets-stare.md @@ -0,0 +1,7 @@ +--- +"@trigger.dev/sdk": patch +"trigger.dev": patch +"@trigger.dev/core": patch +--- + +Fixed incorrect span timings around checkpoints by implementing a precise wall clock that resets after restores diff --git a/.env.example b/.env.example index 59e45945e..ab1d09d7b 100644 --- a/.env.example +++ b/.env.example @@ -63,4 +63,5 @@ COORDINATOR_SECRET=coordinator-secret # generate the actual secret with `openssl # These are needed for the object store (for handling large payloads/outputs) # OBJECT_STORE_BASE_URL="https://{bucket}.{accountId}.r2.cloudflarestorage.com" # OBJECT_STORE_ACCESS_KEY_ID= -# OBJECT_STORE_SECRET_ACCESS_KEY= \ No newline at end of file +# OBJECT_STORE_SECRET_ACCESS_KEY= +# RUNTIME_WAIT_THRESHOLD_IN_MS=10000 \ No newline at end of file diff --git a/apps/webapp/app/env.server.ts b/apps/webapp/app/env.server.ts index 96a788b07..5599bc331 100644 --- a/apps/webapp/app/env.server.ts +++ b/apps/webapp/app/env.server.ts @@ -97,13 +97,26 @@ 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"), + RUNTIME_WAIT_THRESHOLD_IN_MS: z.coerce.number().int().default(30000), }); export type Environment = z.infer; diff --git a/apps/webapp/app/v3/environmentVariables/environmentVariablesRepository.server.ts b/apps/webapp/app/v3/environmentVariables/environmentVariablesRepository.server.ts index 54a80a14b..7cf9c9a06 100644 --- a/apps/webapp/app/v3/environmentVariables/environmentVariablesRepository.server.ts +++ b/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 [ @@ -432,6 +473,10 @@ export class EnvironmentVariablesRepository implements Repository { key: "TRIGGER_API_URL", value: env.APP_ORIGIN, }, + { + key: "TRIGGER_RUNTIME_WAIT_THRESHOLD_IN_MS", + value: String(env.RUNTIME_WAIT_THRESHOLD_IN_MS), + }, ]; } diff --git a/apps/webapp/app/v3/marqs.server.ts b/apps/webapp/app/v3/marqs.server.ts index e9be68233..daf66e124 100644 --- a/apps/webapp/app/v3/marqs.server.ts +++ b/apps/webapp/app/v3/marqs.server.ts @@ -591,14 +591,6 @@ export class MarQS { concurrencyLimitKey: string; currentConcurrencyKey: string; }) { - logger.debug("Calling dequeueMessage", { - messageQueue, - parentQueue, - visibilityQueue, - concurrencyLimitKey, - currentConcurrencyKey, - }); - const result = await this.redis.dequeueMessage( messageQueue, parentQueue, diff --git a/packages/cli-v3/src/workers/dev/worker-facade.ts b/packages/cli-v3/src/workers/dev/worker-facade.ts index f2449538c..37fe0c6e4 100644 --- a/packages/cli-v3/src/workers/dev/worker-facade.ts +++ b/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; -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); diff --git a/packages/cli-v3/src/workers/prod/worker-facade.ts b/packages/cli-v3/src/workers/prod/worker-facade.ts index 07c3fadee..21185c3b8 100644 --- a/packages/cli-v3/src/workers/prod/worker-facade.ts +++ b/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); @@ -200,7 +201,9 @@ const zodIpc = new ZodIpcConnection({ }, }); -const prodRuntimeManager = new ProdRuntimeManager(zodIpc); +const prodRuntimeManager = new ProdRuntimeManager(zodIpc, { + waitThresholdInMs: parseInt(process.env.TRIGGER_RUNTIME_WAIT_THRESHOLD_IN_MS ?? "30000", 10), +}); runtime.setGlobalRuntimeManager(prodRuntimeManager); diff --git a/packages/core/src/v3/clock-api.ts b/packages/core/src/v3/clock-api.ts new file mode 100644 index 000000000..9bcbed8ca --- /dev/null +++ b/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(); diff --git a/packages/core/src/v3/clock/clock.ts b/packages/core/src/v3/clock/clock.ts new file mode 100644 index 000000000..2b2229d5e --- /dev/null +++ b/packages/core/src/v3/clock/clock.ts @@ -0,0 +1,6 @@ +export type ClockTime = [number, number]; + +export interface Clock { + preciseNow(): ClockTime; + reset(): void; +} diff --git a/packages/core/src/v3/clock/index.ts b/packages/core/src/v3/clock/index.ts new file mode 100644 index 000000000..b9a941f99 --- /dev/null +++ b/packages/core/src/v3/clock/index.ts @@ -0,0 +1,37 @@ +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(); + } + + public reset(): void { + this.#getClock().reset(); + } + + #getClock(): Clock { + return getGlobal(API_NAME) ?? SIMPLE_CLOCK; + } +} diff --git a/packages/core/src/v3/clock/preciseWallClock.ts b/packages/core/src/v3/clock/preciseWallClock.ts new file mode 100644 index 000000000..e541287af --- /dev/null +++ b/packages/core/src/v3/clock/preciseWallClock.ts @@ -0,0 +1,46 @@ +import { PreciseDate } from "@google-cloud/precise-date"; +import { Clock, ClockTime } from "./clock"; + +export type PreciseWallClockOptions = { + origin?: ClockTime; + now?: PreciseDate; +}; + +export class PreciseWallClock implements Clock { + private _origin: { + clockTime: ClockTime; + preciseDate: PreciseDate; + }; + + get #originClockTime() { + return this._origin.clockTime; + } + + get #originPreciseDate() { + return this._origin.preciseDate; + } + + constructor(options: PreciseWallClockOptions = {}) { + this._origin = { + clockTime: options.origin ?? process.hrtime(), + preciseDate: 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]; + } + + reset() { + this._origin = { + clockTime: process.hrtime(), + preciseDate: new PreciseDate(), + }; + } +} diff --git a/packages/core/src/v3/clock/simpleClock.ts b/packages/core/src/v3/clock/simpleClock.ts new file mode 100644 index 000000000..db0514282 --- /dev/null +++ b/packages/core/src/v3/clock/simpleClock.ts @@ -0,0 +1,15 @@ +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]; + } + + reset() { + // do nothing + } +} diff --git a/packages/core/src/v3/consoleInterceptor.ts b/packages/core/src/v3/consoleInterceptor.ts index 40c9c5e20..3c926d92a 100644 --- a/packages/core/src/v3/consoleInterceptor.ts +++ b/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 { diff --git a/packages/core/src/v3/index.ts b/packages/core/src/v3/index.ts index 62ada3d7a..d3696950a 100644 --- a/packages/core/src/v3/index.ts +++ b/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 { PreciseWallClock as DurableClock } from "./clock/preciseWallClock"; 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, diff --git a/packages/core/src/v3/logger/taskLogger.ts b/packages/core/src/v3/logger/taskLogger.ts index 0cd36582e..84b4c5b20 100644 --- a/packages/core/src/v3/logger/taskLogger.ts +++ b/packages/core/src/v3/logger/taskLogger.ts @@ -4,7 +4,8 @@ import { iconStringForSeverity } from "../icons"; import { SemanticInternalAttributes } from "../semanticInternalAttributes"; import { TriggerTracer } from "../tracer"; import { flattenAttributes } from "../utils/flattenAttributes"; -import { PreciseDateOrigin, calculatePreciseDateHrTime } from "../utils/preciseDate"; +import { ClockTime } from "../clock/clock"; +import { clock } from "../clock-api"; export type LogLevel = "log" | "error" | "warn" | "info" | "debug"; @@ -14,7 +15,6 @@ export type TaskLoggerConfig = { logger: Logger; tracer: TriggerTracer; level: LogLevel; - preciseDateOrigin: PreciseDateOrigin; }; export interface TaskLogger { @@ -36,41 +36,40 @@ export class OtelTaskLogger implements TaskLogger { debug(message: string, properties?: Record) { if (this._level < 4) return; - this.#emitLog(message, "debug", SeverityNumber.DEBUG, properties); + this.#emitLog(message, this.#getTimestampInHrTime(), "debug", SeverityNumber.DEBUG, properties); } log(message: string, properties?: Record) { if (this._level < 2) return; - this.#emitLog(message, "log", SeverityNumber.INFO, properties); + this.#emitLog(message, this.#getTimestampInHrTime(), "log", SeverityNumber.INFO, properties); } info(message: string, properties?: Record) { if (this._level < 3) return; - this.#emitLog(message, "info", SeverityNumber.INFO, properties); + this.#emitLog(message, this.#getTimestampInHrTime(), "info", SeverityNumber.INFO, properties); } warn(message: string, properties?: Record) { if (this._level < 1) return; - this.#emitLog(message, "warn", SeverityNumber.WARN, properties); + this.#emitLog(message, this.#getTimestampInHrTime(), "warn", SeverityNumber.WARN, properties); } error(message: string, properties?: Record) { if (this._level < 0) return; - this.#emitLog(message, "error", SeverityNumber.ERROR, properties); + this.#emitLog(message, this.#getTimestampInHrTime(), "error", SeverityNumber.ERROR, properties); } #emitLog( message: string, + timestamp: ClockTime, severityText: string, severityNumber: SeverityNumber, properties?: Record ) { - const timestamp = this.#getTimestampInHrTime(); - let attributes: Attributes = { ...flattenAttributes(properties) }; const icon = iconStringForSeverity(severityNumber); @@ -83,7 +82,7 @@ export class OtelTaskLogger implements TaskLogger { severityText, body: message, attributes, - timestamp + timestamp, }); } @@ -91,17 +90,17 @@ export class OtelTaskLogger implements TaskLogger { return this._config.tracer.startActiveSpan(name, fn, options); } - #getTimestampInHrTime(): [number, number] { - return calculatePreciseDateHrTime(this._config.preciseDateOrigin); + #getTimestampInHrTime(): ClockTime { + return clock.preciseNow(); } } export class NoopTaskLogger implements TaskLogger { - debug() { } - log() { } - info() { } - warn() { } - error() { } + debug() {} + log() {} + info() {} + warn() {} + error() {} trace(name: string, fn: (span: Span) => Promise): Promise { return fn({} as Span); } diff --git a/packages/core/src/v3/otel/tracingSDK.ts b/packages/core/src/v3/otel/tracingSDK.ts index 611574ba7..ed233042d 100644 --- a/packages/core/src/v3/otel/tracingSDK.ts +++ b/packages/core/src/v3/otel/tracingSDK.ts @@ -15,16 +15,18 @@ import { detectResourcesSync, processDetectorSync, } from "@opentelemetry/resources"; -import { LoggerProvider, SimpleLogRecordProcessor } from "@opentelemetry/sdk-logs"; import { + BatchLogRecordProcessor, + LoggerProvider, + SimpleLogRecordProcessor, +} from "@opentelemetry/sdk-logs"; +import { + BatchSpanProcessor, NodeTracerProvider, SimpleSpanProcessor, SpanExporter, } from "@opentelemetry/sdk-trace-node"; import { SemanticResourceAttributes } from "@opentelemetry/semantic-conventions"; -import { SemanticInternalAttributes } from "../semanticInternalAttributes"; -import { TaskContextLogProcessor, TaskContextSpanProcessor } from "../tasks/taskContextManager"; -import { getEnvVar } from "../utils/getEnv"; import { OTEL_ATTRIBUTE_PER_EVENT_COUNT_LIMIT, OTEL_ATTRIBUTE_PER_LINK_COUNT_LIMIT, @@ -35,6 +37,9 @@ import { OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, OTEL_SPAN_EVENT_COUNT_LIMIT, } from "../limits"; +import { SemanticInternalAttributes } from "../semanticInternalAttributes"; +import { TaskContextLogProcessor, TaskContextSpanProcessor } from "../tasks/taskContextManager"; +import { getEnvVar } from "../utils/getEnv"; class AsyncResourceDetector implements DetectorSync { private _promise: Promise; @@ -130,8 +135,22 @@ export class TracingSDK { }); traceProvider.addSpanProcessor( - new TaskContextSpanProcessor(new SimpleSpanProcessor(spanExporter)) + new TaskContextSpanProcessor( + getEnvVar("OTEL_BATCH_PROCESSING_ENABLED") === "1" + ? new BatchSpanProcessor(spanExporter, { + maxExportBatchSize: parseInt(getEnvVar("OTEL_SPAN_MAX_EXPORT_BATCH_SIZE") ?? "64"), + scheduledDelayMillis: parseInt( + getEnvVar("OTEL_SPAN_SCHEDULED_DELAY_MILLIS") ?? "200" + ), + exportTimeoutMillis: parseInt( + getEnvVar("OTEL_SPAN_EXPORT_TIMEOUT_MILLIS") ?? "30000" + ), + maxQueueSize: parseInt(getEnvVar("OTEL_SPAN_MAX_QUEUE_SIZE") ?? "512"), + }) + : new SimpleSpanProcessor(spanExporter) + ) ); + traceProvider.register(); registerInstrumentations({ @@ -153,7 +172,16 @@ export class TracingSDK { }); loggerProvider.addLogRecordProcessor( - new TaskContextLogProcessor(new SimpleLogRecordProcessor(logExporter)) + new TaskContextLogProcessor( + getEnvVar("OTEL_BATCH_PROCESSING_ENABLED") === "1" + ? new BatchLogRecordProcessor(logExporter, { + maxExportBatchSize: parseInt(getEnvVar("OTEL_LOG_MAX_EXPORT_BATCH_SIZE") ?? "64"), + scheduledDelayMillis: parseInt(getEnvVar("OTEL_LOG_SCHEDULED_DELAY_MILLIS") ?? "200"), + exportTimeoutMillis: parseInt(getEnvVar("OTEL_LOG_EXPORT_TIMEOUT_MILLIS") ?? "30000"), + maxQueueSize: parseInt(getEnvVar("OTEL_LOG_MAX_QUEUE_SIZE") ?? "512"), + }) + : new SimpleLogRecordProcessor(logExporter) + ) ); this._logProvider = loggerProvider; diff --git a/packages/core/src/v3/runtime/devRuntimeManager.ts b/packages/core/src/v3/runtime/devRuntimeManager.ts index a4df98c89..007749639 100644 --- a/packages/core/src/v3/runtime/devRuntimeManager.ts +++ b/packages/core/src/v3/runtime/devRuntimeManager.ts @@ -5,7 +5,6 @@ import { TaskRunExecution, TaskRunExecutionResult, } from "../schemas"; -import { conditionallyImportPacket } from "../utils/ioSerialization"; import { RuntimeManager } from "./manager"; export class DevRuntimeManager implements RuntimeManager { diff --git a/packages/core/src/v3/runtime/prodRuntimeManager.ts b/packages/core/src/v3/runtime/prodRuntimeManager.ts index c7359802d..859f678e7 100644 --- a/packages/core/src/v3/runtime/prodRuntimeManager.ts +++ b/packages/core/src/v3/runtime/prodRuntimeManager.ts @@ -1,3 +1,4 @@ +import { clock } from "../clock-api"; import { BatchTaskRunExecutionResult, ProdChildToWorkerMessages, @@ -9,6 +10,11 @@ import { } from "../schemas"; import { ZodIpcConnection } from "../zodIpc"; import { RuntimeManager } from "./manager"; +import { setTimeout } from "node:timers/promises"; + +export type ProdRuntimeManagerOptions = { + waitThresholdInMs?: number; +}; export class ProdRuntimeManager implements RuntimeManager { _taskWaits: Map< @@ -21,7 +27,7 @@ export class ProdRuntimeManager implements RuntimeManager { { resolve: (value: BatchTaskRunExecutionResult) => void; reject: (err?: any) => void } > = new Map(); - _waitForRestore: { resolve: (value?: any) => void; reject: (err?: any) => void } | undefined; + _waitForRestore: { resolve: (value: "restore") => void; reject: (err?: any) => void } | undefined; _tasks: Map = new Map(); @@ -29,7 +35,8 @@ export class ProdRuntimeManager implements RuntimeManager { private ipc: ZodIpcConnection< typeof ProdWorkerToChildMessages, typeof ProdChildToWorkerMessages - > + >, + private options: ProdRuntimeManagerOptions = {} ) {} disable(): void { @@ -47,20 +54,16 @@ export class ProdRuntimeManager implements RuntimeManager { } async waitForDuration(ms: number): Promise { - let timeout: NodeJS.Timeout | undefined; - const now = Date.now(); - const resolveAfterDuration = new Promise((resolve) => { - timeout = setTimeout(resolve, ms); - }); + const resolveAfterDuration = setTimeout(ms, "duration" as const); - if (ms < 10_000) { + if (ms <= this.waitThresholdInMs) { await resolveAfterDuration; return; } - const waitForRestore = new Promise((resolve, reject) => { + const waitForRestore = new Promise<"restore">((resolve, reject) => { this._waitForRestore = { resolve, reject }; }); @@ -81,8 +84,6 @@ export class ProdRuntimeManager implements RuntimeManager { // The coordinator can then cancel any in-progress checkpoints this.ipc.send("CANCEL_CHECKPOINT", {}); - - clearTimeout(timeout); } resumeAfterRestore(): void { @@ -90,7 +91,10 @@ export class ProdRuntimeManager implements RuntimeManager { return; } - this._waitForRestore.resolve(); + // Resets the clock to the current time + clock.reset(); + + this._waitForRestore.resolve("restore"); this._waitForRestore = undefined; } @@ -155,4 +159,8 @@ export class ProdRuntimeManager implements RuntimeManager { this._taskWaits.delete(execution.run.id); } + + private get waitThresholdInMs(): number { + return this.options.waitThresholdInMs ?? 30_000; + } } diff --git a/packages/core/src/v3/tracer.ts b/packages/core/src/v3/tracer.ts index 4ee4ffd42..c735e7e7f 100644 --- a/packages/core/src/v3/tracer.ts +++ b/packages/core/src/v3/tracer.ts @@ -10,6 +10,7 @@ import { } from "@opentelemetry/api"; import { Logger, logs } from "@opentelemetry/api-logs"; import { SemanticInternalAttributes } from "./semanticInternalAttributes"; +import { clock } from "./clock-api"; export type TriggerTracerConfig = | { @@ -65,6 +66,7 @@ export class TriggerTracer { { ...options, attributes, + startTime: clock.preciseNow(), }, parentContext, async (span) => { @@ -94,7 +96,7 @@ export class TriggerTracer { throw e; } finally { - span.end(); + span.end(clock.preciseNow()); } } ); diff --git a/packages/core/src/v3/utils/globals.ts b/packages/core/src/v3/utils/globals.ts index 4e8fc0757..c768b933c 100644 --- a/packages/core/src/v3/utils/globals.ts +++ b/packages/core/src/v3/utils/globals.ts @@ -1,3 +1,4 @@ +import { Clock } from "../clock/clock"; import type { RuntimeManager } from "../runtime/manager"; import { _globalThis } from "./platform"; @@ -44,4 +45,5 @@ type TriggerDotDevGlobal = { type TriggerDotDevGlobalAPI = { runtime?: RuntimeManager; logger?: any; + clock?: Clock; }; diff --git a/packages/core/src/v3/utils/preciseDate.ts b/packages/core/src/v3/utils/preciseDate.ts deleted file mode 100644 index 3efc92ecf..000000000 --- a/packages/core/src/v3/utils/preciseDate.ts +++ /dev/null @@ -1,23 +0,0 @@ -import { PreciseDate } from "@google-cloud/precise-date"; - -export type PreciseDateOrigin = { - hrtime: [number, number]; - timestamp: PreciseDate -} - -export function preciseDateOriginNow(): PreciseDateOrigin { - return { - hrtime: process.hrtime(), - timestamp: new PreciseDate() - } -} - -export function calculatePreciseDateHrTime(origin: PreciseDateOrigin): [number, number] { - const elapsedHrTime = process.hrtime(origin.hrtime); - const elapsedNanoseconds = BigInt(elapsedHrTime[0]) * BigInt(1e9) + BigInt(elapsedHrTime[1]); - - const preciseDate = new PreciseDate(origin.timestamp.getFullTime() + elapsedNanoseconds) - const dateStruct = preciseDate.toStruct(); - - return [dateStruct.seconds, dateStruct.nanos]; -} \ No newline at end of file diff --git a/packages/trigger-sdk/src/v3/wait.ts b/packages/trigger-sdk/src/v3/wait.ts index c3130af58..56b48a93e 100644 --- a/packages/trigger-sdk/src/v3/wait.ts +++ b/packages/trigger-sdk/src/v3/wait.ts @@ -33,8 +33,6 @@ export const wait = { const durationInMs = calculateDurationInMs(options); await runtime.waitForDuration(durationInMs); - - span.end(start + durationInMs); }, { attributes: { @@ -65,8 +63,6 @@ export const wait = { const durationInMs = options.date.getTime() - start; await runtime.waitForDuration(durationInMs); - - span.end(start + durationInMs); }, { attributes: { diff --git a/references/v3-catalog/src/trigger/simple.ts b/references/v3-catalog/src/trigger/simple.ts index 1acae5c81..2f68f21bf 100644 --- a/references/v3-catalog/src/trigger/simple.ts +++ b/references/v3-catalog/src/trigger/simple.ts @@ -73,7 +73,7 @@ export const parentTask = task({ console.info("This is an info message"); logger.info("This is an info message from logger.info"); - console.log(JSON.stringify({ ctx, message: "This is the parent task context" })); + console.log(JSON.stringify({ ctx, message: "This is the parent task contexts" })); logger.log(JSON.stringify({ ctx, message: "This is the parent task context from logger.log" })); console.warn("You've been warned buddy"); logger.warn("You've been warned buddy from logger.warn"); @@ -107,10 +107,21 @@ export const parentTask = task({ export const childTask = task({ id: "child-task", - run: async (payload: { message: string; forceError: boolean }, { ctx }) => { + run: async ( + payload: { message: string; forceError: boolean; delayInSeconds?: number }, + { ctx } + ) => { logger.info("Child task payload", { payload }); + logger.info("Child task payload 2", { payload }); + logger.info("Child task payload 3", { payload }); + logger.info("Child task payload 4", { payload }); + logger.info("Child task payload 5", { payload }); - await wait.for({ seconds: 10 }); + await wait.for({ seconds: payload.delayInSeconds ?? 5 }); + + logger.info("Child task payload 6", { payload }); + logger.info("Child task payload 7", { payload }); + logger.info("Child task payload 8", { payload }); const response = await fetch("https://jsonhero.io/api/create.json", { method: "POST",