Support for uploading to S3 (#95)

- support uploading WACZ to s3-compatible storage (via minio client)
- config storage loaded from env vars, enabled when WACZ output is used.
- support pinging either or an http or a redis key-based webhook,
- webhook: include 'completed' bool to indicate if fully completed crawl or partial (eg. interrupted via signal)
- consolidate redis init to redis.js
- support upload filename with custom variables: can interpolate current timestamp (@ts), hostname (@hostname) and user provided id (@Crawlid)
- README: add docs for s3 storage, remove unused args
- update to pywb 2.6.2, browsertrix-behaviors 0.2.4

* fix to `limit` option, ensure limit check uses shared state

* bump version to 0.5.0-beta.1

Author: ikreymer

README.md

@@ -405,6 +405,45 @@ docker run -p 9037:9037 -v $PWD/crawls:/crawls/ webrecorder/browsertrix-crawler
 
 will start a crawl with 3 workers, and show the screen of each of the workers from `http://localhost:9037/`.
 
+### Uploading crawl output to S3-Compatible Storage
+
+Browsertrix Crawler also includes support for uploading WACZ files to S3-compatible storage, and notifying a webhook when the upload succeeds.
+
+(At this time, S3 upload is supported only when WACZ output is enabled, but WARC uploads may be added in the future).
+
+This feature can currently be enabled by setting environment variables (for security reasons, these settings are not passed in as part of the command-line or YAML config at this time).
+
+<details>
+
+<summary>Environment variables for S3-uploads include:</summary>
+
+- `STORE_ACCESS_KEY` / `STORE_SECRET_KEY` - S3 credentials
+- `STORE_ENDPOINT_URL` - S3 endpoint URL
+- `STORE_PATH` - optional path appended to endpoint, if provided
+- `STORE_FILENAME` - filename or template for filename to put on S3
+- `STORE_USER` - optional username to pass back as part of the webhook callback
+- `CRAWL_ID` - unique crawl id (defaults to container hostname)
+- `WEBHOOK_URL` - the URL of the webhook (can be http://, https:// or redis://)
+
+</details>
+
+#### Webhook Notification
+
+The webhook URL can be an HTTP URL which receives a JSON POST request OR a Redis URL, which specifies a redis list key to which the JSON data is pushed as a string.
+
+<details>
+
+<summary>Webhook notification JSON includes:</summary>
+
+- `id` - crawl id (value of `CRAWL_ID`)
+- `userId` - user id (value of `STORE_USER`)
+- `filename` - bucket path + filename of the file
+- `size` - size of WACZ file
+- `hash` - SHA-256 of WACZ file
+- `completed` - boolean of whether crawl fully completed or partially (due to interrupt signal or other error).
+
+</details>
+
 ## Interrupting and Restarting the Crawl
 
 With version 0.5.0, a crawl can be gracefully interrupted with Ctrl-C (SIGINT) or a SIGTERM.

crawler.js

@@ -1,6 +1,7 @@
 const child_process = require("child_process");
 const path = require("path");
 const fs = require("fs");
+const os = require("os");
 const fsp = require("fs/promises");
 
 // to ignore HTTPS error for HEAD check
@@ -17,16 +18,17 @@ const { RedisCrawlState, MemoryCrawlState } = require("./util/state");
 const AbortController = require("abort-controller");
 const Sitemapper = require("sitemapper");
 const { v4: uuidv4 } = require("uuid");
-const Redis = require("ioredis");
 const yaml = require("js-yaml");
 
 const warcio = require("warcio");
 
 const behaviors = fs.readFileSync(path.join(__dirname, "node_modules", "browsertrix-behaviors", "dist", "behaviors.js"), {encoding: "utf8"});
 
 const  TextExtract  = require("./util/textextract");
+const { S3StorageSync } = require("./util/storage");
 const { ScreenCaster } = require("./util/screencaster");
 const { parseArgs } = require("./util/argParser");
+const { initRedis } = require("./util/redis");
 
 const { getBrowserExe, loadProfile } = require("./util/browser");
 
@@ -43,9 +45,6 @@ class Crawler {
 
     this.emulateDevice = null;
 
-    // links crawled counter
-    this.numLinks = 0;
-
     // pages file
     this.pagesFH = null;
 
@@ -148,10 +147,10 @@ class Crawler {
         throw new Error("stateStoreUrl must start with redis:// -- Only redis-based store currently supported");
       }
 
-      const redis = new Redis(redisUrl, {lazyConnect: true});
+      let redis;
 
       try {
-        await redis.connect();
+        redis = await initRedis(redisUrl);
       } catch (e) {
         throw new Error("Unable to connect to state store Redis: " + redisUrl);
       }
@@ -350,6 +349,25 @@ class Crawler {
       return;
     }
 
+    if (this.params.generateWACZ && process.env.STORE_ENDPOINT_URL) {
+      const endpointUrl = process.env.STORE_ENDPOINT_URL + (process.env.STORE_PATH || "");
+      const storeInfo = {
+        endpointUrl,
+        accessKey: process.env.STORE_ACCESS_KEY,
+        secretKey: process.env.STORE_SECRET_KEY,
+      };
+
+      const opts = {
+        crawlId: process.env.CRAWL_ID || os.hostname(),
+        webhookUrl: process.env.WEBHOOK_URL,
+        userId: process.env.STORE_USER,
+        filename: process.env.STORE_FILENAME || "@[email protected]",
+      };
+
+      console.log("Initing Storage...");
+      this.storage = new S3StorageSync(storeInfo, opts);
+    }
+
     // Puppeteer Cluster init and options
     this.cluster = await Cluster.launch({
       concurrency: this.params.newContext,
@@ -436,6 +454,11 @@ class Crawler {
       // Run the wacz create command
       child_process.spawnSync("wacz" , argument_list, {stdio: "inherit"});
       this.debugLog(`WACZ successfully generated and saved to: ${waczPath}`);
+
+      if (this.storage) {
+        const finished = await this.crawlState.finished();
+        await this.storage.uploadCollWACZ(waczPath, finished);
+      }
     }
   }
 
@@ -543,7 +566,7 @@ class Crawler {
       return false;
     }
 
-    if (this.numLinks >= this.params.limit && this.params.limit > 0) {
+    if (this.params.limit > 0 && (await this.crawlState.numRealSeen() >= this.params.limit)) {
       this.limitHit = true;
       return false;
     }
@@ -553,7 +576,6 @@ class Crawler {
     }
 
     await this.crawlState.add(url);
-    this.numLinks++;
     this.cluster.queue({url, seedId, depth});
     return true;
   }
@@ -656,7 +678,7 @@ class Crawler {
   async awaitPendingClear() {
     this.statusLog("Waiting to ensure pending data is written to WARCs...");
 
-    const redis = new Redis("redis://localhost/0");
+    const redis = await initRedis("redis://localhost/0");
 
     while (true) {
       const res = await redis.get(`pywb:${this.params.collection}:pending`);

package.json

@@ -1,6 +1,6 @@
 {
   "name": "browsertrix-crawler",
-  "version": "0.5.0-beta.0",
+  "version": "0.5.0-beta.1",
   "main": "browsertrix-crawler",
   "repository": "https://github.com/webrecorder/browsertrix-crawler",
   "author": "Ilya Kreymer <[email protected]>, Webrecorder Software",
@@ -10,9 +10,10 @@
   },
   "dependencies": {
     "abort-controller": "^3.0.0",
-    "browsertrix-behaviors": "github:webrecorder/browsertrix-behaviors#skip-mp4-video",
+    "browsertrix-behaviors": "^0.2.4",
     "ioredis": "^4.27.1",
     "js-yaml": "^4.1.0",
+    "minio": "^7.0.18",
     "node-fetch": "^2.6.1",
     "puppeteer-cluster": "github:ikreymer/puppeteer-cluster#async-job-queue",
     "puppeteer-core": "^8.0.0",

screencast/index.html

@@ -14,7 +14,7 @@
   }
 </style>
 <script>
-  const ws = new WebSocket(window.location.origin.replace("http", "ws") + "/ws");
+  const ws = new WebSocket(window.location.href.replace("http", "ws") + "ws");
   ws.addEventListener("message", (event) => handleMessage(event.data));
 
   const unusedElems = [];

tests/custom_driver.test.js

@@ -22,10 +22,12 @@ test("ensure custom driver with custom selector crawls JS files as pages", async
     pages.add(url);
   }
 
+  console.log(pages);
+
   const expectedPages = new Set([
     "https://www.iana.org/",
-    "https://www.iana.org/_js/2013.1/jquery.js",
-    "https://www.iana.org/_js/2013.1/iana.js"
+    "https://www.iana.org/_js/jquery.js",
+    "https://www.iana.org/_js/iana.js"
   ]);
 
   expect(pages).toEqual(expectedPages);

util/redis.js

@@ -0,0 +1,7 @@
+const Redis = require("ioredis");
+
+module.exports.initRedis = async function(url) {
+  const redis = new Redis(url, {lazyConnect: true});
+  await redis.connect();
+  return redis;
+};

util/storage.js

@@ -0,0 +1,115 @@
+const fs = require("fs");
+const os = require("os");
+const { Transform } = require("stream");
+const { createHash } = require("crypto");
+
+const fetch = require("node-fetch");
+const Minio = require("minio");
+
+const { initRedis } = require("./redis");
+
+class S3StorageSync
+{
+  constructor(urlOrData, {filename, webhookUrl, userId, crawlId} = {}) {
+    let url;
+    let accessKey;
+    let secretKey;
+
+    if (typeof(urlOrData) === "string") {
+      url = new URL(urlOrData);
+      accessKey = url.username;
+      secretKey = url.password;
+      url.username = "";
+      url.password = "";
+      this.fullPrefix = url.href;
+
+    } else {
+      url = new URL(urlOrData.endpointUrl);
+      accessKey = urlOrData.accessKey;
+      secretKey = urlOrData.secretKey;
+      this.fullPrefix = url.href;
+    }
+
+    this.client = new Minio.Client({
+      endPoint: url.hostname,
+      port: Number(url.port) || (url.protocol === "https:" ? 443 : 80),
+      useSSL: url.protocol === "https:",
+      accessKey,
+      secretKey
+    });
+
+    this.bucketName = url.pathname.slice(1).split("/")[0];
+
+    this.objectPrefix = url.pathname.slice(this.bucketName.length + 2);
+
+    this.resources = [];
+
+    this.userId = userId;
+    this.crawlId = crawlId;
+    this.webhookUrl = webhookUrl;
+
+    filename = filename.replace("@ts", new Date().toISOString().replace(/[:TZz.]/g, ""));
+    filename = filename.replace("@hostname", os.hostname());
+    filename = filename.replace("@id", this.crawlId);
+
+    this.waczFilename = "data/" + filename;
+  }
+
+  async uploadCollWACZ(filename, completed = true) {
+    const origStream = fs.createReadStream(filename);
+
+    const hash = createHash("sha256");
+    let size = 0;
+    let finalHash;
+
+    const hashTrans = new Transform({
+      transform(chunk, encoding, callback) {
+        size += chunk.length;
+        hash.update(chunk);
+        this.push(chunk);
+        callback();
+      },
+
+      flush(callback) {
+        finalHash = "sha256:" + hash.digest("hex");
+        callback();
+      }
+    });
+
+    const fsStream = origStream.pipe(hashTrans);
+    const res = await this.client.putObject(this.bucketName, this.objectPrefix + this.waczFilename, fsStream);
+    console.log(res);
+
+    const resource = {"path": this.waczFilename, "hash": finalHash, "bytes": size};
+
+    if (this.webhookUrl) {
+      const body = {
+        id: this.crawlId,
+        user: this.userId,
+
+        //filename: `s3://${this.bucketName}/${this.objectPrefix}${this.waczFilename}`,
+        filename: this.fullPrefix + this.waczFilename,
+
+        hash: resource.hash,
+        size: resource.bytes,
+
+        completed
+      };
+
+      console.log("Pinging Webhook: " + this.webhookUrl);
+
+      if (this.webhookUrl.startsWith("http://") || this.webhookUrl.startsWith("https://")) {
+        await fetch(this.webhookUrl, {method: "POST", body: JSON.stringify(body)});
+      } else if (this.webhookUrl.startsWith("redis://")) {
+        const parts = this.webhookUrl.split("/");
+        if (parts.length !== 5) {
+          throw new Error("redis webhook url must be in format: redis://<host>:<port>/<db>/<key>");
+        }
+        const redis = await initRedis(parts.slice(0, 4).join("/"));
+        await redis.rpush(parts[4], JSON.stringify(body));
+      }
+    }
+  }
+}
+
+module.exports.S3StorageSync = S3StorageSync;