structured logs for all things coordinator

Author: nicktrn

apps/coordinator/src/checkpointer.ts

@@ -8,6 +8,7 @@ import { Buildah, Crictl, Exec } from "./exec";
 import { setTimeout } from "node:timers/promises";
 import { TempFileCleaner } from "./cleaner";
 import { numFromEnv, boolFromEnv } from "./util";
+import { SimpleStructuredLogger } from "@trigger.dev/core/v3/utils/structuredLogger";
 
 type CheckpointerInitializeReturn = {
   canCheckpoint: boolean;
@@ -86,7 +87,7 @@ export class Checkpointer {
   #canCheckpoint = false;
   #dockerMode: boolean;
 
-  #logger = new SimpleLogger("[checkptr]");
+  #logger = new SimpleStructuredLogger("checkpointer");
   #abortControllers = new Map<string, AbortController>();
   #failedCheckpoints = new Map<string, unknown>();
   #waitingForRetry = new Set<string>();
@@ -137,7 +138,7 @@ export class Checkpointer {
         return this.#getInitReturn(true);
       }
 
-      this.#logger.error(testCheckpoint.message, testCheckpoint.error ?? "");
+      this.#logger.error(testCheckpoint.message, { error: testCheckpoint.error });
       return this.#getInitReturn(false);
     }
 

apps/coordinator/src/exec.ts

@@ -1,4 +1,4 @@
-import { SimpleLogger } from "@trigger.dev/core/v3/apps";
+import { SimpleStructuredLogger } from "@trigger.dev/core/v3/utils/structuredLogger";
 import { randomUUID } from "crypto";
 import { homedir } from "os";
 import { type Result, x } from "tinyexec";
@@ -18,23 +18,23 @@ class TinyResult {
 }
 
 interface ExecOptions {
-  logger?: SimpleLogger;
+  logger?: SimpleStructuredLogger;
   abortSignal?: AbortSignal;
   logOutput?: boolean;
   trimArgs?: boolean;
   neverThrow?: boolean;
 }
 
 export class Exec {
-  private logger: SimpleLogger;
+  private logger: SimpleStructuredLogger;
   private abortSignal: AbortSignal | undefined;
 
   private logOutput: boolean;
   private trimArgs: boolean;
   private neverThrow: boolean;
 
   constructor(opts: ExecOptions) {
-    this.logger = opts.logger ?? new SimpleLogger();
+    this.logger = opts.logger ?? new SimpleStructuredLogger("exec");
     this.abortSignal = opts.abortSignal;
 
     this.logOutput = opts.logOutput ?? true;
@@ -103,15 +103,15 @@ interface BuildahOptions {
 
 export class Buildah {
   private id: string;
-  private logger: SimpleLogger;
+  private logger: SimpleStructuredLogger;
   private exec: Exec;
 
   private containers = new Set<string>();
   private images = new Set<string>();
 
   constructor(opts: BuildahOptions) {
     this.id = opts.id ?? randomUUID();
-    this.logger = new SimpleLogger(`[buildah][${this.id}]`);
+    this.logger = new SimpleStructuredLogger("buildah", undefined, { id: this.id });
 
     this.exec = new Exec({
       logger: this.logger,
@@ -220,14 +220,14 @@ interface CrictlOptions {
 
 export class Crictl {
   private id: string;
-  private logger: SimpleLogger;
+  private logger: SimpleStructuredLogger;
   private exec: Exec;
 
   private archives = new Set<string>();
 
   constructor(opts: CrictlOptions) {
     this.id = opts.id ?? randomUUID();
-    this.logger = new SimpleLogger(`[crictl][${this.id}]`);
+    this.logger = new SimpleStructuredLogger("crictl", undefined, { id: this.id });
 
     this.exec = new Exec({
       logger: this.logger,

apps/coordinator/src/index.ts

@@ -11,12 +11,12 @@ import {
 import { ZodNamespace } from "@trigger.dev/core/v3/zodNamespace";
 import { ZodSocketConnection } from "@trigger.dev/core/v3/zodSocket";
 import { HttpReply, getTextBody } from "@trigger.dev/core/v3/apps";
-import { SimpleLogger } from "@trigger.dev/core/v3/apps";
 import { ChaosMonkey } from "./chaosMonkey";
 import { Checkpointer } from "./checkpointer";
 import { boolFromEnv, numFromEnv } from "./util";
 
 import { collectDefaultMetrics, register, Gauge } from "prom-client";
+import { SimpleStructuredLogger } from "@trigger.dev/core/v3/utils/structuredLogger";
 collectDefaultMetrics();
 
 const HTTP_SERVER_PORT = Number(process.env.HTTP_SERVER_PORT || 8020);
@@ -29,7 +29,7 @@ const PLATFORM_WS_PORT = process.env.PLATFORM_WS_PORT || 3030;
 const PLATFORM_SECRET = process.env.PLATFORM_SECRET || "coordinator-secret";
 const SECURE_CONNECTION = ["1", "true"].includes(process.env.SECURE_CONNECTION ?? "false");
 
-const logger = new SimpleLogger(`[${NODE_NAME}]`);
+const logger = new SimpleStructuredLogger(NODE_NAME);
 const chaosMonkey = new ChaosMonkey(
   !!process.env.CHAOS_MONKEY_ENABLED,
   !!process.env.CHAOS_MONKEY_DISABLE_ERRORS,
@@ -335,12 +335,22 @@ class TaskCoordinator {
         next();
       },
       onConnection: async (socket, handler, sender) => {
-        const logger = new SimpleLogger(`[prod-worker][${socket.id}]`);
+        const logger = new SimpleStructuredLogger("prod-worker", undefined, {
+          socketId: socket.id,
+        });
 
         const getAttemptNumber = () => {
           return socket.data.attemptNumber ? parseInt(socket.data.attemptNumber) : undefined;
         };
 
+        const exitRun = () => {
+          logger.log("exitRun", { runId: socket.data.runId });
+
+          socket.emit("REQUEST_EXIT", {
+            version: "v1",
+          });
+        };
+
         const crashRun = async (error: { name: string; message: string; stack?: string }) => {
           try {
             this.#platformSocket?.send("RUN_CRASHED", {
@@ -349,9 +359,7 @@ class TaskCoordinator {
               error,
             });
           } finally {
-            socket.emit("REQUEST_EXIT", {
-              version: "v1",
-            });
+            exitRun();
           }
         };
 
@@ -669,9 +677,7 @@ class TaskCoordinator {
           });
 
           if (!checkpoint.docker || !willSimulate) {
-            socket.emit("REQUEST_EXIT", {
-              version: "v1",
-            });
+            exitRun();
           }
         });
 
@@ -687,9 +693,7 @@ class TaskCoordinator {
             completion,
           });
 
-          socket.emit("REQUEST_EXIT", {
-            version: "v1",
-          });
+          exitRun();
         });
 
         // MARK: CHECKPOINT
@@ -784,9 +788,7 @@ class TaskCoordinator {
           }
 
           if (!checkpoint.docker || !willSimulate) {
-            socket.emit("REQUEST_EXIT", {
-              version: "v1",
-            });
+            exitRun();
           }
         });
 
@@ -867,9 +869,7 @@ class TaskCoordinator {
           }
 
           if (!checkpoint.docker || !willSimulate) {
-            socket.emit("REQUEST_EXIT", {
-              version: "v1",
-            });
+            exitRun();
           }
         });
 
@@ -951,9 +951,7 @@ class TaskCoordinator {
           }
 
           if (!checkpoint.docker || !willSimulate) {
-            socket.emit("REQUEST_EXIT", {
-              version: "v1",
-            });
+            exitRun();
           }
         });
 
@@ -1081,7 +1079,7 @@ class TaskCoordinator {
   // MARK: HTTP SERVER
   #createHttpServer() {
     const httpServer = createServer(async (req, res) => {
-      logger.log(`[${req.method}]`, req.url);
+      logger.log(`[${req.method}]`, { url: req.url });
 
       const reply = new HttpReply(res);
 
@@ -1111,7 +1109,7 @@ class TaskCoordinator {
     });
 
     httpServer.on("listening", () => {
-      logger.log("server listening on port", HTTP_SERVER_PORT);
+      logger.log("server listening on port", { port: HTTP_SERVER_PORT });
     });
 
     return httpServer;

packages/core/src/v3/utils/structuredLogger.ts

@@ -67,12 +67,12 @@ export class SimpleStructuredLogger implements StructuredLogger {
     ...args: StructuredArgs
   ) {
     const structuredLog = {
-      ...(args.length === 1 ? args[0] : args),
-      ...this.fields,
       timestamp: new Date(),
-      name: this.name,
+      $name: this.name,
       message,
-      level,
+      $level: level,
+      ...(args.length === 1 ? args[0] : args),
+      ...this.fields,
     };
 
     loggerFunction(JSON.stringify(structuredLog));