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