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/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/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/logging_utils.js b/packages/lib/logging_utils.js index c80b6997..37e818e7 100644 --- a/packages/lib/logging_utils.js +++ b/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} [slave_ids] - + * Include log entries for the given slaves and instances of those + * slaves by id. + * @property {Array} [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} [slave_ids] - + * Include log entries for the given slaves and instances of those + * slaves by id. + * @property {Array} [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} 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, diff --git a/packages/lib/stream.js b/packages/lib/stream.js new file mode 100644 index 00000000..98787397 --- /dev/null +++ b/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, +}; 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/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 - // 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} 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} 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/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..cbcb0098 100755 --- a/packages/slave/slave.js +++ b/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({ 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 <> 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/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/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))); + }); + }); + } + }); }); diff --git a/test/lib/stream.js b/test/lib/stream.js new file mode 100644 index 00000000..f85af99f --- /dev/null +++ b/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"); + }); + }); +});