add tmp cleaner

Author: nicktrn

apps/coordinator/src/checkpointer.ts

@@ -5,6 +5,9 @@ import { nanoid } from "nanoid";
 import fs from "node:fs/promises";
 import { ChaosMonkey } from "./chaosMonkey";
 import { Buildah, Crictl, Exec } from "./exec";
+import { setTimeout } from "node:timers/promises";
+import { TempFileCleaner } from "./cleaner";
+import { numFromEnv, boolFromEnv } from "./util";
 
 type CheckpointerInitializeReturn = {
   canCheckpoint: boolean;
@@ -100,6 +103,7 @@ export class Checkpointer {
   private simulatePushFailureSeconds: number;
 
   private chaosMonkey: ChaosMonkey;
+  private tmpCleaner?: TempFileCleaner;
 
   constructor(private opts: CheckpointerOptions) {
     this.#dockerMode = opts.dockerMode;
@@ -116,6 +120,19 @@ export class Checkpointer {
     this.simulatePushFailureSeconds = opts.simulatePushFailureSeconds ?? 300;
 
     this.chaosMonkey = opts.chaosMonkey ?? new ChaosMonkey(!!process.env.CHAOS_MONKEY_ENABLED);
+
+    if (boolFromEnv("TMP_CLEANER_ENABLED", false)) {
+      const pathsOverride = process.env.TMP_CLEANER_PATHS_OVERRIDE?.split(",") ?? [];
+
+      this.tmpCleaner = new TempFileCleaner({
+        paths: pathsOverride.length ? pathsOverride : [Buildah.tmpDir],
+        maxAgeMinutes: numFromEnv("TMP_CLEANER_MAX_AGE_MINUTES", 60),
+        intervalSeconds: numFromEnv("TMP_CLEANER_INTERVAL_SECONDS", 300),
+        leadingEdge: boolFromEnv("TMP_CLEANER_LEADING_EDGE", false),
+      });
+
+      this.tmpCleaner.start();
+    }
   }
 
   async init(): Promise<CheckpointerInitializeReturn> {
@@ -286,7 +303,7 @@ export class Checkpointer {
           });
 
           this.#waitingForRetry.add(runId);
-          await new Promise((resolve) => setTimeout(resolve, delay.milliseconds));
+          await setTimeout(delay.milliseconds);
 
           if (!this.#waitingForRetry.has(runId)) {
             this.#logger.log("Checkpoint canceled while waiting for retry", { runId });

apps/coordinator/src/cleaner.ts

@@ -0,0 +1,102 @@
+import { SimpleLogger } from "@trigger.dev/core/v3/apps";
+import { Exec } from "./exec";
+import { setTimeout } from "timers/promises";
+
+interface TempFileCleanerOptions {
+  paths: string[];
+  maxAgeMinutes: number;
+  intervalSeconds: number;
+  leadingEdge?: boolean;
+}
+
+export class TempFileCleaner {
+  private logger = new SimpleLogger("[tmp-cleaner]");
+  private enabled = false;
+  private exec = new Exec({ logger: this.logger });
+
+  constructor(private opts: TempFileCleanerOptions) {}
+
+  async start() {
+    this.logger.log("start", this.opts);
+    this.enabled = true;
+
+    if (!this.opts.leadingEdge) {
+      await this.wait();
+    }
+
+    while (this.enabled) {
+      try {
+        await this.clean();
+      } catch (error) {
+        this.logger.error("error during tick", error);
+      }
+
+      await this.wait();
+    }
+  }
+
+  stop() {
+    this.logger.log("stop", this.opts);
+    this.enabled = false;
+  }
+
+  private wait() {
+    return setTimeout(this.opts.intervalSeconds * 1000);
+  }
+
+  private async clean() {
+    for (const path of this.opts.paths) {
+      try {
+        await this.cleanSingle(path);
+      } catch (error) {
+        this.logger.error("error while cleaning", { path, error });
+      }
+    }
+  }
+
+  private async cleanSingle(startingPoint: string) {
+    const maxAgeMinutes = this.opts.maxAgeMinutes;
+
+    const ignoreStartingPoint = ["!", "-path", startingPoint];
+    const onlyDirectDecendants = ["-maxdepth", "1"];
+    const onlyOldFiles = ["-mmin", `+${maxAgeMinutes}`];
+
+    const baseArgs = [
+      startingPoint,
+      ...ignoreStartingPoint,
+      ...onlyDirectDecendants,
+      ...onlyOldFiles,
+    ];
+
+    const duArgs = ["-exec", "du", "-ch", "{}", "+"];
+    const rmArgs = ["-exec", "rm", "-rf", "{}", "+"];
+
+    const du = this.x("find", [...baseArgs, ...duArgs]);
+    const duOutput = await du;
+
+    const duLines = duOutput.stdout.trim().split("\n");
+    const fileCount = duLines.length - 1; // last line is the total
+    const fileSize = duLines.at(-1)?.split("\t")[0];
+
+    if (fileCount === 0) {
+      this.logger.log("nothing to delete", { startingPoint, maxAgeMinutes });
+      return;
+    }
+
+    this.logger.log("deleting old files", { fileCount, fileSize, startingPoint, maxAgeMinutes });
+
+    const rm = this.x("find", [...baseArgs, ...rmArgs]);
+    const rmOutput = await rm;
+
+    if (rmOutput.stderr.length > 0) {
+      this.logger.error("delete unsuccessful", rmOutput);
+      return;
+    }
+
+    this.logger.log("deleted old files", { fileCount, fileSize, startingPoint, maxAgeMinutes });
+  }
+
+  private get x() {
+    return this.exec.x.bind(this.exec);
+  }
+}

apps/coordinator/src/index.ts

@@ -14,6 +14,7 @@ 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";
 collectDefaultMetrics();
@@ -22,26 +23,6 @@ const HTTP_SERVER_PORT = Number(process.env.HTTP_SERVER_PORT || 8020);
 const NODE_NAME = process.env.NODE_NAME || "coordinator";
 const DEFAULT_RETRY_DELAY_THRESHOLD_IN_MS = 30_000;
 
-const boolFromEnv = (env: string, defaultValue: boolean): boolean => {
-  const value = process.env[env];
-
-  if (!value) {
-    return defaultValue;
-  }
-
-  return ["1", "true"].includes(value);
-};
-
-const numFromEnv = (env: string, defaultValue: number): number => {
-  const value = process.env[env];
-
-  if (!value) {
-    return defaultValue;
-  }
-
-  return parseInt(value, 10);
-};
-
 const PLATFORM_ENABLED = ["1", "true"].includes(process.env.PLATFORM_ENABLED ?? "true");
 const PLATFORM_HOST = process.env.PLATFORM_HOST || "127.0.0.1";
 const PLATFORM_WS_PORT = process.env.PLATFORM_WS_PORT || 3030;

apps/coordinator/src/util.ts

@@ -0,0 +1,19 @@
+export const boolFromEnv = (env: string, defaultValue: boolean): boolean => {
+  const value = process.env[env];
+
+  if (!value) {
+    return defaultValue;
+  }
+
+  return ["1", "true"].includes(value);
+};
+
+export const numFromEnv = (env: string, defaultValue: number): number => {
+  const value = process.env[env];
+
+  if (!value) {
+    return defaultValue;
+  }
+
+  return parseInt(value, 10);
+};