Interrupt instead of fail crawl when not fatal (#973)

- Add logger.interrupt() which results in crawler being restarted, but
without failing the crawl, unlike logger.fatal()
- Switch a few logger.fatal() -> logger.interrupt(), when crawl could be
retried, only use fatal() when initial crawl config is invalid for the
most part
- Don't increment 'failing' and set to 'failed' after unexpected errors,
set to interrupted as well to allow for retry (and exponential backoff
if in k8s)
- Allow logger to manage opening/closing its open external log file, via
setOutputFile() and closeLog()
- Move closeLog() and setStatusAndExit() to logger itself so
logger.interrupt() and logger.fatal() can call these
- Use the newer logger.interrupt() for: proxy connection unavailable,
out of disk space, interrupt signal and WACZ upload failed

---------

Co-authored-by: Tessa Walsh <tessa@bitarchivist.net>

Author: ikreymer

docs/docs/user-guide/cli-options.md

@@ -95,8 +95,12 @@ Options:
                                             errors, debug
                                                     [array] [default: ["stats"]]
       --logLevel                            Comma-separated list of log levels t
-                                            o include in logs
-                                                           [array] [default: []]
+                                            o include in logs. By default all bu
+                                            t debug are included. To include deb
+                                            ug messages in logs, debug must also
+                                             be passed to logging option.
+  [array] [choices: "debug", "info", "warn", "error", "interrupt", "fatal"] [def
+                                                                       ault: []]
       --context, --logContext               Comma-separated list of contexts to
                                             include in logs
   [array] [choices: "general", "worker", "recorder", "recorderNetwork", "writer"

src/crawler.ts

@@ -64,12 +64,7 @@ import {
 import { Recorder } from "./util/recorder.js";
 import { SitemapReader } from "./util/sitemapper.js";
 import { ScopedSeed, parseSeeds } from "./util/seeds.js";
-import {
-  WARCWriter,
-  createWARCInfo,
-  setWARCInfo,
-  streamFinish,
-} from "./util/warcwriter.js";
+import { WARCWriter, createWARCInfo, setWARCInfo } from "./util/warcwriter.js";
 import { isHTMLMime, isRedirectStatus } from "./util/reqresp.js";
 import { initProxy } from "./util/proxy.js";
 import { initFlow, nextFlowStep } from "./util/flowbehavior.js";
@@ -123,7 +118,6 @@ export class Crawler {
 
   pagesFH?: WriteStream | null = null;
   extraPagesFH?: WriteStream | null = null;
-  logFH: WriteStream | null = null;
 
   crawlId: string;
 
@@ -482,7 +476,7 @@ export class Crawler {
 
   async bootstrap() {
     if (await isDiskFull(this.params.cwd)) {
-      logger.fatal(
+      logger.interrupt(
         "Out of disk space, exiting",
         {},
         "general",
@@ -512,8 +506,7 @@ export class Crawler {
 
     await fsp.mkdir(this.downloadsDir, { recursive: true });
 
-    this.logFH = fs.createWriteStream(this.logFilename, { flags: "a" });
-    logger.setExternalLogStream(this.logFH);
+    logger.setOutputFile(this.logFilename);
 
     this.infoString = await getInfoString();
     setWARCInfo(this.infoString, this.params.warcInfo);
@@ -678,14 +671,11 @@ export class Crawler {
         exitCode = ExitCodes.Failed;
       }
     } catch (e) {
-      logger.error("Crawl failed", e);
-      exitCode = ExitCodes.Failed;
-      status = "failing";
-      if (await this.crawlState.incFailCount()) {
-        status = "failed";
-      }
+      logger.error("Unexpected error, interrupting", e);
+      exitCode = ExitCodes.GenericError;
+      status = "interrupted";
     } finally {
-      await this.setStatusAndExit(exitCode, status);
+      await logger.setStatusAndExit(exitCode, status);
     }
   }
 
@@ -1677,7 +1667,7 @@ self.__bx_behaviors.selectMainBehavior();
   async checkCanceled() {
     if (this.crawlState && (await this.crawlState.isCrawlCanceled())) {
       await this.cleanupOnCancel();
-      await this.setStatusAndExit(ExitCodes.Success, "canceled");
+      await logger.setStatusAndExit(ExitCodes.Success, "canceled");
       return true;
     }
 
@@ -1690,17 +1680,6 @@ self.__bx_behaviors.selectMainBehavior();
     }
   }
 
-  async setStatusAndExit(exitCode: ExitCodes, status: string) {
-    logger.info(`Exiting, Crawl status: ${status}`);
-
-    await this.closeLog();
-
-    if (this.crawlState && status) {
-      await this.crawlState.setStatus(status);
-    }
-    process.exit(exitCode);
-  }
-
   async serializeAndExit() {
     const initState = await this.crawlState.getStatus();
     // don't do anything if already post-processing!
@@ -1716,15 +1695,17 @@ self.__bx_behaviors.selectMainBehavior();
       await this.closeFiles();
 
       if (!this.done) {
-        await this.setStatusAndExit(
+        logger.interrupt(
+          "Forced interrupt by signal",
+          {},
+          "general",
           ExitCodes.SignalInterruptedForce,
-          "interrupted",
         );
         return true;
       }
     }
 
-    await this.setStatusAndExit(ExitCodes.Success, "done");
+    await logger.setStatusAndExit(ExitCodes.Success, "done");
     return true;
   }
 
@@ -2064,17 +2045,6 @@ self.__bx_behaviors.selectMainBehavior();
     this.browser.crashed = true;
   }
 
-  async closeLog(): Promise<void> {
-    // close file-based log
-    logger.setExternalLogStream(null);
-    if (!this.logFH) {
-      return;
-    }
-    const logFH = this.logFH;
-    this.logFH = null;
-    await streamFinish(logFH);
-  }
-
   async generateWACZ(): Promise<WACZ | null> {
     logger.info("Generating WACZ");
     await this.crawlState.setStatus("generate-wacz");
@@ -2095,7 +2065,7 @@ self.__bx_behaviors.selectMainBehavior();
       if ((await this.crawlState.numDone()) > 0) {
         return null;
       }
-      // fail crawl otherwise
+      // interrupt crawl otherwise
       logger.fatal("No WARC Files, assuming crawl failed");
     }
 
@@ -2111,7 +2081,7 @@ self.__bx_behaviors.selectMainBehavior();
 
     logger.debug("End of log file in WACZ, storing logs to WACZ file");
 
-    await this.closeLog();
+    await logger.closeLog();
 
     const requires = await this.crawlState.getDupeDependentCrawls();
 
@@ -2156,12 +2126,12 @@ self.__bx_behaviors.selectMainBehavior();
       }
       return wacz;
     } catch (e) {
-      logger.error("Error creating WACZ", e);
-      if (!streaming) {
-        logger.fatal("Unable to write WACZ successfully");
-      } else if (this.params.restartsOnError) {
-        await this.setStatusAndExit(ExitCodes.UploadFailed, "interrupted");
-      }
+      logger.interrupt(
+        "Error creating / uploading WACZ",
+        formatErr(e),
+        "wacz",
+        ExitCodes.UploadFailed,
+      );
     }
 
     return null;

src/util/argParser.ts

@@ -25,14 +25,17 @@ import { screenshotTypes } from "./screenshots.js";
 import {
   DEFAULT_EXCLUDE_LOG_CONTEXTS,
   LOG_CONTEXT_TYPES,
+  LOG_LEVEL_TYPES,
   LogContext,
+  LogLevel,
   logger,
 } from "./logger.js";
 import { SaveState } from "./state.js";
 import { loadProxyConfig } from "./proxy.js";
 
 // ============================================================================
 export type CrawlerArgs = ReturnType<typeof parseArgs> & {
+  logLevel: LogLevel[];
   logContext: LogContext[];
   logExcludeContext: LogContext[];
   text: string[];
@@ -277,9 +280,11 @@ class ArgParser {
         },
 
         logLevel: {
-          describe: "Comma-separated list of log levels to include in logs",
+          describe:
+            "Comma-separated list of log levels to include in logs. By default all but debug are included. To include debug messages in logs, debug must also be passed to logging option.",
           type: "array",
           default: [],
+          choices: LOG_LEVEL_TYPES,
           coerce,
         },
 

src/util/logger.ts

@@ -4,6 +4,8 @@
 import { Writable } from "node:stream";
 import { RedisCrawlState } from "./state.js";
 import { ExitCodes } from "./constants.js";
+import { streamFinish } from "./warcwriter.js";
+import fs from "node:fs";
 
 // RegExp.prototype.toJSON = RegExp.prototype.toString;
 Object.defineProperty(RegExp.prototype, "toJSON", {
@@ -63,7 +65,16 @@ export const LOG_CONTEXT_TYPES = [
 
 export type LogContext = (typeof LOG_CONTEXT_TYPES)[number];
 
-export type LogLevel = "debug" | "info" | "warn" | "error" | "fatal";
+export const LOG_LEVEL_TYPES = [
+  "debug",
+  "info",
+  "warn",
+  "error",
+  "interrupt",
+  "fatal",
+] as const;
+
+export type LogLevel = (typeof LOG_LEVEL_TYPES)[number];
 
 export const DEFAULT_EXCLUDE_LOG_CONTEXTS: LogContext[] = [
   "recorderNetwork",
@@ -77,7 +88,7 @@ class Logger {
   debugLogging = false;
   logErrorsToRedis = false;
   logBehaviorsToRedis = false;
-  logLevels: string[] = [];
+  logLevels: LogLevel[] = [];
   contexts: LogContext[] = [];
   excludeContexts: LogContext[] = [];
   defaultLogContext: LogContext = "general";
@@ -92,8 +103,8 @@ class Logger {
     this.fatalExitCode = exitCode;
   }
 
-  setExternalLogStream(logFH: Writable | null) {
-    this.logStream = logFH;
+  setOutputFile(filename: string) {
+    this.logStream = fs.createWriteStream(filename, { flags: "a" });
   }
 
   setDebugLogging(debugLog: boolean) {
@@ -108,7 +119,7 @@ class Logger {
     this.logBehaviorsToRedis = logBehaviorsToRedis;
   }
 
-  setLogLevel(logLevels: string[]) {
+  setLogLevel(logLevels: LogLevel[]) {
     this.logLevels = logLevels;
   }
 
@@ -168,7 +179,7 @@ class Logger {
       //
     }
 
-    const redisErrorLogLevels = ["error", "fatal"];
+    const redisErrorLogLevels = ["error", "interrupt", "fatal"];
     if (
       this.logErrorsToRedis &&
       this.crawlState &&
@@ -225,21 +236,51 @@ class Logger {
     }
   }
 
+  interrupt(
+    message: string,
+    data = {},
+    context: LogContext,
+    exitCode: ExitCodes,
+  ) {
+    this.logAsJSON(
+      `${message}. Interrupting, can restart`,
+      data,
+      context,
+      "interrupt",
+    );
+
+    void this.setStatusAndExit(exitCode, "interrupted");
+  }
+
   fatal(
     message: string,
     data = {},
     context: LogContext = this.defaultLogContext,
     exitCode = ExitCodes.Success,
   ) {
-    exitCode = exitCode || this.fatalExitCode;
     this.logAsJSON(`${message}. Quitting`, data, context, "fatal");
 
-    if (this.crawlState) {
-      this.crawlState
-        .setStatus("failed")
-        .catch(() => {})
-        .finally(process.exit(exitCode));
-    } else {
+    void this.setStatusAndExit(exitCode || this.fatalExitCode, "interrupted");
+  }
+
+  async closeLog() {
+    if (this.logStream) {
+      const logFH = this.logStream;
+      this.logStream = null;
+      await streamFinish(logFH);
+    }
+  }
+
+  async setStatusAndExit(exitCode: ExitCodes, status: string): Promise<void> {
+    try {
+      await this.closeLog();
+
+      if (this.crawlState && status) {
+        await this.crawlState.setStatus(status);
+      }
+    } catch (e) {
+      this.error("Error shutting down, exiting anyway", e);
+    } finally {
       process.exit(exitCode);
     }
   }

src/util/proxy.ts

@@ -433,7 +433,7 @@ export async function runSSHD(
   try {
     await waitForSocksPort;
   } catch (e) {
-    logger.fatal(
+    logger.interrupt(
       "Unable to establish SSH connection for proxy",
       {
         error: e,

src/util/state.ts

@@ -1238,15 +1238,6 @@ return inx;
     });
   }
 
-  async incFailCount() {
-    const key = `${this.crawlId}:status:failcount:${this.uid}`;
-    const res = await this.redis.incr(key);
-
-    // consider failed if 3 failed retries in 60 secs
-    await this.redis.expire(key, 60);
-    return res >= 3;
-  }
-
   async addToQueue(
     {
       url,