From 1c8a2fef3267448af34128a391659c2d8a532d76 Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Mon, 30 Nov 2020 02:22:13 +0100 Subject: [PATCH] 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 | 101 ++++++++++++++++++++++--- packages/lib/link/messages.js | 31 ++++++++ packages/lib/logging_utils.js | 9 ++- packages/lib/users.js | 5 ++ packages/master/master.js | 74 +++++++++++++++--- packages/web_ui/src/util/websocket.jsx | 3 + 6 files changed, 201 insertions(+), 22 deletions(-) diff --git a/packages/ctl/ctl.js b/packages/ctl/ctl.js index d9eaf1c8..0658725a 100755 --- a/packages/ctl/ctl.js +++ b/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(); } } diff --git a/packages/lib/link/messages.js b/packages/lib/link/messages.js index 3b1a9b48..1eb28116 100644 --- a/packages/lib/link/messages.js +++ b/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" }, + }, }, }); diff --git a/packages/lib/logging_utils.js b/packages/lib/logging_utils.js index 26a554a8..990b2597 100644 --- a/packages/lib/logging_utils.js +++ b/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; } diff --git a/packages/lib/users.js b/packages/lib/users.js index 81097243..a55ef972 100644 --- a/packages/lib/users.js +++ b/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", diff --git a/packages/master/master.js b/packages/master/master.js index b0cdeebe..caa02c01 100755 --- a/packages/master/master.js +++ b/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 + // 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()) { diff --git a/packages/web_ui/src/util/websocket.jsx b/packages/web_ui/src/util/websocket.jsx index 1805cc8a..f3de0e4d 100644 --- a/packages/web_ui/src/util/websocket.jsx +++ b/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, }); }