diff --git a/apps/webapp/app/v3/otlpExporter.server.ts b/apps/webapp/app/v3/otlpExporter.server.ts index bcfefdabb..56424daad 100644 --- a/apps/webapp/app/v3/otlpExporter.server.ts +++ b/apps/webapp/app/v3/otlpExporter.server.ts @@ -183,7 +183,7 @@ class OTLPExporter { function convertLogsToCreateableEvents(resourceLog: ResourceLogs): Array { const resourceAttributes = resourceLog.resource?.attributes ?? []; - const resourceProperties = extractResourceProperties(resourceAttributes); + const resourceProperties = extractEventProperties(resourceAttributes); return resourceLog.scopeLogs.flatMap((scopeLog) => { return scopeLog.logRecords @@ -194,6 +194,11 @@ function convertLogsToCreateableEvents(resourceLog: ResourceLogs): Array { const resourceAttributes = resourceSpan.resource?.attributes ?? []; - const resourceProperties = extractResourceProperties(resourceAttributes); + const resourceProperties = extractEventProperties( + resourceAttributes, + SemanticInternalAttributes.METADATA + ); return resourceSpan.scopeSpans.flatMap((scopeSpan) => { return scopeSpan.spans @@ -269,6 +300,11 @@ function convertSpansToCreateableEvents(resourceSpan: ResourceSpans): 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 attribute = attributes.find((attribute) => attribute.key === name); + 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): number | undefined; -function extractNumberAttribute(attributes: KeyValue[], name: string, fallback: number): number; function extractNumberAttribute( attributes: KeyValue[], - name: string, + 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 attribute = attributes.find((attribute) => attribute.key === name); + 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 extractDoubleAttribute(attributes: KeyValue[], name: string): number | undefined; -function extractDoubleAttribute(attributes: KeyValue[], name: string, fallback: number): number; function extractDoubleAttribute( attributes: KeyValue[], - name: string, + name: string | Array +): number | undefined; +function extractDoubleAttribute( + attributes: KeyValue[], + name: string | Array, + fallback: number +): number; +function extractDoubleAttribute( + attributes: KeyValue[], + name: string | Array, fallback?: number ): number | undefined { - const attribute = attributes.find((attribute) => attribute.key === name); + 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; } -function extractBooleanAttribute(attributes: KeyValue[], name: string): boolean | undefined; -function extractBooleanAttribute(attributes: KeyValue[], name: string, fallback: boolean): boolean; function extractBooleanAttribute( attributes: KeyValue[], - name: string, + name: string | Array +): boolean | undefined; +function extractBooleanAttribute( + attributes: KeyValue[], + name: string | Array, + fallback: boolean +): boolean; +function extractBooleanAttribute( + attributes: KeyValue[], + name: string | Array, fallback?: boolean ): boolean | undefined { - const attribute = attributes.find((attribute) => attribute.key === name); + const key = Array.isArray(name) ? name.filter(Boolean).join(".") : name; + + const attribute = attributes.find((attribute) => attribute.key === key); if (!attribute) return fallback; diff --git a/packages/cli-v3/src/dev/devSupervisor.ts b/packages/cli-v3/src/dev/devSupervisor.ts index 677999dca..6624b85ae 100644 --- a/packages/cli-v3/src/dev/devSupervisor.ts +++ b/packages/cli-v3/src/dev/devSupervisor.ts @@ -25,6 +25,7 @@ import { import pLimit from "p-limit"; import { resolveLocalEnvVars } from "../utilities/localEnvVars.js"; import type { Metafile } from "esbuild"; +import { TaskRunProcessPool } from "./taskRunProcessPool.js"; export type WorkerRuntimeOptions = { name: string | undefined; @@ -67,6 +68,7 @@ class DevSupervisor implements WorkerRuntime { private socketConnections = new Set(); private runLimiter?: ReturnType; + private taskRunProcessPool?: TaskRunProcessPool; constructor(public readonly options: WorkerRuntimeOptions) {} @@ -95,6 +97,31 @@ class DevSupervisor implements WorkerRuntime { this.runLimiter = pLimit(maxConcurrentRuns); + // Initialize the task run process pool + const env = await this.#getEnvVars(); + + const enableProcessReuse = + typeof this.options.config.experimental_processKeepAlive === "boolean" + ? this.options.config.experimental_processKeepAlive + : false; + + if (enableProcessReuse) { + logger.debug("[DevSupervisor] Enabling process reuse", { + enableProcessReuse, + }); + } + + this.taskRunProcessPool = new TaskRunProcessPool({ + env, + cwd: this.options.config.workingDir, + enableProcessReuse: + typeof this.options.config.experimental_processKeepAlive === "boolean" + ? this.options.config.experimental_processKeepAlive + : false, + maxPoolSize: 3, + maxExecutionsPerProcess: 50, + }); + this.socket = this.#createSocket(); //start an SSE connection for presence @@ -111,6 +138,11 @@ class DevSupervisor implements WorkerRuntime { } catch (error) { logger.debug("[DevSupervisor] shutdown, socket failed to close", { error }); } + + // Shutdown the task run process pool + if (this.taskRunProcessPool) { + await this.taskRunProcessPool.shutdown(); + } } async initializeWorker( @@ -293,12 +325,21 @@ class DevSupervisor implements WorkerRuntime { continue; } + if (!this.taskRunProcessPool) { + logger.debug(`[DevSupervisor] dequeueRuns. No task run process pool`, { + run: message.run.friendlyId, + worker, + }); + continue; + } + //new run runController = new DevRunController({ runFriendlyId: message.run.friendlyId, worker: worker, httpClient: this.options.client, logLevel: this.options.args.logLevel, + taskRunProcessPool: this.taskRunProcessPool, onFinished: () => { logger.debug("[DevSupervisor] Run finished", { runId: message.run.friendlyId }); diff --git a/packages/cli-v3/src/dev/taskRunProcessPool.ts b/packages/cli-v3/src/dev/taskRunProcessPool.ts new file mode 100644 index 000000000..d2cee09c6 --- /dev/null +++ b/packages/cli-v3/src/dev/taskRunProcessPool.ts @@ -0,0 +1,165 @@ +import { + MachinePresetResources, + ServerBackgroundWorker, + WorkerManifest, +} from "@trigger.dev/core/v3"; +import { TaskRunProcess } from "../executions/taskRunProcess.js"; +import { logger } from "../utilities/logger.js"; + +export type TaskRunProcessPoolOptions = { + env: Record; + cwd: string; + enableProcessReuse: boolean; + maxPoolSize?: number; + maxExecutionsPerProcess?: number; +}; + +export class TaskRunProcessPool { + private availableProcesses: TaskRunProcess[] = []; + private busyProcesses: Set = new Set(); + private readonly options: TaskRunProcessPoolOptions; + private readonly maxPoolSize: number; + private readonly maxExecutionsPerProcess: number; + + constructor(options: TaskRunProcessPoolOptions) { + this.options = options; + this.maxPoolSize = options.maxPoolSize ?? 3; + this.maxExecutionsPerProcess = options.maxExecutionsPerProcess ?? 50; + } + + async getProcess( + workerManifest: WorkerManifest, + serverWorker: ServerBackgroundWorker, + machineResources: MachinePresetResources, + env?: Record + ): Promise { + // Try to reuse an existing process if enabled + if (this.options.enableProcessReuse) { + const reusableProcess = this.findReusableProcess(); + if (reusableProcess) { + logger.debug("[TaskRunProcessPool] Reusing existing process", { + availableCount: this.availableProcesses.length, + busyCount: this.busyProcesses.size, + }); + + this.availableProcesses = this.availableProcesses.filter((p) => p !== reusableProcess); + this.busyProcesses.add(reusableProcess); + return reusableProcess; + } else { + logger.debug("[TaskRunProcessPool] No reusable process found", { + availableCount: this.availableProcesses.length, + busyCount: this.busyProcesses.size, + }); + } + } + + // Create new process + logger.debug("[TaskRunProcessPool] Creating new process", { + availableCount: this.availableProcesses.length, + busyCount: this.busyProcesses.size, + }); + + const newProcess = new TaskRunProcess({ + workerManifest, + env: { + ...this.options.env, + ...env, + }, + serverWorker, + machineResources, + cwd: this.options.cwd, + }).initialize(); + + this.busyProcesses.add(newProcess); + return newProcess; + } + + async returnProcess(process: TaskRunProcess): Promise { + this.busyProcesses.delete(process); + + if (this.shouldReuseProcess(process)) { + logger.debug("[TaskRunProcessPool] Returning process to pool", { + availableCount: this.availableProcesses.length, + busyCount: this.busyProcesses.size, + }); + + // Clean up but don't kill the process + try { + await process.cleanup(false); + this.availableProcesses.push(process); + } catch (error) { + logger.debug("[TaskRunProcessPool] Failed to cleanup process for reuse, killing it", { + error, + }); + await this.killProcess(process); + } + } else { + logger.debug("[TaskRunProcessPool] Killing process", { + availableCount: this.availableProcesses.length, + busyCount: this.busyProcesses.size, + }); + await this.killProcess(process); + } + } + + private findReusableProcess(): TaskRunProcess | undefined { + return this.availableProcesses.find((process) => this.isProcessHealthy(process)); + } + + private shouldReuseProcess(process: TaskRunProcess): boolean { + const isHealthy = this.isProcessHealthy(process); + const isBeingKilled = process.isBeingKilled; + const pid = process.pid; + + logger.debug("[TaskRunProcessPool] Checking if process should be reused", { + isHealthy, + isBeingKilled, + pid, + availableCount: this.availableProcesses.length, + busyCount: this.busyProcesses.size, + maxPoolSize: this.maxPoolSize, + }); + + return ( + this.options.enableProcessReuse && + this.isProcessHealthy(process) && + this.availableProcesses.length < this.maxPoolSize + ); + } + + private isProcessHealthy(process: TaskRunProcess): boolean { + // Basic health checks - we can expand this later + return !process.isBeingKilled && process.pid !== undefined; + } + + private async killProcess(process: TaskRunProcess): Promise { + try { + await process.cleanup(true); + } catch (error) { + logger.debug("[TaskRunProcessPool] Error killing process", { error }); + } + } + + async shutdown(): Promise { + logger.debug("[TaskRunProcessPool] Shutting down pool", { + availableCount: this.availableProcesses.length, + busyCount: this.busyProcesses.size, + }); + + // Kill all available processes + await Promise.all(this.availableProcesses.map((process) => this.killProcess(process))); + this.availableProcesses = []; + + // Kill all busy processes + await Promise.all(Array.from(this.busyProcesses).map((process) => this.killProcess(process))); + this.busyProcesses.clear(); + } + + getStats() { + return { + availableCount: this.availableProcesses.length, + busyCount: this.busyProcesses.size, + totalCount: this.availableProcesses.length + this.busyProcesses.size, + }; + } +} diff --git a/packages/cli-v3/src/entryPoints/dev-run-controller.ts b/packages/cli-v3/src/entryPoints/dev-run-controller.ts index 2f7ebc83f..a572558f8 100644 --- a/packages/cli-v3/src/entryPoints/dev-run-controller.ts +++ b/packages/cli-v3/src/entryPoints/dev-run-controller.ts @@ -19,6 +19,7 @@ import { sanitizeEnvVars } from "../utilities/sanitizeEnvVars.js"; import { join } from "node:path"; import { BackgroundWorker } from "../dev/backgroundWorker.js"; import { eventBus } from "../utilities/eventBus.js"; +import { TaskRunProcessPool } from "../dev/taskRunProcessPool.js"; type DevRunControllerOptions = { runFriendlyId: string; @@ -26,6 +27,7 @@ type DevRunControllerOptions = { httpClient: CliApiClient; logLevel: LogLevel; heartbeatIntervalSeconds?: number; + taskRunProcessPool: TaskRunProcessPool; onSubscribeToRunNotifications: (run: Run, snapshot: Snapshot) => void; onUnsubscribeFromRunNotifications: (run: Run, snapshot: Snapshot) => void; onFinished: () => void; @@ -605,25 +607,25 @@ export class DevRunController { this.snapshotPoller.start(); - this.taskRunProcess = new TaskRunProcess({ - workerManifest: this.opts.worker.manifest, - env: { - ...sanitizeEnvVars(envVars ?? {}), - ...sanitizeEnvVars(this.opts.worker.params.env), - TRIGGER_WORKER_MANIFEST_PATH: join(this.opts.worker.build.outputPath, "index.json"), - RUN_WORKER_SHOW_LOGS: this.opts.logLevel === "debug" ? "true" : "false", - TRIGGER_PROJECT_REF: execution.project.ref, - }, - serverWorker: { + // Get process from pool instead of creating new one + this.taskRunProcess = await this.opts.taskRunProcessPool.getProcess( + this.opts.worker.manifest, + { id: "unmanaged", contentHash: this.opts.worker.build.contentHash, version: this.opts.worker.serverWorker?.version, engine: "V2", }, - machineResources: execution.machine, - }).initialize(); + execution.machine, + { + TRIGGER_WORKER_MANIFEST_PATH: join(this.opts.worker.build.outputPath, "index.json"), + RUN_WORKER_SHOW_LOGS: this.opts.logLevel === "debug" ? "true" : "false", + } + ); - logger.debug("executing task run process", { + // Update the process environment for this specific run + // Note: We may need to enhance TaskRunProcess to support updating env vars + logger.debug("executing task run process from pool", { attemptNumber: execution.attempt.number, runId: execution.run.id, }); @@ -635,15 +637,21 @@ export class DevRunController { metrics, }, messageId: run.friendlyId, + env: { + ...sanitizeEnvVars(envVars ?? {}), + ...sanitizeEnvVars(this.opts.worker.params.env), + TRIGGER_PROJECT_REF: execution.project.ref, + }, }); logger.debug("Completed run", completion); + // Return process to pool instead of killing it try { - await this.taskRunProcess.cleanup(true); + await this.opts.taskRunProcessPool.returnProcess(this.taskRunProcess); this.taskRunProcess = undefined; } catch (error) { - logger.debug("Failed to cleanup task run process, submitting completion anyway", { + logger.debug("Failed to return task run process to pool, submitting completion anyway", { error, }); } @@ -758,11 +766,14 @@ export class DevRunController { } private async runFinished() { - // Kill the run process - try { - await this.taskRunProcess?.kill("SIGKILL"); - } catch (error) { - logger.debug("Failed to kill task run process", { error }); + // Return the process to the pool instead of killing it directly + if (this.taskRunProcess) { + try { + await this.opts.taskRunProcessPool.returnProcess(this.taskRunProcess); + this.taskRunProcess = undefined; + } catch (error) { + logger.debug("Failed to return task run process to pool during runFinished", { error }); + } } this.runHeartbeat.stop(); @@ -794,9 +805,10 @@ export class DevRunController { if (this.taskRunProcess && !this.taskRunProcess.isBeingKilled) { try { - await this.taskRunProcess.cleanup(true); + await this.opts.taskRunProcessPool.returnProcess(this.taskRunProcess); + this.taskRunProcess = undefined; } catch (error) { - logger.debug("Failed to cleanup task run process", { error }); + logger.debug("Failed to return task run process to pool during stop", { error }); } } diff --git a/packages/core/src/v3/config.ts b/packages/core/src/v3/config.ts index 2c828df68..8063e517e 100644 --- a/packages/core/src/v3/config.ts +++ b/packages/core/src/v3/config.ts @@ -234,6 +234,14 @@ export type TriggerConfig = { env?: Record; }; + /** + * @default false + * @description Keep the process alive after the task has finished running so the next task doesn't have to wait for the process to start up again. + * + * Note that the process could be killed at any time, and we don't make any guarantees about the process being alive for a certain amount of time + */ + experimental_processKeepAlive?: boolean; + /** * @deprecated Use `dirs` instead */ diff --git a/packages/core/src/v3/otel/tracingSDK.ts b/packages/core/src/v3/otel/tracingSDK.ts index 20a79b3b1..bfa7d3a4f 100644 --- a/packages/core/src/v3/otel/tracingSDK.ts +++ b/packages/core/src/v3/otel/tracingSDK.ts @@ -114,7 +114,7 @@ export class TracingSDK { : {}; const commonResources = detectResourcesSync({ - detectors: [this.asyncResourceDetector, processDetectorSync], + detectors: [processDetectorSync], }) .merge( new Resource({ @@ -123,6 +123,8 @@ export class TracingSDK { getEnvVar("OTEL_SERVICE_NAME") ?? "trigger.dev", [SemanticInternalAttributes.TRIGGER]: true, [SemanticInternalAttributes.CLI_VERSION]: VERSION, + [SemanticInternalAttributes.SDK_VERSION]: VERSION, + [SemanticInternalAttributes.SDK_LANGUAGE]: "typescript", }) ) .merge(config.resource ?? new Resource({})) diff --git a/packages/core/src/v3/taskContext/otelProcessors.ts b/packages/core/src/v3/taskContext/otelProcessors.ts index ff96b34cc..2bc986c5e 100644 --- a/packages/core/src/v3/taskContext/otelProcessors.ts +++ b/packages/core/src/v3/taskContext/otelProcessors.ts @@ -21,6 +21,7 @@ export class TaskContextSpanProcessor implements SpanProcessor { span.setAttributes( flattenAttributes( { + ...taskContext.attributes, [SemanticInternalAttributes.ATTEMPT_ID]: taskContext.ctx.attempt.id, [SemanticInternalAttributes.ATTEMPT_NUMBER]: taskContext.ctx.attempt.number, }, @@ -109,6 +110,7 @@ export class TaskContextLogProcessor implements LogRecordProcessor { logRecord.setAttributes( flattenAttributes( { + ...taskContext.attributes, [SemanticInternalAttributes.ATTEMPT_ID]: taskContext.ctx.attempt.id, [SemanticInternalAttributes.ATTEMPT_NUMBER]: taskContext.ctx.attempt.number, }, diff --git a/packages/core/src/v3/workers/taskExecutor.ts b/packages/core/src/v3/workers/taskExecutor.ts index 3e8f64ab5..64b4fe10b 100644 --- a/packages/core/src/v3/workers/taskExecutor.ts +++ b/packages/core/src/v3/workers/taskExecutor.ts @@ -112,11 +112,9 @@ export class TaskExecutor { runMetadata.enterWithMetadata(execution.run.metadata); } - this._tracingSDK.asyncResourceDetector.resolveWithAttributes({ - ...taskContext.attributes, - [SemanticInternalAttributes.SDK_VERSION]: VERSION, - [SemanticInternalAttributes.SDK_LANGUAGE]: "typescript", - }); + // this._tracingSDK.asyncResourceDetector.resolveWithAttributes({ + // ...taskContext.attributes, + // }); const result = await this._tracer.startActiveSpan( attemptMessage, diff --git a/references/hello-world/trigger.config.ts b/references/hello-world/trigger.config.ts index c3c6aea9e..1664c0d00 100644 --- a/references/hello-world/trigger.config.ts +++ b/references/hello-world/trigger.config.ts @@ -4,6 +4,7 @@ import { syncEnvVars } from "@trigger.dev/build/extensions/core"; export default defineConfig({ compatibilityFlags: ["run_engine_v2"], project: "proj_rrkpdguyagvsoktglnod", + experimental_processKeepAlive: true, logLevel: "log", maxDuration: 3600, retries: {