Merge pull request #445 from Hornwitser/log-fix

Fix logs

.gitignore

@@ -1,7 +1,6 @@
 # Logs
 logs
-*.log
-npm-debug.log*
+*debug.log*
 
 # Runtime data
 pids

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 {

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);

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);

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,
 };

packages/lib/logging_utils.js

@@ -1,4 +1,15 @@
+/**
+ * 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");
+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");
@@ -6,6 +17,11 @@ 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");
+
+const finished = util.promisify(stream.finished);
 
 
 /**
@@ -55,6 +71,7 @@ function formatServerOutput(parsed) {
 
 /**
  * Formats winston log messages for a character terminal.
+ * @static
  */
 class TerminalFormat {
 	constructor(options = {}) {
@@ -95,6 +112,7 @@ class TerminalFormat {
 
 /**
  * Sends logs over a lib/link connection.
+ * @static
  */
 class LinkTransport extends Transport {
 	constructor(options) {
@@ -121,10 +139,385 @@ 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);
+		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;
+		}
+	}
+	return log;
+}
+
+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 finished(output);
+		}
+
+		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,
+	LogIndex,
+	logFilter,
+	queryLog,
+	migrateLogs,
 
 	// for testing only
 	_formatServerOutput: formatServerOutput,

packages/lib/stream.js

@@ -0,0 +1,168 @@
+"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();
+	}
+}
+
+/**
+ * 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,
+};

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,

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"
   },

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 };
 	}

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();

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": {

packages/slave/slave.js

@@ -20,9 +20,11 @@ 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");
+require("winston-daily-rotate-file");
 
 // internal libraries
 const libFileOps = require("@clusterio/lib/file_ops");
@@ -39,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",
@@ -1477,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") {
@@ -1491,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));
@@ -1734,6 +1738,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 +1764,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({

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">

test/file/log/excluded.txt

@@ -0,0 +1 @@
+This file should be excluded by queries and the index

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"}

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"}

test/file/log/test-2021-12-12.log

@@ -0,0 +1 @@
+{"timestamp":"2021-12-12T10:00:00.000Z","leve

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);
 	});
 }
 

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) {

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)));
+				});
+			});
+		}
+	});
 });

test/lib/stream.js

@@ -0,0 +1,151 @@
+"use strict";
+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 lineStream = new libStream.LineSplitter({ readableObjectMode: true });
+			lineStream.on("data", line => lines.push(line.toString("utf-8")));
+			return lineStream;
+		}
+
+		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"]);
+		});
+	});
+
+	describe("class ReverseLineSplitter", function() {
+		function createReverseSplitter(lines) {
+			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() {
+			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 reverseStream = await libStream.createReverseReadStream(
+				path.join("temp", "test", "reverse.txt"),
+				{ encoding: "utf8", highWaterMark: 10 }
+			);
+			let index = 9;
+			reverseStream.on("data", data => {
+				assert.equal(data, String(index).repeat(10));
+				index -= 1;
+			});
+			await finished(reverseStream);
+		});
+
+		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 finished(ws);
+
+			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 finished(lineStream);
+
+			assert.equal(index, 0, "line check did not reach the end");
+		});
+	});
+});