Add log inspection commands

Add log query and log follow commands which allows inspection of the
combined cluster log.  Past entries can be queried via the new queryLog
request.  The log is stored into a single flat file for now, though the
performance of this may good enough.

packages/ctl/ctl.js

@@ -188,7 +188,7 @@ instanceCommands.add(new libCommand.Command({
 	}],
 	handler: async function(args, control) {
 		let instanceId = await libCommand.resolveInstance(control, args.instance);
-		await libLink.messages.setLogSubscriptions.send(control, { all: false, instance_ids: [instanceId] });
+		await control.setLogSubscriptions({ instance_ids: [instanceId] });
 		let response = await libLink.messages.createSave.send(control, {
 			instance_id: instanceId,
 		});
@@ -201,7 +201,7 @@ instanceCommands.add(new libCommand.Command({
 	}],
 	handler: async function(args, control) {
 		let instanceId = await libCommand.resolveInstance(control, args.instance);
-		await libLink.messages.setLogSubscriptions.send(control, { all: false, instance_ids: [instanceId] });
+		await control.setLogSubscriptions({ instance_ids: [instanceId] });
 		let response = await libLink.messages.exportData.send(control, {
 			instance_id: instanceId,
 		});
@@ -218,11 +218,12 @@ instanceCommands.add(new libCommand.Command({
 	}],
 	handler: async function(args, control) {
 		let instanceId = await libCommand.resolveInstance(control, args.instance);
-		await libLink.messages.setLogSubscriptions.send(control, { all: false, instance_ids: [instanceId] });
+		await control.setLogSubscriptions({ instance_ids: [instanceId] });
 		let response = await libLink.messages.startInstance.send(control, {
 			instance_id: instanceId,
 			save: args.save || null,
 		});
+		control.keepOpen = args.keepOpen;
 	},
 }));
 
@@ -236,11 +237,12 @@ instanceCommands.add(new libCommand.Command({
 	}],
 	handler: async function(args, control) {
 		let instanceId = await libCommand.resolveInstance(control, args.instance);
-		await libLink.messages.setLogSubscriptions.send(control, { all: false, instance_ids: [instanceId] });
+		await control.setLogSubscriptions({ instance_ids: [instanceId] });
 		let response = await libLink.messages.loadScenario.send(control, {
 			instance_id: instanceId,
 			scenario: args.scenario || null,
 		});
+		control.keepOpen = args.keepOpen;
 	},
 }));
 
@@ -250,7 +252,7 @@ instanceCommands.add(new libCommand.Command({
 	}],
 	handler: async function(args, control) {
 		let instanceId = await libCommand.resolveInstance(control, args.instance);
-		await libLink.messages.setLogSubscriptions.send(control, { all: false, instance_ids: [instanceId] });
+		await control.setLogSubscriptions({ instance_ids: [instanceId] });
 		let response = await libLink.messages.stopInstance.send(control, {
 			instance_id: instanceId,
 		});
@@ -469,12 +471,75 @@ userCommands.add(new libCommand.Command({
 	},
 }));
 
+const logCommands = new libCommand.CommandTree({ name: "log", description: "Log inspection" });
+logCommands.add(new libCommand.Command({
+	definition: ["follow", "follow cluster log", (yargs) => {
+		yargs.options({
+			"all": { describe: "Follow the whole cluster log", nargs: 0, type: "boolean", default: false },
+			"master": { describe: "Follow log of the master server", nargs: 0, type: "boolean", default: false },
+			"slave": { describe: "Follow log of given slave", nargs: 1, type: "string", default: null },
+			"instance": { describe: "Follow log of given instance", nargs: 1, type: "string", default: null },
+		});
+	}],
+	handler: async function(args, control) {
+		if (!args.all && !args.master && !args.slave && !args.instance) {
+			logger.error("At least one of --all, --master, --slave and --instance must be passed");
+			process.exitCode = 1;
+			return;
+		}
+		let instance_ids = args.instance ? [await libCommand.resolveInstance(control, args.instance)] : [];
+		let slave_ids = args.slave ? [await libCommand.resolveSlave(control, args.slave)] : [];
+		await control.setLogSubscriptions({ all: args.all, master: args.master, slave_ids, instance_ids });
+		control.keepOpen = true;
+	},
+}));
+
+logCommands.add(new libCommand.Command({
+	definition: ["query", "Query cluster log", (yargs) => {
+		yargs.options({
+			"all": { describe: "Query the whole cluster log", nargs: 0, type: "boolean", default: false },
+			"master": { describe: "Query log of the master server", nargs: 0, type: "boolean", default: false },
+			"slave": { describe: "Query log of given slave", nargs: 1, type: "string", default: null },
+			"instance": { describe: "Query log of given instance", nargs: 1, type: "string", default: null },
+			"max-level": { describe: "Maximum log level to return", nargs: 1, type: "string", default: null },
+		});
+	}],
+	handler: async function(args, control) {
+		if (!args.all && !args.master && !args.slave && !args.instance) {
+			logger.error("At least one of --all, --master, --slave and --instance must be passed");
+			process.exitCode = 1;
+			return;
+		}
+		let instance_ids = args.instance ? [await libCommand.resolveInstance(control, args.instance)] : [];
+		let slave_ids = args.slave ? [await libCommand.resolveSlave(control, args.slave)] : [];
+		let result = await libLink.messages.queryLog.send(control, {
+			all: args.all,
+			master: args.master,
+			slave_ids,
+			instance_ids,
+			max_level: args.maxLevel,
+		});
+
+		let stdoutLogger = winston.createLogger({
+			level: "verbose",
+			levels,
+			format: new libLoggingUtils.TerminalFormat({ showTimestamp: true }),
+			transports: [
+				new ConsoleTransport({ errorLevels: [], warnLevels: [] }),
+			],
+		});
+		for (let info of result.log) {
+			stdoutLogger.log(info);
+		}
+	},
+}));
+
 const debugCommands = new libCommand.CommandTree({ name: "debug", description: "Debugging utilities" });
 debugCommands.add(new libCommand.Command({
 	definition: ["dump-ws", "Dump WebSocket messages sent and received by master", (yargs) => { }],
 	handler: async function(args, control) {
 		await libLink.messages.debugDumpWs.send(control);
-		return new Promise(() => {});
+		control.keepOpen = true;
 	},
 }));
 
@@ -519,6 +584,24 @@ class Control extends libLink.Link {
 		for (let controlPlugin of controlPlugins.values()) {
 			libPlugin.attachPluginMessages(this, controlPlugin.info, controlPlugin);
 		}
+
+		/**
+		 * Keep the control connection alive after the command completes.
+		 * @type {boolean}
+		 */
+		this.keepOpen = false;
+	}
+
+	async setLogSubscriptions({
+		all = false,
+		master = false,
+		slave_ids = [],
+		instance_ids = [],
+		max_level = null,
+	}) {
+		await libLink.messages.setLogSubscriptions.send(this, {
+			all, master, slave_ids, instance_ids, max_level,
+		});
 	}
 
 	async logMessageEventHandler(message) {
@@ -608,6 +691,7 @@ async function startControl() {
 	rootCommands.add(permissionCommands);
 	rootCommands.add(roleCommands);
 	rootCommands.add(userCommands);
+	rootCommands.add(logCommands);
 	rootCommands.add(debugCommands);
 
 	logger.verbose("Loading Plugin info");
@@ -704,12 +788,11 @@ async function startControl() {
 		});
 	});
 
-	let keepOpen = Boolean(args.keepOpen);
 	try {
 		await targetCommand.run(args, control);
 
 	} catch (err) {
-		keepOpen = false;
+		control.keepOpen = false;
 		if (err instanceof libErrors.CommandError) {
 			logger.error(`Error running command: ${err.message}`);
 			process.exitCode = 1;
@@ -723,7 +806,7 @@ async function startControl() {
 		}
 
 	} finally {
-		if (!keepOpen) {
+		if (!control.keepOpen) {
 			await control.shutdown();
 		}
 	}

packages/lib/link/messages.js

@@ -641,10 +641,41 @@ messages.setLogSubscriptions = new Request({
 	permission: "core.log.follow",
 	requestProperties: {
 		"all": { type: "boolean" },
+		"master": { type: "boolean" },
+		"slave_ids": {
+			type: "array",
+			items: { type: "integer" },
+		},
 		"instance_ids": {
 			type: "array",
 			items: { type: "integer" },
 		},
+		"max_level": { type: ["string", "null"] },
+	},
+});
+
+messages.queryLog = new Request({
+	type: "query_log",
+	links: ["control-master"],
+	permission: "core.log.query",
+	requestProperties: {
+		"all": { type: "boolean" },
+		"master": { type: "boolean" },
+		"slave_ids": {
+			type: "array",
+			items: { type: "integer" },
+		},
+		"instance_ids": {
+			type: "array",
+			items: { type: "integer" },
+		},
+		"max_level": { type: ["string", "null"] },
+	},
+	responseProperties: {
+		"log": {
+			type: "array",
+			items: { type: "object" },
+		},
 	},
 });
 

packages/lib/logging_utils.js

@@ -64,6 +64,11 @@ class TerminalFormat {
 
 	transform(info, options) {
 		info = this.colorize.transform(info, this.colorize.options);
+		let ts = "";
+		if (options.showTimestamp && info.timestamp) {
+			ts = `${info.timestamp.replace("T", " ")} `;
+		}
+
 		let src = " ";
 		if (info.instance_name) {
 			src += `i:${info.instance_name} `;
@@ -80,9 +85,9 @@ class TerminalFormat {
 		}
 
 		if (info.stack) {
-			info[MESSAGE] = `[${info.level}]${src}${info.stack}`;
+			info[MESSAGE] = `${ts}[${info.level}]${src}${info.stack}`;
 		} else {
-			info[MESSAGE] = `[${info.level}]${src}${info.message}`;
+			info[MESSAGE] = `${ts}[${info.level}]${src}${info.message}`;
 		}
 		return info;
 	}

packages/lib/users.js

@@ -450,6 +450,11 @@ definePermission({
 	title: "Follow cluster log",
 	description: "Receive new entries in the cluster log.  Required to see instance console.",
 });
+definePermission({
+	name: "core.log.query",
+	title: "Query cluster log",
+	description: "Query past entries in the cluster log.  Required to see past entries in instance console.",
+});
 
 definePermission({
 	name: "core.debug.dump_ws",

packages/master/master.js

@@ -562,6 +562,11 @@ class BaseConnection extends libLink.Link {
 	}
 }
 
+const lastQueryLogTime = new libPrometheus.Gauge(
+	"clusterio_master_last_query_log_duration_seconds",
+	"Time in seconds the last log query took to execute."
+);
+
 let controlConnections = [];
 class ControlConnection extends BaseConnection {
 	constructor(registerData, connector, user) {
@@ -586,6 +591,8 @@ class ControlConnection extends BaseConnection {
 		this.logTransport = null;
 		this.logSubscriptions = {
 			all: false,
+			master: false,
+			slave_ids: [],
 			instance_ids: [],
 		};
 
@@ -780,22 +787,37 @@ 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() {
-		if (this.logSubscriptions.all || this.logSubscriptions.instance_ids.length) {
+		let { all, master, slave_ids, instance_ids } = this.logSubscriptions;
+		if (all || master || slave_ids.length || instance_ids.length) {
 			if (!this.logTransport) {
 				this.logTransport = new libLoggingUtils.LinkTransport({ link: this });
 				clusterLogger.add(this.logTransport);
 			}
-			if (this.logSubscriptions.all) {
-				this.logTransport.filter = null;
-			} else {
-				this.logTransport.filter = info => {
-					if (info.instance_id && this.logSubscriptions.instance_ids.includes(info.instance_id)) {
-						return true;
-					}
-					return false;
-				};
-			}
+			this.logTransport.filter = this.constructor.logFilter(this.logSubscriptions);
 
 		} else if (this.logTransport) {
 			clusterLogger.remove(this.logTransport);
@@ -803,6 +825,36 @@ 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: "asc",
+			limit: Infinity,
+			from: new Date(0),
+		});
+
+		log = log.filter(this.constructor.logFilter(message.data));
+		setDuration();
+		return { log };
+	}
+
 	async listPermissionsRequestHandler(message) {
 		let list = [];
 		for (let permission of libUsers.permissions.values()) {

packages/web_ui/src/util/websocket.jsx

@@ -107,7 +107,10 @@ export class Control extends libLink.Link {
 	async updateLogSubscriptions() {
 		await libLink.messages.setLogSubscriptions.send(this, {
 			all: false,
+			master: false,
+			slave_ids: [],
 			instance_ids: [...this.instanceLogHandlers.keys()],
+			max_level: null,
 		});
 	}