runtime manager debug logs

Author: nicktrn

packages/cli-v3/src/entryPoints/managed/execution.ts

@@ -144,6 +144,10 @@ export class RunExecution {
       }
     });
 
+    taskRunProcess.onSendDebugLog.attach(async (debugLog) => {
+      this.sendRuntimeDebugLog(debugLog.message, debugLog.properties);
+    });
+
     return taskRunProcess;
   }
 
@@ -951,6 +955,26 @@ export class RunExecution {
     });
   }
 
+  private sendRuntimeDebugLog(
+    message: string,
+    properties?: SendDebugLogOptions["properties"],
+    runIdOverride?: string
+  ) {
+    this.logger.sendDebugLog({
+      runId: runIdOverride ?? this.runFriendlyId,
+      message: `[runtime] ${message}`,
+      print: false,
+      properties: {
+        ...properties,
+        runId: this.runFriendlyId,
+        snapshotId: this.currentSnapshotId,
+        executionId: this.id,
+        executionRestoreCount: this.restoreCount,
+        lastHeartbeat: this.lastHeartbeat?.toISOString(),
+      },
+    });
+  }
+
   // Ensure we can only set this once
   private set runFriendlyId(id: string) {
     if (this._runFriendlyId) {

packages/cli-v3/src/entryPoints/managed/logger.ts

@@ -9,6 +9,7 @@ export type SendDebugLogOptions = {
   message: string;
   date?: Date;
   properties?: WorkloadDebugLogRequestBody["properties"];
+  print?: boolean;
 };
 
 export type RunLoggerOptions = {
@@ -25,7 +26,7 @@ export class RunLogger {
     this.env = opts.env;
   }
 
-  sendDebugLog({ runId, message, date, properties }: SendDebugLogOptions) {
+  sendDebugLog({ runId, message, date, properties, print = true }: SendDebugLogOptions) {
     if (!runId) {
       runId = this.env.TRIGGER_RUN_ID;
     }
@@ -41,7 +42,9 @@ export class RunLogger {
       workerName: this.env.TRIGGER_WORKER_INSTANCE_NAME,
     };
 
-    console.log(message, mergedProperties);
+    if (print) {
+      console.log(message, mergedProperties);
+    }
 
     this.httpClient.sendDebugLog(runId, {
       message,

packages/cli-v3/src/executions/taskRunProcess.ts

@@ -33,6 +33,11 @@ import {
   SuspendedProcessError,
 } from "@trigger.dev/core/v3/errors";
 
+export type OnSendDebugLogMessage = InferSocketMessageSchema<
+  typeof ExecutorToWorkerMessageCatalog,
+  "SEND_DEBUG_LOG"
+>;
+
 export type TaskRunProcessOptions = {
   workerManifest: WorkerManifest;
   serverWorker: ServerBackgroundWorker;
@@ -68,6 +73,7 @@ export class TaskRunProcess {
   public onExit: Evt<{ code: number | null; signal: NodeJS.Signals | null; pid?: number }> =
     new Evt();
   public onIsBeingKilled: Evt<TaskRunProcess> = new Evt();
+  public onSendDebugLog: Evt<OnSendDebugLogMessage> = new Evt();
 
   private _isPreparedForNextRun: boolean = false;
   private _isPreparedForNextAttempt: boolean = false;
@@ -178,6 +184,9 @@ export class TaskRunProcess {
         UNCAUGHT_EXCEPTION: async (message) => {
           logger.debug("uncaught exception in task run process", { ...message });
         },
+        SEND_DEBUG_LOG: async (message) => {
+          this.onSendDebugLog.post(message);
+        },
       },
     });
 

packages/core/src/v3/runEngineWorker/supervisor/schemas.ts

@@ -126,7 +126,7 @@ export type WorkerApiDequeueFromVersionResponseBody = z.infer<
   typeof WorkerApiDequeueFromVersionResponseBody
 >;
 
-const AttributeValue = z.union([
+export const DebugLogPropertiesValue = z.union([
   z.string(),
   z.number(),
   z.boolean(),
@@ -135,12 +135,13 @@ const AttributeValue = z.union([
   z.array(z.boolean().nullable()),
 ]);
 
-const Attributes = z.record(z.string(), AttributeValue.optional());
+export const DebugLogProperties = z.record(z.string(), DebugLogPropertiesValue.optional());
+export type DebugLogProperties = z.infer<typeof DebugLogProperties>;
 
 export const WorkerApiDebugLogBody = z.object({
   time: z.coerce.date(),
   message: z.string(),
-  properties: Attributes.optional(),
+  properties: DebugLogProperties.optional(),
 });
 export type WorkerApiDebugLogBody = z.infer<typeof WorkerApiDebugLogBody>;
 

packages/core/src/v3/runtime/sharedRuntimeManager.ts

@@ -1,6 +1,7 @@
 import { assertExhaustive } from "../../utils.js";
 import { clock } from "../clock-api.js";
 import { lifecycleHooks } from "../lifecycle-hooks-api.js";
+import { DebugLogProperties } from "../runEngineWorker/index.js";
 import {
   BatchTaskRunExecutionResult,
   CompletedWaitpoint,
@@ -176,7 +177,10 @@ export class SharedRuntimeManager implements RuntimeManager {
     switch (waitpoint.type) {
       case "RUN": {
         if (!waitpoint.completedByTaskRun) {
-          this.log("No completedByTaskRun for RUN waitpoint", waitpoint);
+          this.debugLog(
+            "No completedByTaskRun for RUN waitpoint",
+            this.waitpointForDebugLog(waitpoint)
+          );
           return null;
         }
 
@@ -192,7 +196,10 @@ export class SharedRuntimeManager implements RuntimeManager {
       }
       case "BATCH": {
         if (!waitpoint.completedByBatch) {
-          this.log("No completedByBatch for BATCH waitpoint", waitpoint);
+          this.debugLog(
+            "No completedByBatch for BATCH waitpoint",
+            this.waitpointForDebugLog(waitpoint)
+          );
           return null;
         }
 
@@ -213,18 +220,19 @@ export class SharedRuntimeManager implements RuntimeManager {
   }
 
   private resolveWaitpoint(waitpoint: CompletedWaitpoint, resolverId?: ResolverId | null): void {
+    // This is spammy, don't make this a debug log
     this.log("resolveWaitpoint", waitpoint);
 
     if (waitpoint.type === "BATCH") {
       // We currently ignore these, they're not required to resume after a batch completes
-      this.log("Ignoring BATCH waitpoint", waitpoint);
+      this.debugLog("Ignoring BATCH waitpoint", this.waitpointForDebugLog(waitpoint));
       return;
     }
 
     resolverId = resolverId ?? this.resolverIdFromWaitpoint(waitpoint);
 
     if (!resolverId) {
-      this.log("No resolverId for waitpoint", { ...this.status, ...waitpoint });
+      this.debugLog("No resolverId for waitpoint", this.waitpointForDebugLog(waitpoint));
 
       // No need to store the waitpoint, we'll never be able to resolve it
       return;
@@ -233,7 +241,7 @@ export class SharedRuntimeManager implements RuntimeManager {
     const resolve = this.resolversById.get(resolverId);
 
     if (!resolve) {
-      this.log("No resolver found for resolverId", { ...this.status, resolverId });
+      this.debugLog("No resolver found for resolverId", { ...this.status, resolverId });
 
       // Store the waitpoint for later if we can't find a resolver
       this.waitpointsByResolverId.set(resolverId, waitpoint);
@@ -280,6 +288,42 @@ export class SharedRuntimeManager implements RuntimeManager {
     }
   }
 
+  private waitpointForDebugLog(waitpoint: CompletedWaitpoint): DebugLogProperties {
+    const {
+      completedByTaskRun,
+      completedByBatch,
+      completedAfter,
+      completedAt,
+      output,
+      id,
+      ...rest
+    } = waitpoint;
+
+    return {
+      ...rest,
+      waitpointId: id,
+      output: output?.slice(0, 100),
+      completedByTaskRunId: completedByTaskRun?.id,
+      completedByTaskRunBatchId: completedByTaskRun?.batch?.id,
+      completedByBatchId: completedByBatch?.id,
+      completedAfter: completedAfter?.toISOString(),
+      completedAt: completedAt?.toISOString(),
+    };
+  }
+
+  private debugLog(message: string, properties?: DebugLogProperties) {
+    const status = this.status;
+
+    if (this.showLogs) {
+      console.log(`[${new Date().toISOString()}] ${message}`, { ...status, ...properties });
+    }
+
+    this.ipc.send("SEND_DEBUG_LOG", {
+      message,
+      properties: { ...status, ...properties },
+    });
+  }
+
   private log(message: string, ...args: any[]) {
     if (!this.showLogs) return;
     console.log(`[${new Date().toISOString()}] ${message}`, args);

packages/core/src/v3/schemas/messages.ts

@@ -18,6 +18,7 @@ import {
   WaitReason,
 } from "./schemas.js";
 import { CompletedWaitpoint } from "./runEngine.js";
+import { DebugLogProperties } from "../runEngineWorker/index.js";
 
 export const AckCallbackResult = z.discriminatedUnion("success", [
   z.object({
@@ -174,6 +175,13 @@ export const ExecutorToWorkerMessageCatalog = {
   UNCAUGHT_EXCEPTION: {
     message: UncaughtExceptionMessage,
   },
+  SEND_DEBUG_LOG: {
+    message: z.object({
+      version: z.literal("v1").default("v1"),
+      message: z.string(),
+      properties: DebugLogProperties.optional(),
+    }),
+  },
 };
 
 export const WorkerToExecutorMessageCatalog = {