From 8de76b5cfe36d5193fe6fe95ed64ff96c7e9737f Mon Sep 17 00:00:00 2001
From: Hornwitser <github@hornwitser.no>
Date: Sat, 11 Dec 2021 21:26:22 +0100
Subject: [PATCH 1/6] Refactor LineSplitter

Pull the LineSplitter class out of lib/factorio to its own stream module
and refactor it to be based on stream.Transform.  This allows composing
with other streams by piping it around.
---
 packages/lib/factorio/server.js | 91 +++------------------------------
 packages/lib/stream.js          | 64 +++++++++++++++++++++++
 test/integration/index.js       | 14 ++---
 test/lib/factorio/server.js     | 37 --------------
 test/lib/stream.js              | 57 +++++++++++++++++++++
 5 files changed, 135 insertions(+), 128 deletions(-)
 create mode 100644 packages/lib/stream.js
 create mode 100644 test/lib/stream.js

diff --git a/packages/lib/factorio/server.js b/packages/lib/factorio/server.js
index e1766281..ba7dd424 100644
--- a/packages/lib/factorio/server.js
+++ b/packages/lib/factorio/server.js
@@ -13,6 +13,7 @@ const libFileOps = require("../file_ops");
 const libIni = require("../ini");
 const { logger } = require("../logging");
 const { escapeRegExp } = require("../helpers");
+const { LineSplitter } = require("../stream");
 
 
 /**
@@ -194,82 +195,6 @@ async function generatePassword(length) {
 	}
 }
 
-/**
- * Line splitter for chunked data
- *
- * Splits a stream of bytes into lines terminated by either line feed or
- * carriage return followed by linefeed.  Each line sepparated is passed
- * to the callback one at a time without the line terminator included.
- *
- * Data is streamed via the data method, and once the stream ends the
- * end method should be called.  This ensures an unterminated line at
- * the end of the stream is passed to the callback.
- *
- * @memberof module:lib/factorio
- * @private
- * @inner
- */
-class LineSplitter {
-
-	/**
-	 * Create a line splitter
-	 *
-	 * @param {function(Buffer)} callback - Function called for every line.
-	 */
-	constructor(callback) {
-		this._callback = callback;
-		this._partial = null;
-	}
-
-	/**
-	 * Stream in bytes
-	 *
-	 * Input bytes to split lines over.  Will invoke the callback for every
-	 * line found in buf and keep any potential partial line at the end of
-	 * buf for the next invocation.
-	 *
-	 * @param {Buffer} buf - Data to stream in.
-	 */
-	data(buf) {
-		if (this._partial) {
-			buf = Buffer.concat([this._partial, buf]);
-			this._partial = null;
-		}
-
-		while (buf.length) {
-			let end = buf.indexOf("\n");
-			if (end === -1) {
-				this._partial = buf;
-				break;
-			}
-
-			let next = end + 1;
-			// Eat carriage return as well if present
-			if (end >= 1 && buf[end-1] === "\r".charCodeAt(0)) {
-				end -= 1;
-			}
-
-			let line = buf.slice(0, end);
-			buf = buf.slice(next);
-			this._callback(line);
-		}
-	}
-
-	/**
-	 * Mark end of stream
-	 *
-	 * Signal the end of the bytestream.  If the previous invocation of the
-	 * data method ended with a partial line this will invoke the callback
-	 * with this line.
-	 */
-	end() {
-		if (this._partial) {
-			this._callback(this._partial);
-			this._partial = null;
-		}
-	}
-}
-
 /**
  * Interpret lines of output from Factorio
  *
@@ -563,9 +488,6 @@ class FactorioServer extends events.EventEmitter {
 		// Array of possible causes for an unexpected shutdown
 		this._unexpected = [];
 
-		this._stdout = new LineSplitter((line) => { this._handleOutput(line, "stdout"); });
-		this._stderr = new LineSplitter((line) => { this._handleOutput(line, "stderr"); });
-
 		// Track autosaving
 		this._runningAutosave = null;
 		this.on("_autosave", name => {
@@ -761,10 +683,12 @@ class FactorioServer extends events.EventEmitter {
 	}
 
 	_attachStdio() {
-		this._server.stdout.on("data", chunk => { this._stdout.data(chunk); });
-		this._server.stdout.on("close", () => { this._stdout.end(); });
-		this._server.stderr.on("data", chunk => { this._stderr.data(chunk); });
-		this._server.stderr.on("close", () => { this._stderr.end(); });
+		let stdout = new LineSplitter({ readableObjectMode: true });
+		stdout.on("data", line => { this._handleOutput(line, "stdout"); });
+		this._server.stdout.pipe(stdout);
+		let stderr = new LineSplitter({ readableObjectMode: true });
+		stderr.on("data", line => { this._handleOutput(line, "stderr"); });
+		this._server.stderr.pipe(stderr);
 	}
 
 	_watchExit() {
@@ -1201,6 +1125,5 @@ module.exports = {
 	_findVersion: findVersion,
 	_randomDynamicPort: randomDynamicPort,
 	_generatePassword: generatePassword,
-	_LineSplitter: LineSplitter,
 	_parseOutput: parseOutput,
 };
diff --git a/packages/lib/stream.js b/packages/lib/stream.js
new file mode 100644
index 00000000..535026a8
--- /dev/null
+++ b/packages/lib/stream.js
@@ -0,0 +1,64 @@
+"use strict";
+const assert = require("assert");
+const fs = require("fs-extra");
+const stream = require("stream");
+
+
+/**
+ * Transform splitting a stream into lines
+ *
+ * Splits up a stream of chunks by newlines and passes on the lines one by
+ * one removing the line feed and optional carriage return from the lines.
+ *
+ * @example
+ * let lineStream = new libStream.LineSplitter({ readableObjectMode: true });
+ * lineStream.on("data", line => { console.log(line.toString("utf8")); });
+ * let fileStream = fs.createReadStream(path);
+ * fileStream.pipe(lineStream);
+ *
+ * @memberof module:lib/stream
+ */
+class LineSplitter extends stream.Transform {
+	constructor(options) {
+		super(options);
+		this._partial = null;
+	}
+
+	_transform(chunk, encoding, callback) {
+		if (this._partial) {
+			chunk = Buffer.concat([this._partial, chunk]);
+			this._partial = null;
+		}
+
+		while (chunk.length) {
+			let end = chunk.indexOf("\n");
+			if (end === -1) {
+				this._partial = chunk;
+				break;
+			}
+
+			let next = end + 1;
+			// Eat carriage return as well if present
+			if (end >= 1 && chunk[end-1] === "\r".charCodeAt(0)) {
+				end -= 1;
+			}
+
+			let line = chunk.slice(0, end);
+			chunk = chunk.slice(next);
+			this.push(line);
+		}
+		callback();
+	}
+
+	_flush(callback) {
+		if (this._partial) {
+			this.push(this._partial);
+			this._partial = null;
+		}
+		callback();
+	}
+}
+
+module.exports = {
+	LineSplitter,
+};
diff --git a/test/integration/index.js b/test/integration/index.js
index 0bc08be8..b254e05f 100644
--- a/test/integration/index.js
+++ b/test/integration/index.js
@@ -9,7 +9,7 @@ const util = require("util");
 const events = require("events");
 
 const libLink = require("@clusterio/lib/link");
-const server = require("@clusterio/lib/factorio/server");
+const { LineSplitter } = require("@clusterio/lib/stream");
 const { ConsoleTransport, logger } = require("@clusterio/lib/logging");
 const libLoggingUtils = require("@clusterio/lib/logging_utils");
 
@@ -133,18 +133,18 @@ function spawn(name, cmd, waitFor) {
 		console.log(cmd);
 		let parts = cmd.split(" ");
 		let process = child_process.spawn(parts[0], parts.slice(1), { cwd: path.join("temp", "test") });
-		let stdout = new server._LineSplitter((line) => {
+		let stdout = new LineSplitter({ readableObjectMode: true });
+		stdout.on("data", line => {
 			line = line.toString("utf8");
 			if (waitFor.test(line)) {
 				resolve(process);
 			}
 			console.log(name, line);
 		});
-		let stderr = new server._LineSplitter((line) => { console.log(name, line.toString("utf8")); });
-		process.stdout.on("data", chunk => { stdout.data(chunk); });
-		process.stdout.on("close", () => { stdout.end(); });
-		process.stderr.on("data", chunk => { stderr.data(chunk); });
-		process.stderr.on("close", () => { stderr.end(); });
+		let stderr = new LineSplitter({ readableObjectMode: true });
+		stderr.on("data", line => { console.log(name, line.toString("utf8")); });
+		process.stdout.pipe(stdout);
+		process.stderr.pipe(stderr);
 	});
 }
 
diff --git a/test/lib/factorio/server.js b/test/lib/factorio/server.js
index c62186a3..c6d05bab 100644
--- a/test/lib/factorio/server.js
+++ b/test/lib/factorio/server.js
@@ -119,43 +119,6 @@ describe("lib/factorio/server", function() {
 		});
 	});
 
-	describe("class LineSplitter", function() {
-		function createSplitter(lines) {
-			return new libFactorio._LineSplitter(line => lines.push(line.toString("utf-8")));
-		}
-
-		it("should split three lines", function() {
-			let lines = [];
-			let ls = createSplitter(lines);
-			ls.data(Buffer.from("line 1\nline 2\nline 3\n"));
-			ls.end();
-			assert.deepEqual(lines, ["line 1", "line 2", "line 3"]);
-		});
-		it("should split three Windows line endings lines", function() {
-			let lines = [];
-			let ls = createSplitter(lines);
-			ls.data(Buffer.from("line 1\r\nline 2\r\nline 3\r\n"));
-			assert.deepEqual(lines, ["line 1", "line 2", "line 3"]);
-		});
-		it("should give the last non-terminated line on .end()", function() {
-			let lines = [];
-			let ls = createSplitter(lines);
-			ls.data(Buffer.from("line a\nline b"));
-			assert.deepEqual(lines, ["line a"]);
-			ls.end();
-			assert.deepEqual(lines, ["line a", "line b"]);
-		});
-		it("should handled partial lines", function() {
-			let lines = [];
-			let ls = createSplitter(lines);
-			ls.data(Buffer.from("part 1"));
-			ls.data(Buffer.from(" part 2 "));
-			ls.data(Buffer.from("part 3\n"));
-			ls.end();
-			assert.deepEqual(lines, ["part 1 part 2 part 3"]);
-		});
-	});
-
 	describe("parseOutput()", function() {
 		it("should parse the test lines", function() {
 			for (let [line, reference] of testLines) {
diff --git a/test/lib/stream.js b/test/lib/stream.js
new file mode 100644
index 00000000..40d45fc3
--- /dev/null
+++ b/test/lib/stream.js
@@ -0,0 +1,57 @@
+"use strict";
+const assert = require("assert").strict;
+const events = require("events");
+const fs = require("fs-extra");
+const path = require("path");
+
+const libStream = require("@clusterio/lib/stream");
+
+
+describe("lib/stream", function() {
+	describe("class LineSplitter", function() {
+		function createSplitter(lines) {
+			let stream = new libStream.LineSplitter({ readableObjectMode: true });
+			stream.on("data", line => lines.push(line.toString("utf-8")));
+			return stream;
+		}
+
+		it("should split three lines", function() {
+			let lines = [];
+			let ls = createSplitter(lines);
+			ls.write(Buffer.from("line 1\nline 2\nline 3\n"));
+			ls.end();
+			assert.deepEqual(lines, ["line 1", "line 2", "line 3"]);
+		});
+		it("should split three Windows line endings lines", function() {
+			let lines = [];
+			let ls = createSplitter(lines);
+			ls.write(Buffer.from("line 1\r\nline 2\r\nline 3\r\n"));
+			ls.end();
+			assert.deepEqual(lines, ["line 1", "line 2", "line 3"]);
+		});
+		it("should handle mixed line endings", function() {
+			let lines = [];
+			let ls = createSplitter(lines);
+			ls.write(Buffer.from("1\n\n2\r\n\r\n3\n4"));
+			ls.end();
+			assert.deepEqual(lines, ["1", "", "2", "", "3", "4"]);
+		});
+		it("should give the last non-terminated line on .end()", function() {
+			let lines = [];
+			let ls = createSplitter(lines);
+			ls.write(Buffer.from("line a\nline b"));
+			assert.deepEqual(lines, ["line a"]);
+			ls.end();
+			assert.deepEqual(lines, ["line a", "line b"]);
+		});
+		it("should handled partial lines", function() {
+			let lines = [];
+			let ls = createSplitter(lines);
+			ls.write(Buffer.from("part 1"));
+			ls.write(Buffer.from(" part 2 "));
+			ls.write(Buffer.from("part 3\n"));
+			ls.end();
+			assert.deepEqual(lines, ["part 1 part 2 part 3"]);
+		});
+	});
+});

From 285d71e13d419c765267b16a40dc169590cc097e Mon Sep 17 00:00:00 2001
From: Hornwitser <github@hornwitser.no>
Date: Sat, 11 Dec 2021 21:32:25 +0100
Subject: [PATCH 2/6] Add ReverseLineSplitter

Add splitter based on LineSplitter which splits lines in reverse from a
reverse stream created from the added createReverseReadStream function.
Enables line processing of files starting from the last line and going
towards the first.
---
 packages/lib/stream.js | 104 +++++++++++++++++++++++++++++++++++++++++
 test/lib/stream.js     |  90 +++++++++++++++++++++++++++++++++++
 2 files changed, 194 insertions(+)

diff --git a/packages/lib/stream.js b/packages/lib/stream.js
index 535026a8..98787397 100644
--- a/packages/lib/stream.js
+++ b/packages/lib/stream.js
@@ -59,6 +59,110 @@ class LineSplitter extends stream.Transform {
 	}
 }
 
+/**
+ * Transform splitting a reverse stream into lines
+ *
+ * Splits up a reverse stream of chunks created by {@link
+ * module:lib/stream.createReverseReadStream} by newlines and passes on the
+ * lines one by one in reverse order removing the line feed and optional
+ * carriage return from the lines.
+ *
+ * @example
+ * let revLineStream = new libStream.ReverseLineSplitter({ readableObjectMode: true });
+ * revLineStream.on("data", line => { console.log(line.toString("utf8")); });
+ * let revFileStream = await libStream.createReverseReadStream(path);
+ * revFileStream.pipe(revLineStream);
+ *
+ * @memberof module:lib/stream
+ */
+class ReverseLineSplitter extends stream.Transform {
+	constructor(options) {
+		super(options);
+		this._partial = null;
+	}
+
+	_transform(chunk, encoding, callback) {
+		if (this._partial) {
+			chunk = Buffer.concat([chunk, this._partial]);
+			this._partial = null;
+		}
+
+		while (chunk.length) {
+			let next = chunk.lastIndexOf("\n", -2);
+			if (next === -1) {
+				this._partial = chunk;
+				break;
+			}
+
+			let end = chunk.length - 1;
+			if (chunk[end] === "\n".charCodeAt(0)) {
+				end -= 1;
+				// Eat carriage return as well if present
+				if (end - next > 0 && chunk[end] === "\r".charCodeAt(0)) {
+					end -= 1;
+				}
+			}
+
+			let line = chunk.slice(next + 1, end + 1);
+			chunk = chunk.slice(0, next + 1);
+			this.push(line);
+		}
+		callback();
+	}
+
+	_flush(callback) {
+		if (this._partial) {
+			let end = this._partial.length - 1;
+			if (this._partial[end] === "\n".charCodeAt(0)) {
+				end -= 1;
+				// Eat carriage return as well if present
+				if (end > 0 && this._partial[end] === "\r".charCodeAt(0)) {
+					end -= 1;
+				}
+			}
+			this.push(this._partial.slice(0, end + 1));
+			this._partial = null;
+		}
+		callback();
+	}
+}
+
+
+/**
+ * Create a file read stream running in reverse
+ *
+ * Like fs.createReadStream but provides data chunks starting from the end
+ * of the file and progressing towards the start of the file.  The chunks
+ * are not reversed, thus the file content can be recreated by concatenating
+ * the chunks together in the reverse order they are read.
+ *
+ * @param {string} path -
+ *     Path to file to open. May be anything accepted by fs.open.
+ * @param {object} options -
+ *     Options to pass to fs.createReadStream.  Values passed for `fd` and
+ *     `fs` will be ignored.
+ * @memberof module:lib/stream
+ */
+async function createReverseReadStream(path, options) {
+	const fileFd = await fs.open(path, "r");
+	const fileSize = (await fs.fstat(fileFd)).size;
+	let filePosition = fileSize;
+	const reverseFs = {
+		read(fd, buffer, offset, length, position, callback) {
+			assert(fd === fileFd);
+			assert(position === undefined);
+			length = Math.min(length, filePosition);
+			filePosition -= length;
+			return fs.read(fd, buffer, offset, length, filePosition, callback);
+		},
+		open() { assert(false); },
+		close(fd, callback) { return fs.close(fd, callback); },
+	};
+	return fs.createReadStream("", { ...options, fd: fileFd, fs: reverseFs });
+}
+
 module.exports = {
 	LineSplitter,
+	ReverseLineSplitter,
+	createReverseReadStream,
 };
diff --git a/test/lib/stream.js b/test/lib/stream.js
index 40d45fc3..721ed090 100644
--- a/test/lib/stream.js
+++ b/test/lib/stream.js
@@ -54,4 +54,94 @@ describe("lib/stream", function() {
 			assert.deepEqual(lines, ["part 1 part 2 part 3"]);
 		});
 	});
+
+	describe("class ReverseLineSplitter", function() {
+		function createReverseSplitter(lines) {
+			let stream = new libStream.ReverseLineSplitter({ readableObjectMode: true });
+			stream.on("data", line => lines.push(line.toString("utf-8")));
+			return stream;
+		}
+
+		it("should split three lines", function() {
+			let lines = [];
+			let ls = createReverseSplitter(lines);
+			ls.write(Buffer.from("line 1\nline 2\nline 3\n"));
+			ls.end();
+			assert.deepEqual(lines, ["line 3", "line 2", "line 1"]);
+		});
+		it("should split three Windows line endings lines", function() {
+			let lines = [];
+			let ls = createReverseSplitter(lines);
+			ls.write(Buffer.from("line 1\r\nline 2\r\nline 3\r\n"));
+			ls.end();
+			assert.deepEqual(lines, ["line 3", "line 2", "line 1"]);
+		});
+		it("should handle mixed line endings", function() {
+			let lines = [];
+			let ls = createReverseSplitter(lines);
+			ls.write(Buffer.from("1\n\n2\r\n\r\n3\n4"));
+			ls.end();
+			assert.deepEqual(lines, ["4", "3", "", "2", "", "1"]);
+		});
+		it("should give the first line on .end()", function() {
+			let lines = [];
+			let ls = createReverseSplitter(lines);
+			ls.write(Buffer.from("line a\nline b"));
+			assert.deepEqual(lines, ["line b"]);
+			ls.end();
+			assert.deepEqual(lines, ["line b", "line a"]);
+		});
+		it("should handled partial lines", function() {
+			let lines = [];
+			let ls = createReverseSplitter(lines);
+			ls.write(Buffer.from("part 3\n"));
+			ls.write(Buffer.from(" part 2 "));
+			ls.write(Buffer.from("part 1"));
+			ls.end();
+			assert.deepEqual(lines, ["part 1 part 2 part 3"]);
+		});
+	});
+
+	describe("createReverseReadStream()", function() {
+		it("should read chunks in a file in reverse", async function() {
+			let content = "";
+			for (let i = 0; i < 10; i++) { content += String(i).repeat(10); }
+			await fs.outputFile(path.join("temp", "test", "reverse.txt"), content);
+			let stream = await libStream.createReverseReadStream(
+				path.join("temp", "test", "reverse.txt"),
+				{ encoding: "utf8", highWaterMark: 10 }
+			);
+			let index = 9;
+			stream.on("data", data => {
+				assert.equal(data, String(index).repeat(10));
+				index -= 1;
+			});
+			await events.once(stream, "end");
+		});
+
+		it("should reverse the lines of a file", async function() {
+			let ws = fs.createWriteStream(path.join("temp", "test", "reverse.txt"));
+			for (let i = 1; i < 100000; i++) {
+				ws.write(Buffer.from(`${i}\n`));
+			}
+			ws.end();
+			await events.once(ws, "finish");
+
+			let lineStream = new libStream.ReverseLineSplitter({ readableObjectMode: true });
+			let index = 99999;
+			let rs = await libStream.createReverseReadStream(path.join("temp", "test", "reverse.txt"));
+			lineStream.on("data", line => {
+				let n = Number(line.toString());
+				if (index !== n) {
+					rs.destroy();
+					assert.equal(n, index);
+				}
+				index -= 1;
+			});
+			rs.pipe(lineStream);
+			await events.once(lineStream, "end");
+
+			assert.equal(index, 0, "line check did not reach the end");
+		});
+	});
 });

From 200ccf8e06b0b534e346213dbfd98f9669b89eaa Mon Sep 17 00:00:00 2001
From: Hornwitser <github@hornwitser.no>
Date: Wed, 15 Dec 2021 00:19:47 +0100
Subject: [PATCH 3/6] Fix LogConsole re-requesting logs

Turns out objects in the second argument to useEffect is compared by
Object.is(), this means if the array containing the instances or slaves
to show logs for get re-created during an update the LogConsole
component would send a new request for the logs.

Fix by converting the slave_ids and instance_ids arrays into strings
before passing them on to useEffect.
---
 packages/web_ui/src/components/LogConsole.jsx | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/packages/web_ui/src/components/LogConsole.jsx b/packages/web_ui/src/components/LogConsole.jsx
index a9fd58d7..60b90bea 100644
--- a/packages/web_ui/src/components/LogConsole.jsx
+++ b/packages/web_ui/src/components/LogConsole.jsx
@@ -94,7 +94,7 @@ export default function LogConsole(props) {
 		return () => {
 			control.offLog(logFilter, logHandler);
 		};
-	}, [props.all, props.master, props.slaves, props.instances]);
+	}, [props.all, props.master, (props.slaves || []).join(), (props.instances || []).join()]);
 
 	return <>
 		<Paragraph code className="instance-console">

From fd1f3329bdd0e0d067c305497e5d36f95b85ab66 Mon Sep 17 00:00:00 2001
From: Hornwitser <github@hornwitser.no>
Date: Wed, 15 Dec 2021 16:37:36 +0100
Subject: [PATCH 4/6] Split logs by date

Move the default location of the log files into a configurable logs
folder and split the written logs up by UTC date to more easily allow
deleting of old logs to be performed.  The log rotation is implemented
by winston-daily-rotate-file.

The old log format will automatically be migrated to the new format and
location if the target folder for the new logs don't yet exists.
---
 packages/lib/logging_utils.js | 61 +++++++++++++++++++++++++++++++++++
 packages/master/master.js     | 40 ++++++++++++++++++++---
 packages/master/package.json  |  1 +
 packages/slave/package.json   |  1 +
 packages/slave/slave.js       | 31 +++++++++++++++---
 5 files changed, 125 insertions(+), 9 deletions(-)

diff --git a/packages/lib/logging_utils.js b/packages/lib/logging_utils.js
index c80b6997..6a8ac8da 100644
--- a/packages/lib/logging_utils.js
+++ b/packages/lib/logging_utils.js
@@ -1,4 +1,7 @@
 "use strict";
+const events = require("events");
+const fs = require("fs-extra");
+const path = require("path");
 const winston = require("winston");
 const Transport = require("winston-transport");
 const { LEVEL, MESSAGE } = require("triple-beam");
@@ -6,6 +9,7 @@ const chalk = require("chalk");
 
 const libLink = require("./link");
 const libErrors = require("./errors");
+const libStream = require("./stream");
 
 
 /**
@@ -121,10 +125,67 @@ class LinkTransport extends Transport {
 	}
 }
 
+async function migrateLogs(log, directory, pattern) {
+	const dateRegExp = /^\d{4}-\d{2}-\d{2}/;
+	let lineStream = new libStream.LineSplitter({ readableObjectMode: true });
+	let fileStream = fs.createReadStream(log);
+	fileStream.pipe(lineStream);
+
+	let currentDate = null;
+	let output = null;
+
+	async function switchOutput(newOutputPath) {
+		if (output) {
+			output.end();
+			await events.once(output, "close");
+		}
+
+		output = fs.createWriteStream(newOutputPath, { flags: "ax" });
+		try {
+			await events.once(output, "open");
+			// eslint-disable-next-line no-console
+			console.log(`writing ${newOutputPath}`);
+		} catch (err) {
+			if (err.code === "EEXIST") {
+				// eslint-disable-next-line no-console
+				console.warn(`Discarding logs for ${newDate} due to target logfile already existing`);
+				output = null;
+			} else {
+				throw err;
+			}
+		}
+	}
+
+	for await (let line of lineStream) {
+		let info = {};
+		try {
+			info = JSON.parse(line);
+		} catch (err) {
+			// Ignore
+		}
+
+		if (typeof info.timestamp === "string" && dateRegExp.test(info.timestamp)) {
+			let newDate = info.timestamp.slice(0, 10);
+			// Only update date forward in time.
+			if (newDate !== currentDate && (!currentDate || newDate > currentDate)) {
+				await switchOutput(path.join(directory, pattern.replace("%DATE%", newDate)));
+				currentDate = newDate;
+			}
+		}
+
+		if (output) {
+			if (!output.write(Buffer.concat([line, Buffer.from("\n")]))) {
+				await events.once(output, "drain");
+			}
+		}
+	}
+}
+
 
 module.exports = {
 	TerminalFormat,
 	LinkTransport,
+	migrateLogs,
 
 	// for testing only
 	_formatServerOutput: formatServerOutput,
diff --git a/packages/master/master.js b/packages/master/master.js
index 89d60bf8..f2a09f0d 100755
--- a/packages/master/master.js
+++ b/packages/master/master.js
@@ -22,6 +22,7 @@ const setBlocking = require("set-blocking");
 const yargs = require("yargs");
 const util = require("util");
 const winston = require("winston");
+require("winston-daily-rotate-file");
 const jwt = require("jsonwebtoken");
 
 // homebrew modules
@@ -189,6 +190,12 @@ async function initialize() {
 			choices: ["none"].concat(Object.keys(levels)),
 			type: "string",
 		})
+		.option("log-directory", {
+			nargs: 1,
+			describe: "Directory to place logs in",
+			default: "logs",
+			type: "string",
+		})
 		.option("config", {
 			nargs: 1,
 			describe: "master config file to use",
@@ -227,20 +234,45 @@ async function initialize() {
 		.argv
 	;
 
+	{
+		// Migration from alpha-10 single file logs, note that we can't use
+		// the logger here as it's not initalized yet.
+		/* eslint-disable no-console */
+		let clusterLogDirectory = path.join(parameters.args.logDirectory, "cluster");
+		if (!await fs.pathExists(clusterLogDirectory) && await fs.pathExists("cluster.log")) {
+			console.log("Migrating cluster log...");
+			await fs.ensureDir(clusterLogDirectory);
+			await libLoggingUtils.migrateLogs("cluster.log", clusterLogDirectory, "cluster-%DATE%.log");
+			console.log("Migration complete, you should delete cluster.log now");
+		}
+
+		let masterLogDirectory = path.join(parameters.args.logDirectory, "master");
+		if (!await fs.pathExists(masterLogDirectory) && await fs.pathExists("master.log")) {
+			console.log("Migrating master log...");
+			await fs.ensureDir(masterLogDirectory);
+			await libLoggingUtils.migrateLogs("master.log", masterLogDirectory, "master-%DATE%.log");
+			console.log("Migration complete, you should delete master.log now");
+		}
+		/* eslint-enable no-console */
+	}
+
 	// Combined log stream of the whole cluster.
 	parameters.clusterLogger = winston.createLogger({
 		format: winston.format.json(),
 		level: "verbose",
 		levels,
 	});
-	parameters.clusterLogger.add(new winston.transports.File({
-		filename: "cluster.log",
+	parameters.clusterLogger.add(new winston.transports.DailyRotateFile({
+		filename: "cluster-%DATE%.log",
+		utc: true,
+		dirname: path.join(parameters.args.logDirectory, "cluster"),
 	}));
 
 	// Log stream for the master server.
-	logger.add(new winston.transports.File({
+	logger.add(new winston.transports.DailyRotateFile({
 		format: winston.format.json(),
-		filename: "master.log",
+		filename: "master-%DATE%.log",
+		dirname: path.join(parameters.args.logDirectory, "master"),
 	}));
 	logger.add(new winston.transports.Stream({
 		stream: parameters.clusterLogger,
diff --git a/packages/master/package.json b/packages/master/package.json
index 76283a82..a5bf1a14 100644
--- a/packages/master/package.json
+++ b/packages/master/package.json
@@ -28,6 +28,7 @@
     "jszip": "^3.2.2",
     "set-blocking": "^2.0.0",
     "winston": "^3.3.3",
+    "winston-daily-rotate-file": "^4.5.5",
     "ws": "^7.2.3",
     "yargs": "^14.2.0"
   },
diff --git a/packages/slave/package.json b/packages/slave/package.json
index eed960b0..3aad3aa5 100644
--- a/packages/slave/package.json
+++ b/packages/slave/package.json
@@ -24,6 +24,7 @@
     "pidusage": "^2.0.18",
     "set-blocking": "^2.0.0",
     "winston": "^3.3.3",
+    "winston-daily-rotate-file": "^4.5.5",
     "yargs": "^14.2.0"
   },
   "publishConfig": {
diff --git a/packages/slave/slave.js b/packages/slave/slave.js
index 46fea491..f33ae4dc 100755
--- a/packages/slave/slave.js
+++ b/packages/slave/slave.js
@@ -23,6 +23,7 @@ const phin = require("phin");
 const util = require("util");
 const version = require("./package").version;
 const winston = require("winston");
+require("winston-daily-rotate-file");
 
 // internal libraries
 const libFileOps = require("@clusterio/lib/file_ops");
@@ -1734,6 +1735,12 @@ async function startSlave() {
 			choices: ["none"].concat(Object.keys(levels)),
 			type: "string",
 		})
+		.option("log-directory", {
+			nargs: 1,
+			describe: "Directory to place logs in",
+			default: "logs",
+			type: "string",
+		})
 		.option("config", {
 			nargs: 1,
 			describe: "slave config file to use",
@@ -1754,11 +1761,25 @@ async function startSlave() {
 		.argv
 	;
 
-	logger.add(new winston.transports.File({
-		format: winston.format.combine(
-			winston.format.json(),
-		),
-		filename: "slave.log",
+	{
+		// Migration from alpha-10 single file logs, note that we can't use
+		// the logger here as it's not initalized yet.
+		/* eslint-disable no-console */
+		let slaveLogDirectory = path.join(args.logDirectory, "slave");
+		if (!await fs.pathExists(slaveLogDirectory) && await fs.pathExists("slave.log")) {
+			console.log("Migrating slave log...");
+			await fs.ensureDir(slaveLogDirectory);
+			await libLoggingUtils.migrateLogs("slave.log", slaveLogDirectory, "slave-%DATE%.log");
+			console.log("Migration complete, you should delete slave.log now");
+		}
+		/* eslint-enable no-console */
+	}
+
+	logger.add(new winston.transports.DailyRotateFile({
+		format: winston.format.json(),
+		filename: "slave-%DATE%.log",
+		utc: true,
+		dirname: path.join(args.logDirectory, "slave"),
 	}));
 	if (args.logLevel !== "none") {
 		logger.add(new ConsoleTransport({

From 601ebba91979f9b0b82a676deabe61f6d3e7d55a Mon Sep 17 00:00:00 2001
From: Hornwitser <github@hornwitser.no>
Date: Sun, 19 Dec 2021 01:37:50 +0100
Subject: [PATCH 5/6] Replace log query implementation

The interface provided by Winston is unusably slow for querying large
logs.  Reimplement the log querying with both forwards and backwards
line reading and a simple file based index to skip files which do not
contain any lines matching the filter.
---
 .gitignore                               |   3 +-
 packages/lib/logging_utils.js            | 322 +++++++++++++++++++++++
 packages/master/src/ControlConnection.js |  61 +----
 packages/master/src/Master.js            |  40 +++
 test/file/log/excluded.txt               |   1 +
 test/file/log/test-2021-12-10.log        |   9 +
 test/file/log/test-2021-12-11.log        |   3 +
 test/file/log/test-2021-12-12.log        |   1 +
 test/file/log/test-2100-12-31.log        |   0
 test/lib/logging_utils.js                | 111 ++++++++
 10 files changed, 497 insertions(+), 54 deletions(-)
 create mode 100644 test/file/log/excluded.txt
 create mode 100644 test/file/log/test-2021-12-10.log
 create mode 100644 test/file/log/test-2021-12-11.log
 create mode 100644 test/file/log/test-2021-12-12.log
 create mode 100644 test/file/log/test-2100-12-31.log

diff --git a/.gitignore b/.gitignore
index 56c8fbd4..e05a2f70 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1,7 +1,6 @@
 # Logs
 logs
-*.log
-npm-debug.log*
+*debug.log*
 
 # Runtime data
 pids
diff --git a/packages/lib/logging_utils.js b/packages/lib/logging_utils.js
index 6a8ac8da..6b2d3af4 100644
--- a/packages/lib/logging_utils.js
+++ b/packages/lib/logging_utils.js
@@ -1,3 +1,9 @@
+/**
+ * Utilities for logging on the Node.js side
+ *
+ * @author Hornwitser
+ * @module lib/logging_utils
+ */
 "use strict";
 const events = require("events");
 const fs = require("fs-extra");
@@ -9,6 +15,8 @@ const chalk = require("chalk");
 
 const libLink = require("./link");
 const libErrors = require("./errors");
+const libFileOps = require("./file_ops");
+const { levels, logger } = require("./logging");
 const libStream = require("./stream");
 
 
@@ -59,6 +67,7 @@ function formatServerOutput(parsed) {
 
 /**
  * Formats winston log messages for a character terminal.
+ * @static
  */
 class TerminalFormat {
 	constructor(options = {}) {
@@ -99,6 +108,7 @@ class TerminalFormat {
 
 /**
  * Sends logs over a lib/link connection.
+ * @static
  */
 class LinkTransport extends Transport {
 	constructor(options) {
@@ -125,6 +135,315 @@ class LinkTransport extends Transport {
 	}
 }
 
+const logFileGlob = /^[a-z]+-(\d{4}-\d{2}-\d{2})\.log$/;
+/**
+ * Filter object for logs
+ * @typedef {Object} module:lib/logging_utils~LogFilter
+ * @property {string} [max_level] -
+ *     Maximum log level to include. Higher levels are more verbose.
+ * @property {boolean} [all] -
+ *     Include log entries from master, all slaves and all instances.
+ * @property {boolean} [master] -
+ *     Include log entries from the master server.
+ * @property {Array<number>} [slave_ids] -
+ *     Include log entries for the given slaves and instances of those
+ *     slaves by id.
+ * @property {Array<number>} [instance_ids] -
+ *     Include log entries for the given instances by id.
+ */
+
+/**
+ * Keeps an index over a log directory to speed up queries to it
+ * @static
+ */
+class LogIndex {
+	constructor(logDirectory, serialized) {
+		this.logDirectory = logDirectory;
+		this.index = new Map();
+
+		if (serialized.version !== 0) {
+			return;
+		}
+
+		for (let [file, serializedEntry] of Object.entries(serialized.files)) {
+			this.index.set(file, {
+				levels: new Set(serializedEntry.levels),
+				master: serializedEntry.master,
+				slave_ids: new Set(serializedEntry.slave_ids),
+				instance_ids: new Set(serializedEntry.instance_ids),
+			});
+		}
+	}
+
+	serialize() {
+		let serialized = {
+			version: 0,
+			files: {},
+		};
+
+		for (let [file, entry] of this.index) {
+			serialized.files[file] = {
+				levels: [...entry.levels],
+				master: entry.master,
+				slave_ids: [...entry.slave_ids],
+				instance_ids: [...entry.instance_ids],
+			};
+		}
+
+		return serialized;
+	}
+
+	/**
+	 * Load or create an empty index for the given directory
+	 *
+	 * Loads index.json from the given log directy expecting it to contain
+	 * data about previously indexed files saved with {@link
+	 * module:lib/logging_utils.LogIndex#save}.
+	 *
+	 * @param {string} logDirectory - Path to directory to load from.
+	 * @returns {module:lib/logging_utils.LogIndex} loaded or created log index.
+	 */
+	static async load(logDirectory) {
+		try {
+			let content = await fs.readFile(path.join(logDirectory, "index.json"));
+			return new LogIndex(logDirectory, JSON.parse(content));
+		} catch (err) {
+			if (err.code !== "ENOENT") {
+				logger.warn(`Failed to load ${path.join(logDirectory, "index.json")}: ${err}`);
+			}
+			return new LogIndex(logDirectory, {});
+		}
+	}
+
+	/**
+	 * Save index
+	 *
+	 * Saves corrently held index into index.json in the log directory it
+	 * was loaded/created from.
+	 */
+	async save() {
+		await libFileOps.safeOutputFile(
+			path.join(this.logDirectory, "index.json"),
+			JSON.stringify(this.serialize(), null, 4)
+		);
+	}
+
+	async indexFile(file) {
+		let filePath = path.join(this.logDirectory, file);
+		let lineStream = new libStream.LineSplitter({ readableObjectMode: true });
+		let fileStream = fs.createReadStream(filePath);
+		fileStream.pipe(lineStream);
+		let entry = {
+			levels: new Set(),
+			master: false,
+			slave_ids: new Set(),
+			instance_ids: new Set(),
+		};
+		for await (let line of lineStream) {
+			let info;
+			try {
+				info = JSON.parse(line);
+			} catch (err) {
+				entry.levels.add("info");
+				entry.master = true;
+				continue;
+			}
+			if (info.level) {
+				entry.levels.add(info.level);
+			}
+			if (info.slave_id === undefined) {
+				entry.master = true;
+			} else if (info.instance_id === undefined) {
+				entry.slave_ids.add(info.slave_id);
+			}
+			if (info.instance_id !== undefined) {
+				entry.instance_ids.add(info.instance_id);
+			}
+		}
+		return entry;
+	}
+
+	/**
+	 * Build index for files currently present in the log directory
+	 */
+	async buildIndex() {
+		let today = new Date(Date.now() - 300e3).toISOString().slice(0, 10);
+
+		for (let file of await fs.readdir(this.logDirectory)) {
+			let match = logFileGlob.exec(file);
+			if (!match || match[1] >= today || this.index.has(file)) {
+				continue;
+			}
+
+			let entry = await this.indexFile(file);
+			this.index.set(file, entry);
+		}
+	}
+
+	/**
+	 * Check if a file includes entries for the given filter
+	 *
+	 * @param {string} file -
+	 *     Name of file in the log directory to check for.
+	 * @param {module:lib/logging_utils~LogFilter} filter -
+	 *     Filter to check index if file contains entries for.
+	 * @returns {boolean}
+	 *     true if the file may contain entries included by the filter.
+	 */
+	filterIncludesFile(file, { max_level, all, master, instance_ids, slave_ids }) {
+		let entry = this.index.get(file);
+		if (!entry) {
+			return true;
+		}
+
+		const hasOwnProperty = Object.prototype.hasOwnProperty;
+		if (max_level && hasOwnProperty.call(levels, max_level)) {
+			let fileLevels = [...entry.levels].map(
+				fileLevel => (hasOwnProperty.call(levels, fileLevel) ? levels[fileLevel] : levels.verbose)
+			);
+			if (fileLevels.every(level => level > levels[max_level])) {
+				return false;
+			}
+		}
+
+		if (all) {
+			return true;
+		}
+		if (master && entry.master) {
+			return true;
+		}
+		if (slave_ids && slave_ids.some(id => entry.slave_ids.has(id))) {
+			return true;
+		}
+		if (instance_ids && instance_ids.some(id => entry.instance_ids.has(id))) {
+			return true;
+		}
+		return false;
+	}
+}
+
+/**
+ * Create log filter by level and source.
+ *
+ * @param {module:lib/logging_utils~LogFilter} filter -
+ *     Filter to filter log entries by.
+ * @returns {function(object): boolean}
+ *     filter returning true for log entries that match it.
+ * @static
+ */
+function logFilter({ all, master, slave_ids, instance_ids, max_level }) {
+	return info => {
+		// Note: reversed to filter out undefined levels
+		if (max_level && !(levels[info.level] <= levels[max_level])) {
+			return false;
+		}
+
+		if (all) {
+			return true;
+		}
+		if (master && info.slave_id === undefined) {
+			return true;
+		}
+		if (
+			slave_ids
+			&& info.slave_id !== undefined
+			&& info.instance_id === undefined
+			&& slave_ids.includes(info.slave_id)
+		) {
+			return true;
+		}
+		if (instance_ids && info.instance_id !== undefined && instance_ids.includes(info.instance_id)) {
+			return true;
+		}
+		return false;
+	};
+}
+
+/**
+ * Filter object for querying logs
+ * @typedef {Object} module:lib/logging_utils~QueryLogFilter
+ * @property {number} [limit=100] - Maximum number of entries to return.
+ * @property {string} [order="asc"] -
+ *     Return entries in ascending ("asc") date order or desceding ("desc")
+ *     date order.
+ * @property {string} [max_level] -
+ *     Maximum log level to include. Higher levels are more verbose.
+ * @property {boolean} [all] -
+ *     Include log entries from master, all slaves and all instances.
+ * @property {boolean} [master] -
+ *     Include log entries from the master server.
+ * @property {Array<number>} [slave_ids] -
+ *     Include log entries for the given slaves and instances of those
+ *     slaves by id.
+ * @property {Array<number>} [instance_ids] -
+ *     Include log entries for the given instances by id.
+ */
+
+/**
+ * Query log directory
+ *
+ * @param {string} logDirectory - path to directory with logs.
+ * @param {module:lib/logging_utils~QueryLogFilter} filter -
+ *     Filter to limit logs by.
+ * @param {module:lib/logging_utils.LogIndex=} index -
+ *     Index to speed up query with.
+ * @returns {Array<Object>} log entries matching the filter
+ */
+async function queryLog(logDirectory, filter, index) {
+	let files = (await fs.readdir(logDirectory)).filter(entry => logFileGlob.test(entry));
+
+	filter = {
+		limit: 100,
+		order: "asc",
+		...filter,
+	};
+
+	files.sort();
+	if (filter.order === "desc") {
+		files.reverse();
+	}
+
+	const includeEntry = logFilter(filter);
+	let log = [];
+	for (let file of files) {
+		if (index && !index.filterIncludesFile(file, filter)) {
+			continue;
+		}
+		let filePath = path.join(logDirectory, file);
+		let fileStream;
+		let lineStream;
+		if (filter.order === "asc") {
+			lineStream = new libStream.LineSplitter({ readableObjectMode: true });
+			fileStream = fs.createReadStream(filePath);
+		} else {
+			lineStream = new libStream.ReverseLineSplitter({ readableObjectMode: true });
+			fileStream = await libStream.createReverseReadStream(filePath);
+		}
+		lineStream.on("data", line => {
+			let info;
+			try {
+				info = JSON.parse(line);
+			} catch (err) {
+				info = { level: "info", message: line.toString("utf8") };
+			}
+			if (includeEntry(info)) {
+				log.push(info);
+				if (log.length >= filter.limit) {
+					lineStream.destroy();
+					fileStream.destroy();
+				}
+			}
+		});
+		fileStream.pipe(lineStream);
+		await events.once(lineStream, "close");
+
+		if (log.length >= filter.limit) {
+			break;
+		}
+	}
+	return log;
+}
+
 async function migrateLogs(log, directory, pattern) {
 	const dateRegExp = /^\d{4}-\d{2}-\d{2}/;
 	let lineStream = new libStream.LineSplitter({ readableObjectMode: true });
@@ -185,6 +504,9 @@ async function migrateLogs(log, directory, pattern) {
 module.exports = {
 	TerminalFormat,
 	LinkTransport,
+	LogIndex,
+	logFilter,
+	queryLog,
 	migrateLogs,
 
 	// for testing only
diff --git a/packages/master/src/ControlConnection.js b/packages/master/src/ControlConnection.js
index 7062ce7d..5535bcc1 100644
--- a/packages/master/src/ControlConnection.js
+++ b/packages/master/src/ControlConnection.js
@@ -1,12 +1,10 @@
 "use strict";
 const jwt = require("jsonwebtoken");
-const util = require("util");
-const winston = require("winston");
 
 const libConfig = require("@clusterio/lib/config");
 const libErrors = require("@clusterio/lib/errors");
 const libLink = require("@clusterio/lib/link");
-const { levels, logger } = require("@clusterio/lib/logging");
+const { logger } = require("@clusterio/lib/logging");
 const libLoggingUtils = require("@clusterio/lib/logging_utils");
 const libPlugin = require("@clusterio/lib/plugin");
 const libPrometheus = require("@clusterio/lib/prometheus");
@@ -402,29 +400,6 @@ class ControlConnection extends BaseConnection {
 		this.updateLogSubscriptions();
 	}
 
-	static logFilter({ all, master, slave_ids, instance_ids, max_level }) {
-		return info => {
-			// Note: reversed to filter out undefined levels
-			if (max_level && !(levels[info.level] <= levels[max_level])) {
-				return false;
-			}
-
-			if (all) {
-				return true;
-			}
-			if (master && info.slave_id === undefined) {
-				return true;
-			}
-			if (info.slave_id && slave_ids.includes(info.slave_id)) {
-				return true;
-			}
-			if (info.instance_id && instance_ids.includes(info.instance_id)) {
-				return true;
-			}
-			return false;
-		};
-	}
-
 	updateLogSubscriptions() {
 		let { all, master, slave_ids, instance_ids } = this.logSubscriptions;
 		if (all || master || slave_ids.length || instance_ids.length) {
@@ -432,7 +407,7 @@ class ControlConnection extends BaseConnection {
 				this.logTransport = new libLoggingUtils.LinkTransport({ link: this });
 				this._master.clusterLogger.add(this.logTransport);
 			}
-			this.logTransport.filter = this.constructor.logFilter(this.logSubscriptions);
+			this.logTransport.filter = libLoggingUtils.logFilter(this.logSubscriptions);
 
 		} else if (this.logTransport) {
 			this._master.clusterLogger.remove(this.logTransport);
@@ -441,34 +416,16 @@ class ControlConnection extends BaseConnection {
 	}
 
 	async queryLogRequestHandler(message) {
-		let transport = new winston.transports.File({ filename: "cluster.log" });
-		let query = util.promisify((options, callback) => transport.query(options, callback));
-
-		// Available options and defaults inferred from reading the source
-		// rows: number = limit || 10
-		// limit: number // alias of rows.
-		// start: number = 0
-		// until: Date = now
-		// from: Date = until - 24h
-		// order: "asc"|"desc" = "desc"
-		// fields: Array<string>
-		// level: string
-		// This interface is junk:
-		// - The level option filters by exact match
-		// - No way to add your own filter
-		// - No way to stream results
-		// - The log file is read starting from the beginning
 		let setDuration = lastQueryLogTime.startTimer();
-		let log = await query({
-			order: message.data.order,
-			limit: Infinity,
-			from: new Date(0),
-		});
+		let { all, master, slave_ids, instance_ids } = message.data;
 
-		log = log.filter(this.constructor.logFilter(message.data));
-		if (message.data.limit) {
-			log = log.slice(0, message.data.limit);
+		let log;
+		if (!all && master && !slave_ids.length && !instance_ids.length) {
+			log = await this._master.queryMasterLog(message.data);
+		} else {
+			log = await this._master.queryClusterLog(message.data);
 		}
+
 		setDuration();
 		return { log };
 	}
diff --git a/packages/master/src/Master.js b/packages/master/src/Master.js
index 5c7de4f0..b5944ac5 100644
--- a/packages/master/src/Master.js
+++ b/packages/master/src/Master.js
@@ -12,6 +12,7 @@ const libConfig = require("@clusterio/lib/config");
 const libErrors = require("@clusterio/lib/errors");
 const libFileOps = require("@clusterio/lib/file_ops");
 const libLink = require("@clusterio/lib/link");
+const libLoggingUtils = require("@clusterio/lib/logging_utils");
 const libPlugin = require("@clusterio/lib/plugin");
 const libPluginLoader = require("@clusterio/lib/plugin_loader");
 const libPrometheus = require("@clusterio/lib/prometheus");
@@ -117,6 +118,14 @@ class Master {
 	}
 
 	async _startInternal(args) {
+		this.logDirectory = args.logDirectory;
+		this.clusterLogIndex = await libLoggingUtils.LogIndex.load(path.join(this.logDirectory, "cluster"));
+		this.clusterLogBuildInterval = setInterval(() => {
+			this.clusterLogIndex.buildIndex().catch(
+				err => logger.error(`Error building cluster log index:\n${err.stack}`)
+			);
+		}, 600e3);
+
 		// Start webpack development server if enabled
 		if (args.dev || args.devPlugin) {
 			logger.warn("Webpack development mode enabled");
@@ -240,6 +249,10 @@ class Master {
 		// This function should never throw.
 		this._state = "stopping";
 		logger.info("Stopping master");
+		clearInterval(this.clusterLogBuildInterval);
+		if (this.clusterLogIndex) {
+			await this.clusterLogIndex.save();
+		}
 
 		logger.info("Saving config");
 		await libFileOps.safeOutputFile(this.configPath, JSON.stringify(this.config.serialize(), null, 4));
@@ -333,6 +346,33 @@ class Master {
 		await libFileOps.safeOutputFile(filePath, JSON.stringify(serialized, null, 4));
 	}
 
+	/**
+	 * Query master log
+	 *
+	 * @param {module:lib/logging_utils~QueryLogFilter} filter -
+	 *     Filter to limit entries with. Note that only the master log can
+	 *     be queried from this function.
+	 * @returns {Array<Object>} log entries matching the filter
+	 */
+	async queryMasterLog(filter) {
+		return await libLoggingUtils.queryLog(
+			path.join(this.logDirectory, "master"), filter,
+		);
+	}
+
+	/**
+	 * Query cluster log
+	 *
+	 * @param {module:lib/logging_utils~QueryLogFilter} filter -
+	 *     Filter to limit entries with.
+	 * @returns {Array<Object>} log entries matching the filter
+	 */
+	async queryClusterLog(filter) {
+		return await libLoggingUtils.queryLog(
+			path.join(this.logDirectory, "cluster"), filter, this.clusterLogIndex,
+		);
+	}
+
 	static addAppRoutes(app, pluginInfos) {
 		app.use((req, res, next) => {
 			hitCounter.inc();
diff --git a/test/file/log/excluded.txt b/test/file/log/excluded.txt
new file mode 100644
index 00000000..c6fcef91
--- /dev/null
+++ b/test/file/log/excluded.txt
@@ -0,0 +1 @@
+This file should be excluded by queries and the index
diff --git a/test/file/log/test-2021-12-10.log b/test/file/log/test-2021-12-10.log
new file mode 100644
index 00000000..e3370ff6
--- /dev/null
+++ b/test/file/log/test-2021-12-10.log
@@ -0,0 +1,9 @@
+{"timestamp":"2021-12-10T10:00:00.000Z","level":"verbose","message":"Master start"}
+{"timestamp":"2021-12-10T10:00:01.000Z","level":"verbose","message":"Loading"}
+{"timestamp":"2021-12-10T10:00:03.000Z","level":"info","message":"Ready"}
+{"timestamp":"2021-12-10T10:00:10.000Z","slave_id":1,"level":"verbose","message":"Slave connected"}
+{"timestamp":"2021-12-10T10:00:11.000Z","slave_id":1,"level":"verbose","message":"Found instance 10"}
+{"timestamp":"2021-12-10T10:00:13.000Z","slave_id":1,"instance_id":10,"level":"server","message":"Starting up"}
+{"timestamp":"2021-12-10T10:00:14.000Z","slave_id":1,"instance_id":10,"level":"server","message":"Loading map"}
+{"timestamp":"2021-12-10T10:00:15.000Z","slave_id":1,"instance_id":10,"level":"server","message":"Game ready"}
+{"timestamp":"2021-12-10T10:01:00.000Z","slave_id":1,"instance_id":10,"level":"error","message":"Unexpected crash"}
diff --git a/test/file/log/test-2021-12-11.log b/test/file/log/test-2021-12-11.log
new file mode 100644
index 00000000..c09dbd63
--- /dev/null
+++ b/test/file/log/test-2021-12-11.log
@@ -0,0 +1,3 @@
+{"timestamp":"2021-12-11T10:00:00.000Z","level":"verbose","message":"Master start"}
+{"timestamp":"2021-12-11T10:00:01.000Z","level":"verbose","message":"Loading"}
+{"timestamp":"2021-12-11T10:00:03.000Z","level":"fatal","message":"An error occured"}
diff --git a/test/file/log/test-2021-12-12.log b/test/file/log/test-2021-12-12.log
new file mode 100644
index 00000000..61bbe1b0
--- /dev/null
+++ b/test/file/log/test-2021-12-12.log
@@ -0,0 +1 @@
+{"timestamp":"2021-12-12T10:00:00.000Z","leve
diff --git a/test/file/log/test-2100-12-31.log b/test/file/log/test-2100-12-31.log
new file mode 100644
index 00000000..e69de29b
diff --git a/test/lib/logging_utils.js b/test/lib/logging_utils.js
index 9d3c5809..bc0afeb1 100644
--- a/test/lib/logging_utils.js
+++ b/test/lib/logging_utils.js
@@ -1,8 +1,13 @@
 "use strict";
 const assert = require("assert").strict;
+const fs = require("fs-extra");
+const path = require("path");
+const util = require("util");
+
 
 const { testLines } = require("./factorio/lines");
 const libLoggingUtils = require("@clusterio/lib/logging_utils");
+const libStream = require("@clusterio/lib/stream");
 
 
 describe("lib/logging_utils.js", function() {
@@ -16,4 +21,110 @@ describe("lib/logging_utils.js", function() {
 			}
 		});
 	});
+
+	let logDir = path.join("test", "file", "log");
+	describe("class LogIndex", function() {
+		it("should create a new empty index if index.json does not exist", async function() {
+			let logIndex = await libLoggingUtils.LogIndex.load(logDir);
+			assert.deepEqual(logIndex.index, new Map());
+		});
+		it("should build index of only past log files", async function() {
+			let logIndex = await libLoggingUtils.LogIndex.load(logDir);
+			await logIndex.buildIndex();
+			assert.deepEqual(
+				new Set(logIndex.index.keys()),
+				new Set(["test-2021-12-10.log", "test-2021-12-11.log", "test-2021-12-12.log"])
+			);
+		});
+		it("should roundtrip index on save load cycle", async function() {
+			let logIndex = await libLoggingUtils.LogIndex.load(logDir);
+			await logIndex.buildIndex();
+			let tempDir = path.join("temp", "test", "log");
+			logIndex.logDirectory = tempDir;
+			await logIndex.save();
+			let loadedIndex = await libLoggingUtils.LogIndex.load(tempDir);
+			assert.deepEqual(
+				loadedIndex,
+				logIndex
+			);
+		});
+		it("should handle loading broken index", async function() {
+			let tempDir = path.join("temp", "test", "log");
+			await fs.outputFile(path.join(tempDir, "index.json"), "Broken JSON");
+			let logIndex = await libLoggingUtils.LogIndex.load(tempDir);
+			assert.deepEqual(logIndex.index, new Map());
+		});
+	});
+
+	let allLines = [];
+	let logIndex = null;
+	before(async function() {
+		for (let file of (await fs.readdir(logDir)).sort()) {
+			if (file === "excluded.txt") {
+				continue;
+			}
+			let fileStream = fs.createReadStream(path.join(logDir, file));
+			let lineStream = new libStream.LineSplitter({ readableObjectMode: true });
+			fileStream.pipe(lineStream);
+			for await (let line of lineStream) {
+				try {
+					allLines.push(JSON.parse(line));
+				} catch (err) {
+					allLines.push({ "level": "info", "message": line.toString() });
+				}
+			}
+		}
+
+		logIndex = await libLoggingUtils.LogIndex.load(logDir);
+		await logIndex.buildIndex();
+	});
+
+	describe("queryLog", function() {
+		for (let index of [() => undefined, () => logIndex]) {
+			describe(index() === undefined ? "without index" : "with index", function() {
+				it("returns nothing with an empty filter", async function() {
+					let log = await libLoggingUtils.queryLog(logDir, {}, index());
+					assert.deepEqual(log, []);
+				});
+				it("should return the whole log when querying all", async function() {
+					let log = await libLoggingUtils.queryLog(logDir, { all: true }, index());
+					assert.deepEqual(log, allLines);
+				});
+				it("should return the whole log in reverse when querying all desc", async function() {
+					let log = await libLoggingUtils.queryLog(logDir, { all: true, order: "desc" }, index());
+					assert.deepEqual(log, [...allLines].reverse());
+				});
+				it("should limit by limit", async function() {
+					let log = await libLoggingUtils.queryLog(logDir, { limit: 2, all: true }, index());
+					assert.deepEqual(log, allLines.slice(0, 2));
+				});
+				it("should filter by max_level", async function() {
+					let log = await libLoggingUtils.queryLog(logDir, { max_level: "info", all: true }, index());
+					assert.deepEqual(log, allLines.filter(
+						info => ["fatal", "error", "warn", "audit", "info"].includes(info.level)
+					));
+					log = await libLoggingUtils.queryLog(logDir, { max_level: "fatal", all: true }, index());
+					assert.deepEqual(log, allLines.filter(
+						info => ["fatal"].includes(info.level)
+					));
+				});
+				it("should filter by master", async function() {
+					let log = await libLoggingUtils.queryLog(logDir, { master: true }, index());
+					assert.deepEqual(log, allLines.filter(
+						info => info.slave_id === undefined && info.instance_id === undefined
+					));
+				});
+				it("should filter by slave_ids", async function() {
+					let log = await libLoggingUtils.queryLog(logDir, { slave_ids: [1, 2] }, index());
+					assert.deepEqual(log, allLines.filter(
+						info => [1, 2].includes(info.slave_id) && info.instance_id === undefined
+					));
+				});
+				it("should filter by instance_ids", async function() {
+					let log = await libLoggingUtils.queryLog(logDir, { instance_ids: [10, 11] }, index());
+					assert.deepEqual(log, allLines.filter(info => [10, 11].includes(info.instance_id)));
+				});
+			});
+		}
+	});
 });

From 170a3d3cf79a00f188558eec0ebf38b75c5ea920 Mon Sep 17 00:00:00 2001
From: Hornwitser <github@hornwitser.no>
Date: Sun, 19 Dec 2021 03:21:00 +0100
Subject: [PATCH 6/6] Use stream.finished for detecting end of streams

There are some corner cases with streams that cause them to not emit
finish close or end events which in turn means that code liseting for
these events directly may end up hanging forever.  Use stream.finished
to detect end of streams which will handle these cases correctly.
---
 packages/create/create.js     |  3 ++-
 packages/ctl/ctl.js           | 14 +++++++++-----
 packages/lib/build_mod.js     |  6 +++++-
 packages/lib/logging_utils.js | 14 ++++++++++++--
 packages/slave/slave.js       | 13 ++++++++-----
 test/lib/stream.js            | 26 +++++++++++++++-----------
 6 files changed, 51 insertions(+), 25 deletions(-)

diff --git a/packages/create/create.js b/packages/create/create.js
index b885f780..b72b6a8e 100755
--- a/packages/create/create.js
+++ b/packages/create/create.js
@@ -15,6 +15,7 @@ const yargs = require("yargs");
 const { levels, logger, setLogLevel } = require("./logging");
 let dev = false;
 const scriptExt = process.platform === "win32" ? ".cmd" : "";
+const finished = util.promisify(stream.finished);
 
 
 const availablePlugins = [
@@ -480,7 +481,7 @@ async function downloadLinuxServer() {
 		const writeStream = fs.createWriteStream(tmpArchivePath);
 		res.pipe(writeStream);
 
-		await events.once(res, "end");
+		await finished(writeStream);
 
 		await fs.rename(tmpArchivePath, archivePath);
 		try {
diff --git a/packages/ctl/ctl.js b/packages/ctl/ctl.js
index 1c4ab74d..64e1ff15 100755
--- a/packages/ctl/ctl.js
+++ b/packages/ctl/ctl.js
@@ -16,6 +16,8 @@ const setBlocking = require("set-blocking");
 const phin = require("phin");
 const os = require("os");
 const child_process = require("child_process");
+const stream = require("stream");
+const util = require("util");
 
 const libLink = require("@clusterio/lib/link");
 const libErrors = require("@clusterio/lib/errors");
@@ -30,6 +32,8 @@ const libHelpers = require("@clusterio/lib/helpers");
 const libFactorio = require("@clusterio/lib/factorio");
 const libFileOps = require("@clusterio/lib/file_ops");
 
+const finished = util.promisify(stream.finished);
+
 
 function print(...content) {
 	// eslint-disable-next-line no-console
@@ -640,12 +644,12 @@ instanceCommands.add(new libCommand.Command({
 			stream: true,
 		});
 
-		let stream;
+		let writeStream;
 		let tempFilename = args.save.replace(/(\.zip)?$/, ".tmp.zip");
 		while (true) {
 			try {
-				stream = fs.createWriteStream(tempFilename, { flags: "wx" });
-				await events.once(stream, "open");
+				writeStream = fs.createWriteStream(tempFilename, { flags: "wx" });
+				await events.once(writeStream, "open");
 				break;
 			} catch (err) {
 				if (err.code === "EEXIST") {
@@ -655,8 +659,8 @@ instanceCommands.add(new libCommand.Command({
 				}
 			}
 		}
-		response.pipe(stream);
-		await events.once(stream, "finish");
+		response.pipe(writeStream);
+		await finished(writeStream);
 
 		let filename = await libFileOps.findUnusedName(".", args.save, ".zip");
 		await fs.rename(tempFilename, filename);
diff --git a/packages/lib/build_mod.js b/packages/lib/build_mod.js
index 13211900..3ac716c6 100644
--- a/packages/lib/build_mod.js
+++ b/packages/lib/build_mod.js
@@ -5,8 +5,12 @@ const fs = require("fs-extra");
 const JSZip = require("jszip");
 const klaw = require("klaw");
 const path = require("path");
+const stream = require("stream");
+const util = require("util");
 const yargs = require("yargs");
 
+const finished = util.promisify(stream.finished);
+
 
 async function buildMod(args, info) {
 	if (args.build) {
@@ -42,7 +46,7 @@ async function buildMod(args, info) {
 			let modPath = path.join(args.outputDir, `${modName}.zip`);
 			console.log(`Writing ${modPath}`);
 			let writeStream = zip.generateNodeStream().pipe(fs.createWriteStream(modPath));
-			await events.once(writeStream, "finish");
+			await finished(writeStream);
 
 		} else {
 			let modDir = path.join(args.outputDir, modName);
diff --git a/packages/lib/logging_utils.js b/packages/lib/logging_utils.js
index 6b2d3af4..37e818e7 100644
--- a/packages/lib/logging_utils.js
+++ b/packages/lib/logging_utils.js
@@ -8,6 +8,8 @@
 const events = require("events");
 const fs = require("fs-extra");
 const path = require("path");
+const stream = require("stream");
+const util = require("util");
 const winston = require("winston");
 const Transport = require("winston-transport");
 const { LEVEL, MESSAGE } = require("triple-beam");
@@ -19,6 +21,8 @@ const libFileOps = require("./file_ops");
 const { levels, logger } = require("./logging");
 const libStream = require("./stream");
 
+const finished = util.promisify(stream.finished);
+
 
 /**
  * Format a parsed Factorio message with colors
@@ -435,7 +439,13 @@ async function queryLog(logDirectory, filter, index) {
 			}
 		});
 		fileStream.pipe(lineStream);
-		await events.once(lineStream, "close");
+		try {
+			await finished(lineStream);
+		} catch (err) {
+			if (log.length < filter.limit || err.code !== "ERR_STREAM_PREMATURE_CLOSE") {
+				throw err;
+			}
+		}
 
 		if (log.length >= filter.limit) {
 			break;
@@ -456,7 +466,7 @@ async function migrateLogs(log, directory, pattern) {
 	async function switchOutput(newOutputPath) {
 		if (output) {
 			output.end();
-			await events.once(output, "close");
+			await finished(output);
 		}
 
 		output = fs.createWriteStream(newOutputPath, { flags: "ax" });
diff --git a/packages/slave/slave.js b/packages/slave/slave.js
index f33ae4dc..cbcb0098 100755
--- a/packages/slave/slave.js
+++ b/packages/slave/slave.js
@@ -20,6 +20,7 @@ const events = require("events");
 const pidusage = require("pidusage");
 const setBlocking = require("set-blocking");
 const phin = require("phin");
+const stream = require("stream");
 const util = require("util");
 const version = require("./package").version;
 const winston = require("winston");
@@ -40,6 +41,8 @@ const { ConsoleTransport, levels, logger } = require("@clusterio/lib/logging");
 const libLoggingUtils = require("@clusterio/lib/logging_utils");
 const libHelpers = require("@clusterio/lib/helpers");
 
+const finished = util.promisify(stream.finished);
+
 
 const instanceRconCommandDuration = new libPrometheus.Histogram(
 	"clusterio_instance_rcon_command_duration_seconds",
@@ -1478,11 +1481,11 @@ class Slave extends libLink.Link {
 
 		let savesDir = path.join(instanceInfo.path, "saves");
 		let tempFilename = filename.replace(/(\.zip)?$/, ".tmp.zip");
-		let stream;
+		let writeStream;
 		while (true) {
 			try {
-				stream = fs.createWriteStream(path.join(savesDir, tempFilename), { flags: "wx" });
-				await events.once(stream, "open");
+				writeStream = fs.createWriteStream(path.join(savesDir, tempFilename), { flags: "wx" });
+				await events.once(writeStream, "open");
 				break;
 			} catch (err) {
 				if (err.code === "EEXIST") {
@@ -1492,8 +1495,8 @@ class Slave extends libLink.Link {
 				}
 			}
 		}
-		response.pipe(stream);
-		await events.once(stream, "finish");
+		response.pipe(writeStream);
+		await finished(writeStream);
 
 		filename = await libFileOps.findUnusedName(savesDir, filename, ".zip");
 		await fs.rename(path.join(savesDir, tempFilename), path.join(savesDir, filename));
diff --git a/test/lib/stream.js b/test/lib/stream.js
index 721ed090..f85af99f 100644
--- a/test/lib/stream.js
+++ b/test/lib/stream.js
@@ -3,16 +3,20 @@ const assert = require("assert").strict;
 const events = require("events");
 const fs = require("fs-extra");
 const path = require("path");
+const stream = require("stream");
+const util = require("util");
 
 const libStream = require("@clusterio/lib/stream");
 
+const finished = util.promisify(stream.finished);
+
 
 describe("lib/stream", function() {
 	describe("class LineSplitter", function() {
 		function createSplitter(lines) {
-			let stream = new libStream.LineSplitter({ readableObjectMode: true });
-			stream.on("data", line => lines.push(line.toString("utf-8")));
-			return stream;
+			let lineStream = new libStream.LineSplitter({ readableObjectMode: true });
+			lineStream.on("data", line => lines.push(line.toString("utf-8")));
+			return lineStream;
 		}
 
 		it("should split three lines", function() {
@@ -57,9 +61,9 @@ describe("lib/stream", function() {
 
 	describe("class ReverseLineSplitter", function() {
 		function createReverseSplitter(lines) {
-			let stream = new libStream.ReverseLineSplitter({ readableObjectMode: true });
-			stream.on("data", line => lines.push(line.toString("utf-8")));
-			return stream;
+			let lineStream = new libStream.ReverseLineSplitter({ readableObjectMode: true });
+			lineStream.on("data", line => lines.push(line.toString("utf-8")));
+			return lineStream;
 		}
 
 		it("should split three lines", function() {
@@ -107,16 +111,16 @@ describe("lib/stream", function() {
 			let content = "";
 			for (let i = 0; i < 10; i++) { content += String(i).repeat(10); }
 			await fs.outputFile(path.join("temp", "test", "reverse.txt"), content);
-			let stream = await libStream.createReverseReadStream(
+			let reverseStream = await libStream.createReverseReadStream(
 				path.join("temp", "test", "reverse.txt"),
 				{ encoding: "utf8", highWaterMark: 10 }
 			);
 			let index = 9;
-			stream.on("data", data => {
+			reverseStream.on("data", data => {
 				assert.equal(data, String(index).repeat(10));
 				index -= 1;
 			});
-			await events.once(stream, "end");
+			await finished(reverseStream);
 		});
 
 		it("should reverse the lines of a file", async function() {
@@ -125,7 +129,7 @@ describe("lib/stream", function() {
 				ws.write(Buffer.from(`${i}\n`));
 			}
 			ws.end();
-			await events.once(ws, "finish");
+			await finished(ws);
 
 			let lineStream = new libStream.ReverseLineSplitter({ readableObjectMode: true });
 			let index = 99999;
@@ -139,7 +143,7 @@ describe("lib/stream", function() {
 				index -= 1;
 			});
 			rs.pipe(lineStream);
-			await events.once(lineStream, "end");
+			await finished(lineStream);
 
 			assert.equal(index, 0, "line check did not reach the end");
 		});