diff --git a/.eslintrc.js b/.eslintrc.js index 19bddc01..0dd888b7 100644 --- a/.eslintrc.js +++ b/.eslintrc.js @@ -214,7 +214,7 @@ module.exports = { "no-useless-rename": "error", "no-useless-return": "error", "no-var": "warn", - "no-void": "error", + "no-void": ["error", { "allowAsStatement": true }], "no-warning-comments": "off", "no-whitespace-before-property": "error", "nonblock-statement-body-position": "error", diff --git a/CHANGELOG.md b/CHANGELOG.md index 049f024a..3ded288a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -47,6 +47,8 @@ Version 2.0.0 - Added synchronization of in-game adminlist, banlist and whitelist to Clusterio core. Previously this was handled by the playerManager plugin. - Added stripping of long paths in the Factorio server log. +- Added option to configure maximum number of commands to send in parallel. +- Added option to auto start instances on slave startup. ### Changes diff --git a/docs/configuration.md b/docs/configuration.md index 2128d6b6..e69d983e 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -218,6 +218,15 @@ directory for the instance is created on the slave. Defaults to null meaning not assigned. +### instance.auto_start + +If enabled start this instance when the slave it is assigned to is +started up. Does not affect start or stop of the instance while the +slave is running. + +Defaults to false + + ### factorio.version Version of Factorio to use for this instance. Can be a version like @@ -342,6 +351,39 @@ banned status while the instance is running. Defaults to true +### factorio.max_concurrent_commands + +Maximum number of commands being transmitted into the game in parallel. +Since the rate at which long commands are streamed into the game is by +default limeted to 100 bytes/tick and scales down to 25 bytes/tick when +there's 20 players connected, long commands can hold up the command +interface and cause large latencies in getting data pushed into the +game. (The rates are controlled by the `segment_size` options in the +server settings.) To counteract this Clusterio sends up to this +configured value number of commands in parallel over the RCON interface. +This causes the game updates sent to clients to have this many command +streams in parallel contained in them, drastically increasing the +maxiumum size they can become. Since commands are only processed so +ofter the maximum rate of commands that can be sustained is roughly 30 +times this configured value per second, though note that large commands +will lower this value. + +The game updates sent by the server is split into roughly 500 byte +packets, and if any one of those parts are lost the entire update is +resent. This means a high number of concurrent commands can amplify +resends due to packet loss and degrade the player experience up to the +point of becomming completely unplayable. + +Prior to 0.18.7 the maximum practical size of game update sent where +arround 8 kB due to the UDP receive buffer on windows defaulting to this +value, larger updates would not get through the receive buffer and cause +the connection to drop. To stay below the 8 kB limit the maximum safe +value for this option would be around 7000 / (3 * game_speed * +maximum_segment_size), which is around 20 in normal circumstances. + +Defaults to 5 + + ### .enabled Whether to enable the given plugin on the instance. Note that plugins diff --git a/docs/writing-plugins.md b/docs/writing-plugins.md index e3a00183..367922f1 100644 --- a/docs/writing-plugins.md +++ b/docs/writing-plugins.md @@ -230,11 +230,11 @@ Communicating with Factorio For pushing data into Factorio there's RCON, which lets you send arbitrary Lua commands to invoke whatever code you want in the game. -This is done by calling the `sendRcon` method on the instance's server object. +This is done by calling the `sendRcon` method on the plugin object. For example: async onStart() { - let response = await this.instance.server.sendRcon( + let response = await this.sendRcon( "/sc rcon.print('data')" ); diff --git a/packages/ctl/ctl.js b/packages/ctl/ctl.js index 9e9e27d5..15241642 100755 --- a/packages/ctl/ctl.js +++ b/packages/ctl/ctl.js @@ -203,13 +203,13 @@ instanceConfigCommands.add(new libCommand.Command({ instanceCommands.add(instanceConfigCommands); instanceCommands.add(new libCommand.Command({ - definition: ["assign ", "Assign instance to a slave", (yargs) => { + definition: ["assign [slave]", "Assign instance to a slave", (yargs) => { yargs.positional("instance", { describe: "Instance to assign", type: "string" }); - yargs.positional("slave", { describe: "Slave to assign to", type: "string" }); + yargs.positional("slave", { describe: "Slave to assign to or unassign if none", type: "string" }); }], handler: async function(args, control) { let instanceId = await libCommand.resolveInstance(control, args.instance); - let slaveId = await libCommand.resolveSlave(control, args.slave); + let slaveId = args.slave ? await libCommand.resolveSlave(control, args.slave) : null; await libLink.messages.assignInstanceCommand.send(control, { instance_id: instanceId, slave_id: slaveId, diff --git a/packages/lib/config/definitions.js b/packages/lib/config/definitions.js index cc273315..d15406ce 100644 --- a/packages/lib/config/definitions.js +++ b/packages/lib/config/definitions.js @@ -196,6 +196,12 @@ InstanceGroup.define({ type: "number", optional: true, }); +InstanceGroup.define({ + name: "auto_start", + description: "Automatically start this instance when the slave hosting it is started up", + type: "boolean", + initial_value: false, +}); InstanceGroup.finalize(); /** @@ -280,6 +286,12 @@ FactorioGroup.define({ type: "boolean", initial_value: true, }); +FactorioGroup.define({ + name: "max_concurrent_commands", + description: "Maximum number of RCON commands trasmitted in parallel", + type: "number", + initial_value: 5, +}); FactorioGroup.finalize(); /** diff --git a/packages/lib/factorio/server.js b/packages/lib/factorio/server.js index 49da075a..17423007 100644 --- a/packages/lib/factorio/server.js +++ b/packages/lib/factorio/server.js @@ -518,6 +518,9 @@ class FactorioServer extends events.EventEmitter { * @param {boolean} options.enableWhitelist - Turn on whitelisting. * @param {boolean} options.verboseLogging - Enable verbose logging. * @param {boolean} options.stripPaths - Strip paths in the console. + * @param {number} options.maxConcurrentCommands - + * Maximum number of RCON commands transmitted in parallel on the + * RCON connection. */ constructor(factorioDir, writeDir, options) { super(); @@ -540,6 +543,8 @@ class FactorioServer extends events.EventEmitter { this.enableWhitelist = options.enableWhitelist; /** Enable verbose logging */ this.verboseLogging = options.verboseLogging; + /** Maximum number of RCON commands transmitted in parallel on the RCON connection */ + this.maxConcurrentCommands = options.maxConcurrentCommands || 5; this._state = "new"; this._server = null; this._rconClient = null; @@ -689,7 +694,7 @@ class FactorioServer extends events.EventEmitter { port: this.rconPort, password: this.rconPassword, timeout: 200000, // 200s, should allow for commands up to 1250kB in length - maxPending: 5, + maxPending: this.maxConcurrentCommands, }; this._rconClient = new rconClient.Rcon(config); @@ -700,9 +705,31 @@ class FactorioServer extends events.EventEmitter { this._rconReady = false; this._startRcon().catch(() => { /* Ignore */ }); }); + // XXX: Workaround to suppress bogus event listener warning + if (this._rconClient.emitter instanceof events.EventEmitter) { + this._rconClient.emitter.setMaxListeners(this.maxConcurrentCommands + 5); + } await this._rconClient.connect(); } + get maxConcurrentCommands() { + return this._maxConcurrentCommands; + } + + set maxConcurrentCommands(value) { + this._maxConcurrentCommands = value; + this.setMaxListeners(value + 5); + if (this._rconClient) { + if (this._rconClient.sendQueue) { + this._rconClient.sendQueue.maxConcurrent = value; + } + // XXX: Workaround to suppress bogus event listener warning + if (this._rconClient.emitter instanceof events.EventEmitter) { + this._rconClient.emitter.setMaxListeners(value + 5); + } + } + } + /** * Initialize class instance * diff --git a/packages/lib/helpers.js b/packages/lib/helpers.js index a31322ba..cf1a657c 100644 --- a/packages/lib/helpers.js +++ b/packages/lib/helpers.js @@ -19,6 +19,41 @@ function basicType(value) { return typeof value; } + +/** + * Asynchronously wait for the given duration + * + * @param {number} duration - Time to wait for in milliseconds. + */ +async function wait(duration) { + await new Promise(resolve => { setTimeout(resolve, duration); }); +} + + +/** + * Resolve a promise with a timeout. + * + * @param {Promise} promise - Promise to wait for. + * @param {number} time - Maximum time im milliseconds to wait for. + * @param {*=} timeoutResult - Value to return if the operation timed out. + */ +async function timeout(promise, time, timeoutResult) { + let timer; + try { + return await Promise.race([ + promise, + new Promise(resolve => { + timer = setTimeout(() => resolve(timeoutResult), time); + }), + ]); + + } finally { + clearTimeout(timer); + } +} + module.exports = { basicType, + wait, + timeout, }; diff --git a/packages/lib/link/messages.js b/packages/lib/link/messages.js index 36454012..b9e011eb 100644 --- a/packages/lib/link/messages.js +++ b/packages/lib/link/messages.js @@ -411,19 +411,10 @@ messages.assignInstanceCommand = new Request({ permission: "core.instance.assign", requestProperties: { "instance_id": { type: "integer" }, - "slave_id": { type: "integer" }, + "slave_id": { type: ["integer", "null"] }, }, }); -messages.assignInstance = new Request({ - type: "assign_instance", - links: ["master-slave"], - requestProperties: { - "serialized_config": { type: "object" }, - }, - forwardTo: "instance", -}); - messages.startInstance = new Request({ type: "start_instance", links: ["control-master", "master-slave", "slave-instance"], @@ -655,6 +646,44 @@ messages.deleteUser = new Request({ }, }); + +// Internal requests +messages.updateInstances = new Request({ + type: "update_instances", + links: ["slave-master"], + requestProperties: { + "instances": { + type: "array", + items: { + type: "object", + additionalProperties: false, + required: ["serialized_config", "status"], + properties: { + "serialized_config": { type: "object" }, + "status": { enum: [ + "stopped", "starting", "running", "creating_save", "exporting_data", + ]}, + }, + }, + }, + }, +}); + +messages.assignInstance = new Request({ + type: "assign_instance", + links: ["master-slave"], + requestProperties: { + "serialized_config": { type: "object" }, + }, + forwardTo: "instance", +}); + +messages.unassignInstance = new Request({ + type: "unassigne_instance", + links: ["master-slave"], + forwardTo: "instance", +}); + messages.getMetrics = new Request({ type: "get_metrics", links: ["master-slave", "slave-instance"], @@ -846,27 +875,6 @@ messages.instanceStatusChanged = new Event({ }, }); -messages.updateInstances = new Event({ - type: "update_instances", - links: ["slave-master"], - eventProperties: { - "instances": { - type: "array", - items: { - type: "object", - additionalProperties: false, - required: ["serialized_config", "status"], - properties: { - "serialized_config": { type: "object" }, - "status": { enum: [ - "stopped", "starting", "running", "creating_save", "exporting_data", - ]}, - }, - }, - }, - }, -}); - messages.instanceOutput = new Event({ type: "instance_output", links: ["instance-slave", "slave-master", "master-control"], diff --git a/packages/lib/plugin.js b/packages/lib/plugin.js index f5447f3a..3a461fea 100644 --- a/packages/lib/plugin.js +++ b/packages/lib/plugin.js @@ -192,6 +192,10 @@ class BaseInstancePlugin { * @param {string} event.name - Name of the player that joined/left. */ async onPlayerEvent(event) { } + + async sendRcon(message, expectEmpty) { + return await this.instance.sendRcon(message, expectEmpty, this.info.name); + } } /** diff --git a/packages/lib/prometheus.js b/packages/lib/prometheus.js index a9f4650a..aaf34c11 100644 --- a/packages/lib/prometheus.js +++ b/packages/lib/prometheus.js @@ -1,6 +1,122 @@ +/** + * Asynchronous Prometheus client + * + * Prometheus client for exposing metrics to a Prometheus server based on + * (possibly) asynchronous callbacks. This library was developed as a + * replacement for prom-client in Clusterio due to the lack of callbacks on + * metric collections in prom-client. + * + * The ordinary use case of instrumenting a code base should be covered by + * {@link module:lib/prometheus.Counter}, {@link + * module:lib/prometheus.Gauge}, {@link module:lib/prometheus.Histogram} and + * {@link module:lib/prometheus.exposition}. Note that the summary + * collector is not implemented. See documentation for each of the listed + * interfaces for more information. + * + * @example + * const { Counter, exposition } = require("@clusterio/lib/prometheus"); + * + * // Collectors are by default registered to the default collector regitry. + * const totalRequests = new Counter( + * "app_request_count_total", + * "Total requests handled.", + * ); + * + * function handleRequest(...) { + * // Do stuff + * totalRequests.inc(); + * } + * + * // Code handling the /metrics HTTP request, here shown for express + * // but any http framework may be used + * const app = require("express")(); + * async function getMetrics(req, res) { + * // By default exposition uses the default collector registry + * let text = await exposition(); + * res.set("Content-Type", exposition.contentType); + * res.send(text); + * } + * app.get("/metrics", (req, res, next) => getMetrics(req, res).catch(next)); + * app.listen(9100); + * @module lib/prometheus + */ "use strict"; + +/** + * Result from collecting a {@link module:lib/prometheus.Collector} + * @typedef {Object} CollectorResult + * @property {module:lib/prometheus.Metric} metric - + * Metric collected. + * @property {Map>} samples - + * Mapping of metric suffix to mapping of label keys to values + * collected. For normal metrics the first level contains a single + * entry under the empty string as key. If the metric does not have + * labels the second level also contains a single entry under the empty + * string as a key. + * + * @example Simple Metric + * let result = { + * metric: new Metric("count", "simple_total_count", "A simple counter"), + * samples: new Map([ + * ["", new Map([ + * ["", 123], + * ])], + * ]), + * } + * + * @example Labeled Metric + * let result = { + * metric: new Metric( + * "count", "labeled_total_count", "A labeled counter", ["a", "b"] + * ), + * samples: new Map([ + * ["", new Map([ + * ['a="1",b="3"', 123], + * ['a="2",b="1"', 34], + * ['a="2",b="9"', 7], + * ])], + * ]), + * } + * + * @example Histogram Metric + * let result = { + * metric: new Metric( + * "histogram", "histogram_size", "A histogram of sizes" + * ), + * samples: new Map([ + * ["_bucket", new Map([ + * ['le="1"', 1], + * ['le="5"', 4], + * ['le="+Inf"', 5], + * ])], + * ["_sum", new Map([ + * ["", 48], + * ])], + * ["_count", new Map([ + * ["", 5], + * ])], + * ]), + * } + */ + +/** + * Represents a collectable metric + * + * Used in implementing collectors in order to validate the name, help text + * and labels attached to the collector. + * @static + */ class Metric { - constructor(type, name, help, labels = []) { + /** + * @param {string} type - + * Metric type, should be one of `counter`, `gauge`, `histogram`, + * `summary` or `untyped`, + * @param {string} name - Name of the metric. + * @param {string} help - Help text for the metric. + * @param {Array} labels - + * @param {Array} _reserved_labels - + */ + constructor(type, name, help, labels = [], _reserved_labels = []) { if (typeof name !== "string") { throw new Error("Expected name to be a string"); } @@ -16,29 +132,92 @@ class Metric { for (let label of labels) { if (!/^[a-zA-Z_][a-zA-Z0-9_]*$/.test(label)) { throw new Error(`Invalid label '${label}'`); + } else if (_reserved_labels.includes(label)) { + throw new Error(`Reserved label '${label}'`); } } + /** + * Name of the metric + * @type {string} + */ this.name = name; + + /** + * Help text for the metric + * @type {string} + */ this.help = help; + + /** + * Metric type, should be one of `counter`, `gauge`, `histogram`, + * `summary` or `untyped`. + * @type {string} + */ this.type = type; + + /** + * Labels for this metric. + * @type {Array} + */ this.labels = labels; } } +/** + * The default registry which collectors are regististered to. + * @type {module:lib/prometheus.CollectorRegistry} + * @static + */ let defaultRegistry; +/** + * Base class for all collectors + * + * This servers mostly as a conceptual base for all collectors, the only + * feature implemented here is registring to the default registry on + * construction. If you want to implement a custom collector you most + * likely want to base it on {@link module:lib/prometheus.LabeledCollector} + * instead. + * + * @static + */ class Collector { + /** + * Create collector + * + * @param {boolean=} register - + * Whether to register this collector to the default registry. + */ constructor(register = true) { if (register) { defaultRegistry.register(this); } } + /** + * Retrieve metric data from this collctor. + * + * Called by {@link module:lib/prometheus.CollectorRegistry} to gather + * the metric data this Collector exports. + * + * @returns {*} Async iterator of {@link CollectorResult}. + */ async* collect() { } } +const labelEscapesChars = /[\\\n\"]/; +/** + * Escapes a label value in accordance with the text exposition format + * + * @param {string} value - Value to escape. + * @returns {string} escaped value with \, ", and newline escaped. + * @private + */ function escapeLabelValue(value) { + if (!labelEscapesChars.test(value)) { + return value; + } return value .replace(/\\/g, "\\\\") .replace(/\n/g, "\\n") @@ -46,6 +225,15 @@ function escapeLabelValue(value) { ; } +/** + * Convert label expression to unique string + * + * @param {Array<(string|Object)>} labels - + * label values to compute key for. + * @param {Array} metricLabels - labels defined for the metric. + * @returns {string} computed key + * @private + */ function labelsToKey(labels, metricLabels) { let items = []; if (labels.length === 1 && typeof labels[0] === "object") { @@ -87,6 +275,10 @@ function labelsToKey(labels, metricLabels) { } } + if (items.length === 1) { + return items[0]; + } + return items.join(","); } @@ -120,14 +312,50 @@ function removeMatchingLabels(mapping, labels) { } } -class ValueCollector extends Collector { +/** + * Base class for implementing labeled collectors + * + * Provides the scaffolding for creating labled collectors where each label + * set is a child to the collector that can be retrieved with the .labels() + * method. + * + * @extends module:lib/prometheus.Collector + * @static + */ +class LabeledCollector extends Collector { + /** + * Create optionally labeled collector + * + * @param {string} type - Type for metric. + * @param {string} name - Name of the metric. + * @param {string} help - Help text for metric. + * @param {Object} options - options for collector. + * @param {Array=} options.labels - + * Labels for this metric, defaults to no labels. + * @param {Array=} options._reservedLabels - + * Labels which may not be used. Is passed to Metric constructor. + * @param {boolean=} options.register - + * If true registers this collector with the default registry. + * Defaults to true. + * @param {function()=} options.callaback - + * Possibly async function that is called before the metric is + * collected. + * @param { + * function(module:lib/prometheus.LabeledCollector,string) + * } childClass - + * Constructor taking instance of collector and label key as + * arguments and returns a child instance. + */ constructor(type, name, help, options, childClass) { let labels = []; + let reservedLabels = []; let register = true; let callback = null; for (let [key, value] of Object.entries(options)) { if (key === "labels") { labels = value; + } else if (key === "_reservedLabels") { + reservedLabels = value; } else if (key === "register") { register = value; } else if (key === "callback") { @@ -138,78 +366,201 @@ class ValueCollector extends Collector { } // Make sure we don't register to the default registry if metric throws. - let metric = new Metric(type, name, help, labels); + let metric = new Metric(type, name, help, labels, reservedLabels); super(register); this.callback = callback; this.metric = metric; - this._values = new Map(); this._children = new Map(); this._childClass = childClass; - this._defaultChild = labels.length ? null : this.labels({}); - } - - async* collect() { - if (this.callback) { await this.callback(this); } - yield { metric: this.metric, samples: this._values }; - } - - get() { - return this._defaultChild.get(); } + /** + * Access child collector with the given labels. + * + * Creates a labeled child collector for this collector. The child + * supports the methods of Collector for modifying and querying the + * value of the metric. + * + * Once created, a value is initialized for the label values used and + * exported from the collector until it's removed explicitly with + * the .remove() or .removeAll() methods. + * + * @param {...string|Object} labels - + * A string value passed for each label defined on the metric in the + * same order as the labels option given to the collector, or an + * object mapping label name to label value. + * @returns {*} Child collector for the given labels. + */ labels(...labels) { let key = labelsToKey(labels, this.metric.labels); let child = this._children.get(key); if (child === undefined) { - child = new this._childClass(this._values, key); + child = new this._childClass(this, key); this._children.set(key, child); } return child; } + /** + * Remove child collector and data for a given label set + * + * Remove the child collector and the value it stores from the collector + * itself. This will remove the entry exported for the given labels. + * + * @param {...string|Object} labels - + * A string value passed for each label defined on the metric in the + * same order as the labels option given to the collector, or an + * object mapping label name to label value. + * @returns {string} key for labels to remove (for use in subclasses). + */ remove(...labels) { let key = labelsToKey(labels, this.metric.labels); if (key === "") { throw new Error("labels cannot be empty"); } this._children.delete(key); - this._values.delete(key); + return key; } + /** + * Remove child collectors matching a partial label set + * + * Removes all child collecters which has labels matching the ones given + * in labels parameter, Unline .remove this can be a partial set of + * labels and all child collectors with their stored values that shares + * this set of labels will be removed. + * + * @param {Object} labels - + * Object mapping with label name to label values that should be + * matches. + */ removeAll(labels) { if (!Object.keys(labels).length) { throw new Error("labels cannot be empty"); } - removeMatchingLabels(this._values, labels); removeMatchingLabels(this._children, labels); } + /** + * Clear all labeles on this collector + * + * Clears all child collectors from this collector along with their + * stored values. This effectively removes all the exported values. + */ clear() { if (!this.metric.labels.length) { throw new Error("Cannot clear unlabeled metric"); } this._children = new Map(); + } +} + +/** + * Base class for implementing single value per label collectors + * + * @extends module:lib/prometheus.LabeledCollector + * @static + */ +class ValueCollector extends LabeledCollector { + /** + * Create optionally labeled value collector + * + * @param {string} type - Type for metric. + * @param {string} name - Name of the metric. + * @param {string} help - Help text for metric. + * @param {Object} options - options for collector. + * @param {Array=} options.labels - + * Labels for this metric, defaults to no labels. + * @param {Array=} options._reservedLabels - + * Labels which may not be used. Is passed to Metric constructor. + * @param {boolean=} options.register - + * If true registers this collector with the default registry. + * Defaults to true. + * @param {function()=} options.callaback - + * Possibly async function that is called before the collector value + * is collected. + * @param { + * function( + * new:module:lib/prometheus.ValueCollectorChild, + * module:lib/prometheus.LabeledCollector, + * string + * ) + * } childClass - + * Constructor taking instance of collector and label key as + * arguments and returns a child instance. + */ + constructor(type, name, help, options, childClass) { + super(type, name, help, options, childClass); + + this._values = new Map(); + this._defaultChild = this.metric.labels.length ? null : this.labels({}); + } + + async* collect() { + if (this.callback) { await this.callback(this); } + yield { metric: this.metric, samples: new Map([["", this._values]]) }; + } + + /** + * Get the current value for this collector. + * + * Note: Only works if this is an unlabeled collector. + * @returns {number} value stored. + */ + get() { + return this._defaultChild.get(); + } + + remove(...labels) { + let key = super.remove(...labels); + this._values.delete(key); + } + + removeAll(labels) { + super.removeAll(labels); + removeMatchingLabels(this._values, labels); + } + + clear() { + super.clear(); this._values = new Map(); } } +/** + * Child collector representing the value of a single label set + */ class ValueCollectorChild { - constructor(values, key) { - this._values = values; + constructor(collector, key) { + this._values = collector._values; this._key = key; this._values.set(key, 0); } + /** + * Returns the current value of label set + * + * Note: Only works if this is an unlabeled collector. + * @returns {number} value stored. + */ get() { return this._values.get(this._key); } } +/** + * Child counter holding the value for a single label set. + * @extends module:lib/prometheus~ValueCollectorChild + */ class CounterChild extends ValueCollectorChild { + /** + * Increment counter for label set + * @param {number} value - Positive number to increment by. + */ inc(value = 1) { // Note: Inverted to also catch NaN if (!(value >= 0)) { @@ -220,61 +571,714 @@ class CounterChild extends ValueCollectorChild { } } +/** + * Basic increasing counter + * + * Stores and exports an optionally labeled counter metric. This is one of + * the two basic building blocks for instrumenting code and represents a + * metric that can only increase in value, such as the total number of + * requests processed or total time spent processing requests. + * + * Counters should be created at module load time and referenced in the + * functions that increment them, for example: + * + * ```js + * const totalRequests = new Counter( + * "app_request_count_total", + * "Total requests handled.", + * ); + * + * function handleRequest(...) { + * // Do stuff + * totalRequests.inc(); + * } + * ``` + * + * The `totalRequests` counter will register with the default registry and + * provided exposition is set up (see {@link + * module:lib/prometheus.exposition}) the counter will be exported to + * Prometheus starting out with a value of 0. And that is all there is to + * it. + * + * It is sometimes useful however to divide a metric up into diffrent + * sections, for example to have a different count for each endpoint handled + * or one count for successfull requests and one for requests resulting in + * an error. For this Prometheus provides labels, and to use them pass the + * labels option as the third argument to Counter: + * + * ```js + * const totalRequests = new Counter( + * "app_request_count_total", + * "Total requests handled.", + * { labels: ["endpoint", "status"] }, + * ); + * + * function handleRequest(endpoint, ...) { + * let status = "ok"; + * try { + * // Do stuff + * } catch (err) { + * status = "err"; + * } finally { + * totalRequests.labels(endpoint, status).inc(); + * } + * } + * ``` + * + * When using labels the counter no longer gets a default value and it's no + * longer possible to use the `.inc()` method on the counter itself, the + * `.labels()` method has to be invoked with the values for the labels + * defined. `.labels()` returns a child counter for the label values given + * to it and this child can be operated on and cached for performance if + * that is critical. + * + * The lack of a default value may cause issues with aggregating and + * querying the counter values in Prometheus. It is therefore recommended + * where possible to initialize all possible combinations of label values + * that will be used. This is done by calling `.labels()` for each + * combination, for example: + * + * ```js + * for (let endpoint of allEndpoints) { + * for (let status of ["ok", "err"]) { + * totalRequests.labels(endpoint, status); + * } + * } + * ``` + * + * Note that every combination of label values used creates a new time + * series to be stored and processed. You should carefully evaluate which + * labels you actually need as resource usage for a metric increases + * exponentially with the number of labels used. + * + * @extends module:lib/prometheus.ValueCollector + * @static + */ class Counter extends ValueCollector { + /** + * Create optionally labeled counter + * + * @param {string} name - Name of the metric. + * @param {string} help - Help text for metric. + * @param {Object=} options - options for collector. + * @param {Array=} options.labels - + * Labels for this metric, defaults to no labels. + * @param {boolean=} options.register - + * If true registers this collector with the default registry. + * Defaults to true. + * @param {function()=} options.callaback - + * Possibly async function that is called when the metric is + * collected. + */ constructor(name, help, options = {}) { super("counter", name, help, options, CounterChild); } + /** + * Increment counter value + * + * Note: Only works if this is an unlabeled collector. + * @param {number} value - Positive number to increment by. + */ inc(value = 1) { this._defaultChild.inc(value); } } +/** + * Child gauge holding the value for a single label set. + * @extends module:lib/prometheus~ValueCollectorChild + */ class GaugeChild extends ValueCollectorChild { + /** + * Increment gague for label set + * + * @param {number} value - number to increment by. + */ inc(value = 1) { this._values.set(this._key, this._values.get(this._key) + value); } + /** + * Decrement gague for label set + * + * @param {number} value - number to decrease by. + */ dec(value = 1) { this._values.set(this._key, this._values.get(this._key) - value); } + /** + * Set gague for label set + * + * @param {number} value - number to set gauge to. + */ set(value) { this._values.set(this._key, value); } + /** + * Set to current Unix epoch time in seconds for label set + */ setToCurrentTime() { this._values.set(this._key, Date.now() / 1000); } } +/** + * Basic value metric + * + * Stores and exports an optionally labeled value metric. This is one of + * the two basic building blocks for instrumenting code and represents a + * value that can both increase and decrease over time, such as the number + * of requests in-flight or the number of users in a database. + * + * Gauges should be created at module load time and referenced in the + * functions that modify them, for example: + * + * ```js + * const activeRequests = new Gauge( + * "app_active_request_count", + * "Number of requests in-flight.", + * ); + * + * async function handleRequest(...) { + * activeRequests.inc(); + * try { + * // Do async stuff + * } finally { + * activeRequests.dec(); + * } + * } + * ``` + * + * The `activeRequests` gauge will register with the default registry and + * provided exposition is set up (see {@link + * module:lib/prometheus.exposition}) the gauge will be exported to + * Prometheus starting out with a value of 0. + * + * Sometimes keeping track of the value measured is impractical or + * prohibitly expensive. In those cases you can update the value + * of the collector as it's being collected for export with a callback + * function passed as one of the options. + * + * ```js + * const userCount = new Gauge( + * "app_user_count", + * "Number of users in the app.", + * { + * callback: async function() { + * // Make sure this request can not take a long time to + * // complete as that will cause the metrics gathering + * // to time out. + * userCount.set(await someApi.getUserCount()); + * }, + * }, + * ); + * ``` + * + * Keep in mind the callbacks are executed one by one at the time the + * collectors are collected for the exposition given the Prometheus. The + * default timeout in Prometheus for a collection job is 10 seconds. This + * means the callback completion time should be in the order of milliseconds + * to avoid having the time of many collectors add up to too much. + * + * Like with the Counter the Gauge also supports labels. When using labels + * the methods for changing the value of the counter is not longer usable + * directly on the counter itself, instead a child counter with label values + * set has to be retrieved with the `.labels()` method. For example: + * + * ```js + * const userCount = new Gauge( + * "app_user_count", + * "Number of users in the app", + * { + * labels: ["role"], + * callback: async function() { + * userCount.labels("system").set(await someApi.getSystemUserCount()); + * userCount.labels("admin").set(await someApi.getAdminUserCount()); + * userCount.labels("normal").set(await someApi.getNormalUserCount()); + * } + * }, + * ); + * ``` + * + * When using labels the gauge no longer gets a default value, and this may + * cause issues with aggregating and querying the gauge values in + * Prometheus. This is not an issue with the example shown above as all + * label combinations used are given a value when the gauge is collected. + * But if these labeled values were calculated through some other means + * dynamically there may be cases where values for timeseries that are + * occasionally used are missing. When dynamically setting labels it is + * recommended where possible to initialize all possible combinations of + * label values that will be used. This is done by calling `.labels()` for + * each combination, for example: + * + * ```js + * for (let role of ["system", "admin", "normal") { + * userCount.labels(role); + * } + * ``` + * + * Note that every combination of label values used creates a new time + * series to be stored and processed. You should carefully evaluate which + * labels you actually need as resource usage for a metric increases + * exponentially with the number of labels used. + * + * @extends module:lib/prometheus.ValueCollector + * @static + */ class Gauge extends ValueCollector { + /** + * Create optionally labeled gauge + * + * @param {string} name - Name of the metric. + * @param {string} help - Help text for metric. + * @param {Object=} options - options for collector. + * @param {Array=} options.labels - + * Labels for this metric, defaults to no labels. + * @param {boolean=} options.register - + * If true registers this collector with the default registry. + * Defaults to true. + * @param {function()=} options.callaback - + * Possibly async function that is called when the metric is + * collected. + */ constructor(name, help, options = {}) { super("gauge", name, help, options, GaugeChild); } + /** + * Increment gague value + * + * Note: Only works if this is an unlabeled collector. + * @param {number} value - number to increment by. + */ inc(value = 1) { this._defaultChild.inc(value); } + /** + * Decrement gague value + * + * Note: Only works if this is an unlabeled collector. + * @param {number} value - number to decrease by. + */ dec(value = 1) { this._defaultChild.dec(value); } + /** + * Set gague value + * + * Note: Only works if this is an unlabeled collector. + * @param {number} value - number to set gauge to. + */ set(value) { this._defaultChild.set(value); } + /** + * Set to current Unix epoch time in seconds + * + * Note: Only works if this is an unlabeled collector. + */ setToCurrentTime() { this._defaultChild.setToCurrentTime(); } } +function formatValue(value) { + if (value === Infinity) { + return "+Inf"; + } + if (value === -Infinity) { + return "-Inf"; + } + return value.toString(); +} + +function formatBucketKey(bucket, key) { + return `${key === "" ? "" : `${key},`}le="${formatValue(bucket)}"`; +} + +/** + * Child histogram holding the buckets for a single label set. + */ +class HistogramChild { + constructor(collector, key) { + this._bucketValues = collector._bucketValues; + this._sumValues = collector._sumValues; + this._countValues = collector._countValues; + this._key = key; + + this._bucketKeys = new Map(); + for (let bucket of collector._buckets) { + let bucketKey = formatBucketKey(bucket, this._key); + + this._bucketKeys.set(bucket, bucketKey); + this._bucketValues.set(bucketKey, 0); + } + + this._sumValues.set(key, 0); + this._countValues.set(key, 0); + } + + /** + * Mapping of bucket upper bounds to count of observations for label set + * @type {Map} + */ + get buckets() { + return new Map( + [...this._bucketKeys].map( + ([bucket, key]) => [bucket, this._bucketValues.get(key)] + ) + ); + } + + /** + * Sum of all observations for label set + * @type {number} + */ + get sum() { + return this._sumValues.get(this._key); + } + + /** + * Count of observations for label set + * @type {number} + */ + get count() { + return this._countValues.get(this._key); + } + + /** + * Observe a given value and increment matching buckets for label set + * + * @param {number} value - number to count into histogram buckets. + */ + observe(value) { + for (let [bound, key] of this._bucketKeys) { + if (value <= bound) { + this._bucketValues.set(key, this._bucketValues.get(key) + 1); + } + } + + this._sumValues.set(this._key, this._sumValues.get(this._key) + value); + this._countValues.set(this._key, this._countValues.get(this._key) + 1); + } + + /** + * Start a timer for observing a duration for label set + * + * @returns {function()} + * function that when called will store the duration in seconds from + * when the timer was started into the metric. + */ + startTimer() { + let start = process.hrtime.bigint(); + return () => { + let end = process.hrtime.bigint(); + this.observe(Number(end - start) / 1e9); + }; + } +} + +/** + * Histogram metric + * + * Keeps track of observed values in a set of buckets in a similar vein to a + * histogram. This is useful when you have a frequent operation that + * reports a metric that you want insight into the distribution of values + * for. A common case for this is request duration, for example: + * + * ```js + * const requestDuration = new Histogram( + * "app_request_duration_seconds", + * "Time to process app requests", + * ); + * + * async function handleRequest(...) { + * const observeDuration = requestDuration.startTimer(); + * try { + * // Do async stuff. + * } finally { + * observeDuration(); + * } + * } + * ``` + * + * The default buckets used for the histogram is suitable for observing + * HTTP requests durations in seconds, and the `.startTimer()` method + * provides a convenient interface for adding observed durations to the + * histogram. Other types values can be added to a histogram with the + * `.observe()` method. + * + * Like with the Counter and Gauge the Histogram collector also supports + * labels. When using labels the methods for observing values into the + * histogram is not longer usable directly on the counter itself, instead a + * child histogram with label values set has to be retrieved with the + * `.labels()` method. For example: + * + * ```js + * const requestDuration = new Histogram( + * "app_request_duration_seconds", + * "Time to process app requests", + * { labels: ["endpoint"] } + * ); + * + * async function handleRequest(endpoint, ...) { + * const observeDuration = requestDuration.labels(endpoint).startTimer(); + * try { + * // Do async stuff. + * } finally { + * observeDuration(); + * } + * } + * ``` + * + * When using labels the histogram is no longer initalized with a default value, and this may + * cause issues with aggregating and querying the histogram in + * Prometheus. When dynamically setting labels it is recommended where + * possible to initialize all possible combinations of label values that + * will be used. This is done by calling `.labels()` for each combination, + * for example: + * + * ```js + * for (let endpoint of ["/status", "/api", ...) { + * requestDuration.labels(endpoint); + * } + * ``` + * + * Note that every combination of label values used creates a new time + * series for each bucket in the histogram to be stored and processed. + * (e.g., 10 buckets with 10 possible label combinations will result in 100 + * time series being made.) You should carefully evaluate which labels you + * actually need as resource usage for a metric increases exponentially with + * the number of labels used. + * + * @static + */ +class Histogram extends LabeledCollector { + /** + * Create optionally labeled histogram + * + * @param {string} name - Name of the metric. + * @param {string} help - Help text for metric. + * @param {Object=} options - options for collector. + * @param {Array=} options.buckets - + * Buckets to use for the histogram. This is an array of inclusive + * upper bounds. Values observed will increment a bucket if it is + * less than or equal to the upper bound. + * @param {Array=} options.labels - + * Labels for this metric, defaults to no labels. + * @param {boolean=} options.register - + * If true registers this collector with the default registry. + * Defaults to true. + * @param {function()=} options.callaback - + * Possibly async function that is called when the metric is + * collected. + */ + constructor(name, help, options = {}) { + // These defaults are taken from the Python Prometheus client + let buckets = [0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1, 2.5, 5, 7.5, 10, Infinity]; + + let parentOptions = { _reservedLabels: ["le"] }; + for (let [key, value] of Object.entries(options)) { + if (key === "buckets") { + buckets = value; + } else { + parentOptions[key] = value; + } + } + + super("histogram", name, help, parentOptions, HistogramChild); + + if (buckets.slice(-1)[0] !== Infinity) { + buckets = [...buckets, Infinity]; + } + + this._buckets = buckets; + this._bucketValues = new Map(); + this._sumValues = new Map(); + this._countValues = new Map(); + this._defaultChild = this.metric.labels.length ? null : this.labels({}); + } + + async* collect() { + if (this.callback) { await this.callback(this); } + yield { + metric: this.metric, + samples: new Map([ + ["_bucket", this._bucketValues], + ["_sum", this._sumValues], + ["_count", this._countValues], + ]), + }; + } + + /** + * Mapping of bucket upper bounds to count of observations for label set + * + * Note: Only available if this is an unlabeled collector. + * @type {Map} + */ + get buckets() { + return this._defaultChild.buckets; + } + + /** + * Sum of all observations for label set + * + * Note: Only available if this is an unlabeled collector. + * @type {number} + */ + get sum() { + return this._defaultChild.sum; + } + + /** + * Count of observations for label set + * + * Note: Only available if this is an unlabeled collector. + * @type {number} + */ + get count() { + return this._defaultChild.count; + } + + /** + * Observe a given value and increment matching buckets + * + * Note: Only works if this is an unlabeled collector. + * @param {number} value - number to count into histogram buckets. + */ + observe(value) { + this._defaultChild.observe(value); + } + + /** + * Start a timer for observing a duration + * + * Note: Only works if this is an unlabeled collector. + * @returns {function()} + * function that when called will store the duration in seconds from + * when the timer was started into the metric. + */ + startTimer() { + return this._defaultChild.startTimer(); + } + + remove(...labels) { + let key = super.remove(...labels); + for (let bucket of this._buckets) { + let bucketKey = formatBucketKey(bucket, key); + this._bucketValues.delete(bucketKey); + } + this._sumValues.delete(key); + this._countValues.delete(key); + } + + removeAll(labels) { + super.removeAll(labels); + for (let bucket of this._buckets) { + removeMatchingLabels( + this._bucketValues, { ...labels, le: formatValue(bucket) } + ); + } + removeMatchingLabels(this._sumValues, labels); + removeMatchingLabels(this._countValues, labels); + } + + clear() { + super.clear(); + this._bucketValues = new Map(); + this._sumValues = new Map(); + this._countValues = new Map(); + } + + /** + * Helper function for generating linear bucket series + * + * Creates an array of numbers suitable for using as buckets to a + * histogram. The array returned contains `count` numbers starting at + * `start` and with each subsequent number being equal to the previous + * one plus `width`. + * + * @param {number} start - Number to start buckets at. + * @param {number} width - Distance between ecah bucket. + * @param {number} count - Number of buckets. + * @returns {Array} array of buckets. + */ + static linear(start, width, count) { + let buckets = []; + for (let i = 0; i < count; i++) { + buckets.push(start); + start += width; + } + return buckets; + } + + /** + * Helper function for generating exponential bucket series + * + * Creates an array of numbers suitable for using as buckets to a + * histogram. The array returned contains `count` numbers starting at + * `start` and with each subsequent number being equal to the previous + * multiplied with `factor`. + * + * @param {number} start - Number to start buckets at. + * @param {number} factor - Ratio between each bucket. + * @param {number} count - Number of buckets. + * @returns {Array} array of buckets. + */ + static exponential(start, factor, count) { + let buckets = []; + for (let i = 0; i < count; i++) { + buckets.push(start); + start *= factor; + } + return buckets; + } +} + +/** + * Collection of collectors + * + * Provides convienece methods for grouping collectors together and + * gathering the metrics from all of the contained collectors. By default + * collectors are registered and collected from the default registry so + * using this is usually not necessary. + * + * The basic way to use a custom registry is to create Collectors without + * registring them to the default registry and then adding them to your own + * registry, for example: + * + * ```js + * const myRegistry = new CollectorRegistry(); + * const myCounter = new Counter( "a_counter", "A counter.", { register: false }); + * myRegistry.register(myCounter); + * + * // In the /metrics HTTP request handler + * let text = await exposition(myRegistry.collect()); + * ``` + * + * The same collector can be registered to multiple registries. This may be + * used to implement responding with different sets of metrics depending on + * what is requested. + * + * @static + */ class CollectorRegistry { constructor() { this.collectors = []; } + /** + * Collect metrics from all registered collectors. + * + * @returns {*} Async iterator of {@link CollectorResult}. + */ async* collect() { for (let collector of this.collectors) { for await (let result of collector.collect()) { @@ -283,19 +1287,29 @@ class CollectorRegistry { } } - register(metric) { - let index = this.collectors.lastIndexOf(metric); + /** + * Add collector to the registry. + * + * @param {lib/prometheus.Collector} collector - Collector to add. + */ + register(collector) { + let index = this.collectors.lastIndexOf(collector); if (index !== -1) { throw new Error( "Collector is already registered in this registry." ); } - this.collectors.push(metric); + this.collectors.push(collector); } - unregister(metric) { - let index = this.collectors.lastIndexOf(metric); + /** + * Remove collector from the registry. + * + * @param {lib/prometheus.Collector} collector - Collector to remove. + */ + unregister(collector) { + let index = this.collectors.lastIndexOf(collector); if (index === -1) { throw new Error( "Collector is not registered in this registry." @@ -315,16 +1329,6 @@ function escapeHelp(help) { ; } -function formatValue(value) { - if (value === Infinity) { - return "+Inf"; - } - if (value === -Infinity) { - return "-Inf"; - } - return value.toString(); -} - async function* expositionLines(resultsIterator) { let first = true; for await (let result of resultsIterator) { @@ -337,16 +1341,53 @@ async function* expositionLines(resultsIterator) { yield `# HELP ${result.metric.name} ${escapeHelp(result.metric.help)}\n`; yield `# TYPE ${result.metric.name} ${result.metric.type}\n`; - for (let [key, value] of result.samples) { - if (key === "") { - yield `${result.metric.name} ${formatValue(value)}\n`; - } else { - yield `${result.metric.name}{${key}} ${formatValue(value)}\n`; + for (let [suffix, samples] of result.samples) { + for (let [key, value] of samples) { + if (key === "") { + yield `${result.metric.name}${suffix} ${formatValue(value)}\n`; + } else { + yield `${result.metric.name}${suffix}{${key}} ${formatValue(value)}\n`; + } } } } } +/** + * Serialize into Prometheus text exposition format + * + * Asynchronously collects metrics and converts them into the text based + * exposition format for Prometheus. The collectors collected by default is + * taken from the default registry, but this can be overidden by passing a + * resultsIterator either from another registry or custom created. + * + * The resulting text from calling this method should be served to + * prometheus, typically by hosting an HTTP server in the app and responding + * to the /metrics endpoint. See example below for how this is done with + * express.js. + * + * @example + * const app = require("express")(); + * async function getMetrics(req, res) { + * // By default exposition uses the default collector registry + * let text = await exposition(); + * res.set("Content-Type", exposition.contentType); + * res.send(text); + * } + * app.get("/metrics", (req, res, next) => getMetrics(req, res).catch(next)); + * app.listen(9100); + * + * @param {*} resultsIterator - + * Asynchronously itreable of {@link CollectorResult} results to create + * exposition for. Defaults to collecting results from {@link + * module:lib/prometheus.defaultRegistry}. + * @returns {string} Prometheus exposition. + * + * @property {string} exposition.contentType + * HTTP Content-Type for the exposition format that's implemented + * + * @static + */ async function exposition(resultsIterator = defaultRegistry.collect()) { let lines = ""; for await (let line of expositionLines(resultsIterator)) { @@ -355,9 +1396,26 @@ async function exposition(resultsIterator = defaultRegistry.collect()) { return lines; } -// HTTP Content-Type for the exposition format that's implemented exposition.contentType = "text/plain; version=0.0.4"; +/** + * Serialize CollectorResult into a plain object + * + * Converts a {@link module:lib/prometheus~CollectorResult} into a plain + * object form that can be stringified to JSON. + * + * @param {module:lib/prometheus~CollectorResult} result - + * Result to serialize into plain object form. + * @param {Object} options - Options for controlling the serialization. + * @param {string} options.addLabels - + * Additional labels to append to each value. This may be used if + * multiple sources are combined have the same metric and a qualifier is + * needed to make sure the label sets are unique. + * @param {string} options.metricName - Override metric name of the result. + * @param {string} options.metricHelp - Override metric help of the result. + * @returns {Object} plain object form of the result. + * @static + */ function serializeResult(result, options = {}) { let addLabels = null; let metricName = result.metric.name; @@ -382,12 +1440,16 @@ function serializeResult(result, options = {}) { } else { samples = new Map(); let key = labelsToKey([addLabels], [...Object.keys(addLabels)]); - for (let [labels, value] of result.samples.entries()) { - if (labels === "") { - samples.set(key, value); - } else { - samples.set(`${labels},${key}`, value); + for (let [suffix, suffixSamples] of result.samples) { + let labeledSamples = new Map(); + for (let [labels, value] of suffixSamples) { + if (labels === "") { + labeledSamples.set(key, value); + } else { + labeledSamples.set(`${labels},${key}`, value); + } } + samples.set(suffix, labeledSamples); } } @@ -398,10 +1460,19 @@ function serializeResult(result, options = {}) { help: metricHelp, labels: [...result.metric.labels, ...Object.keys(addLabels)], }, - samples: [...samples.entries()], + samples: [...samples].map(([name, metricSamples]) => [name, [...metricSamples]]), }; } +/** + * Deserialize CollectorResult from plain object + * + * Reverse counterpart to {@link module:lib/prometheus.serializeResult}. + * + * @param {Object} serializedResult - Previously serialized result object. + * @returns {module:lib/prometheus~CollectorResult} deserialized result. + * @static + */ function deserializeResult(serializedResult) { return { metric: new Metric( @@ -410,10 +1481,18 @@ function deserializeResult(serializedResult) { serializedResult.metric.help, serializedResult.metric.labels, ), - samples: new Map(serializedResult.samples), + samples: new Map( + serializedResult.samples.map( + ([suffix, suffixSamples]) => [suffix, new Map(suffixSamples)] + ) + ), }; } +/** + * Default collectors provided by this library + * @type {Object} + */ let defaultCollectors = {}; defaultCollectors.processCpuSecondsTotal = new Gauge( "process_cpu_seconds_total", @@ -456,8 +1535,10 @@ defaultCollectors.processStartTimeSeconds.setToCurrentTime(); module.exports = { Counter, Gauge, + Histogram, CollectorRegistry, Collector, + LabeledCollector, ValueCollector, Metric, diff --git a/packages/master/master.js b/packages/master/master.js index e3b7f6af..f469272c 100755 --- a/packages/master/master.js +++ b/packages/master/master.js @@ -54,6 +54,7 @@ const libPrometheus = require("@clusterio/lib/prometheus"); const libConfig = require("@clusterio/lib/config"); const libUsers = require("@clusterio/lib/users"); const libSharedCommands = require("@clusterio/lib/shared_commands"); +const libHelpers = require("@clusterio/lib/helpers"); const express = require("express"); const compression = require("compression"); @@ -147,8 +148,7 @@ async function getMetrics(req, res, next) { let requests = []; for (let slaveConnection of slaveConnections.values()) { - // XXX this needs a timeout - requests.push(libLink.messages.getMetrics.send(slaveConnection)); + requests.push(libHelpers.timeout(libLink.messages.getMetrics.send(slaveConnection), 8e3, null)); } for await (let result of await libPrometheus.defaultRegistry.collect()) { @@ -157,14 +157,28 @@ async function getMetrics(req, res, next) { let resultMap = new Map(); for (let response of await Promise.all(requests)) { + if (!response) { + // TODO: Log timeout occured? + continue; + } + for (let result of response.results) { if (!resultMap.has(result.metric.name)) { resultMap.set(result.metric.name, result); } else { // Merge metrics received by multiple slaves - let stored = resultMap.get(result.metric.name); - stored.samples.push(...result.samples); + let receivedSamples = new Map(result.samples); + for (let [suffix, suffixSamples] of resultMap.get(result.metric.name).samples) { + if (receivedSamples.has(suffix)) { + suffixSamples.push(...receivedSamples.get(suffix)); + receivedSamples.delete(suffix); + } + } + + for (let entry of receivedSamples) { + result.samples.push(entry); + } } } } @@ -712,29 +726,43 @@ class ControlConnection extends BaseConnection { } async assignInstanceCommandRequestHandler(message, request) { - let instance = db.instances.get(message.data.instance_id); + let { slave_id, instance_id } = message.data; + let instance = db.instances.get(instance_id); if (!instance) { - throw new libErrors.RequestError(`Instance with ID ${message.data.instance_id} does not exist`); + throw new libErrors.RequestError(`Instance with ID ${instance_id} does not exist`); } - // XXX: Should the instance be stopped if it's running on another slave? - - let connection = slaveConnections.get(message.data.slave_id); - if (!connection) { - // The case of the slave not getting the assign instance message - // still have to be handled, so it's not a requirement that the - // target slave be connected to the master while doing the - // assignment, but it is IMHO a better user experience if this - // is the case. - throw new libErrors.RequestError("Target slave is not connected to the master server"); + // Check if target slave is connected + let newSlaveConnection; + if (slave_id !== null) { + newSlaveConnection = slaveConnections.get(slave_id); + if (!newSlaveConnection) { + // The case of the slave not getting the assign instance message + // still have to be handled, so it's not a requirement that the + // target slave be connected to the master while doing the + // assignment, but it is IMHO a better user experience if this + // is the case. + throw new libErrors.RequestError("Target slave is not connected to the master server"); + } } - instance.config.set("instance.assigned_slave", message.data.slave_id); + // Unassign from currently assigned slave if it is connected. + let currentAssignedSlave = instance.config.get("instance.assigned_slave"); + if (currentAssignedSlave !== null && slave_id !== currentAssignedSlave) { + let oldSlaveConnection = slaveConnections.get(currentAssignedSlave); + if (oldSlaveConnection && !oldSlaveConnection.connector.closing) { + await libLink.messages.unassignInstance.send(oldSlaveConnection, { instance_id }); + } + } - return await libLink.messages.assignInstance.send(connection, { - instance_id: instance.config.get("instance.id"), - serialized_config: instance.config.serialize(), - }); + // Assign to target + instance.config.set("instance.assigned_slave", slave_id); + if (slave_id !== null) { + await libLink.messages.assignInstance.send(newSlaveConnection, { + instance_id, + serialized_config: instance.config.serialize(), + }); + } } async setInstanceOutputSubscriptionsRequestHandler(message) { @@ -967,13 +995,23 @@ class SlaveConnection extends BaseConnection { async instanceStatusChangedEventHandler(message, event) { let instance = db.instances.get(message.data.instance_id); + + // It's possible to get updates from an instance that does not exist + // or is not assigned to the slave it originated from if it was + // reassigned or deleted while the connection to the slave it was + // originally on was down at the time. + if (!instance || instance.config.get("instance.assigned_slave") !== this._id) { + console.log(`Clusterio | Got bogus update for instance id ${message.data.instance_id}`); + return; + } + let prev = instance.status; instance.status = message.data.status; console.log(`Clusterio | Instance ${instance.config.get("instance.name")} State: ${instance.status}`); await libPlugin.invokeHook(masterPlugins, "onInstanceStatusChanged", instance, prev); } - async updateInstancesEventHandler(message) { + async updateInstancesRequestHandler(message) { // Push updated instance configs for (let instance of db.instances.values()) { if (instance.config.get("instance.assigned_slave") === this._id) { @@ -991,10 +1029,21 @@ class SlaveConnection extends BaseConnection { let masterInstance = db.instances.get(instanceConfig.get("instance.id")); if (masterInstance) { + // Check if this instance is assigned somewhere else. + if (masterInstance.config.get("instance.assigned_slave") !== this._id) { + await libLink.messages.unassignInstance.send(this, { + instance_id: masterInstance.config.get("instance.id"), + }); + continue; + } + // Already have this instance, update state instead if (masterInstance.status !== instance.status) { let prev = masterInstance.status; masterInstance.status = instance.status; + console.log( + `Clusterio | Instance ${instanceConfig.get("instance.name")} State: ${instance.status}` + ); await libPlugin.invokeHook(masterPlugins, "onInstanceStatusChanged", instance, prev); } continue; diff --git a/packages/slave/slave.js b/packages/slave/slave.js index 7644541c..3345b68f 100755 --- a/packages/slave/slave.js +++ b/packages/slave/slave.js @@ -36,12 +36,21 @@ const libConfig = require("@clusterio/lib/config"); const libSharedCommands = require("@clusterio/lib/shared_commands"); -const instanceRconCommandsCounter = new libPrometheus.Counter( - "clusterio_instance_rcon_commands_total", - "How many commands have been sent to the instance", +const instanceRconCommandDuration = new libPrometheus.Histogram( + "clusterio_instance_rcon_command_duration_seconds", + "Histogram of the RCON command duration from request to response.", { labels: ["instance_id"] } ); +const instanceRconCommandSize = new libPrometheus.Histogram( + "clusterio_instance_rcon_command_size_bytes", + "Histogram of the RCON command sizes that are sent.", + { + labels: ["instance_id", "plugin"], + buckets: libPrometheus.Histogram.exponential(16, 2, 12), + } +); + const instanceFactorioCpuTime = new libPrometheus.Gauge( "clusterio_instance_factorio_cpu_time_total", "Factorio CPU time spent in seconds.", @@ -61,7 +70,7 @@ const instanceFactorioAutosaveSize = new libPrometheus.Gauge( ); function applyAsConfig(name) { - return async function action(server, value) { + return async function action(instance, value) { if (name === "tags" && value instanceof Array) { // Replace spaces with non-break spaces and delimit by spaces. // This does change the defined tags, but there doesn't seem to @@ -69,7 +78,7 @@ function applyAsConfig(name) { value = value.map(tag => tag.replace(/ /g, "\u00a0")).join(" "); } try { - await server.sendRcon(`/config set ${name} ${value}`); + await instance.sendRcon(`/config set ${name} ${value}`); } catch (err) { console.error(`Error applying server setting ${name}`, err); } @@ -91,11 +100,11 @@ const serverSettingsActions = { "game_password": applyAsConfig("password"), "require_user_verification": applyAsConfig("require-user-verification"), "tags": applyAsConfig("tags"), - "visibility": async (server, value) => { + "visibility": async (instance, value) => { for (let scope of ["lan", "public", "steam"]) { try { let enabled = Boolean(value[scope]); - await server.sendRcon(`/config set visibility-${scope} ${enabled}`); + await instance.sendRcon(`/config set visibility-${scope} ${enabled}`); } catch (err) { console.error(`Error applying visibility ${scope}`, err); } @@ -124,6 +133,9 @@ class Instance extends libLink.Link{ } else if (group.name === "factorio" && field === "enable_whitelist") { this.updateFactorioWhitelist(group.get(field)).finally(hook); } else { + if (group.name === "factorio" && field === "max_concurrent_commands") { + this.server.maxConcurrentCommands = group.get(field); + } hook(); } }; @@ -137,6 +149,7 @@ class Instance extends libLink.Link{ enableWhitelist: this.config.get("factorio.enable_whitelist"), verboseLogging: this.config.get("factorio.verbose_logging"), stripPaths: this.config.get("factorio.strip_paths"), + maxConcurrentCommands: this.config.get("factorio.max_concurrent_commands"), }; this._status = "stopped"; @@ -145,12 +158,6 @@ class Instance extends libLink.Link{ factorioDir, this._dir, serverOptions ); - let originalSendRcon = this.server.sendRcon; - this.server.sendRcon = (...args) => { - instanceRconCommandsCounter.labels(String(this.config.get("instance.id"))).inc(); - return originalSendRcon.call(this.server, ...args); - }; - this.server.on("output", (output) => { libLink.messages.instanceOutput.send(this, { instance_id: this.config.get("instance.id"), output }); @@ -176,6 +183,17 @@ class Instance extends libLink.Link{ }); } + async sendRcon(message, expectEmpty, plugin = "") { + let instanceId = String(this.config.get("instance.id")); + let observeDuration = instanceRconCommandDuration.labels(instanceId).startTimer(); + try { + return await this.server.sendRcon(message, expectEmpty); + } finally { + observeDuration(); + instanceRconCommandSize.labels(instanceId, plugin).observe(Buffer.byteLength(message, "utf8")); + } + } + async _autosave(name) { let stat = await fs.stat(this.path("saves", `${name}.zip`)); instanceFactorioAutosaveSize.labels(String(this.config.get("instance.id"))).set(stat.size); @@ -515,7 +533,7 @@ class Instance extends libLink.Link{ async updateInstanceData() { let name = libLuaTools.escapeString(this.config.get("instance.name")); let id = this.config.get("instance.id"); - await this.server.sendRcon(`/sc clusterio_private.update_instance(${id}, "${name}")`, true); + await this.sendRcon(`/sc clusterio_private.update_instance(${id}, "${name}")`, true); } async updateFactorioSettings(current, previous) { @@ -524,7 +542,7 @@ class Instance extends libLink.Link{ for (let [key, action] of Object.entries(serverSettingsActions)) { if (current[key] !== undefined && !util.isDeepStrictEqual(current[key], previous[key])) { - await action(this.server, current[key]); + await action(this, current[key]); } } } @@ -537,18 +555,18 @@ class Instance extends libLink.Link{ */ async updateFactorioWhitelist(enable) { if (!enable) { - await this.server.sendRcon("/whitelist disable"); + await this.sendRcon("/whitelist disable"); } if (this.config.get("factorio.sync_whitelist")) { - await this.server.sendRcon("/whitelist clear"); + await this.sendRcon("/whitelist clear"); for (let player of this._slave.whitelist) { - await this.server.sendRcon(`/whitelist ${player}`); + await this.sendRcon(`/whitelist ${player}`); } } if (enable) { - await this.server.sendRcon("/whitelist enable"); + await this.sendRcon("/whitelist enable"); } } @@ -559,7 +577,7 @@ class Instance extends libLink.Link{ let { name, admin } = message.data; let command = admin ? `/promote ${name}` : `/demote ${name}`; - await this.server.sendRcon(command); + await this.sendRcon(command); } async banlistUpdateEventHandler(message) { @@ -569,7 +587,7 @@ class Instance extends libLink.Link{ let { name, banned, reason } = message.data; let command = banned ? `/ban ${name} ${reason}` : `/unban ${name}`; - await this.server.sendRcon(command); + await this.sendRcon(command); } async whitelistUpdateEventHandler(message) { @@ -579,7 +597,7 @@ class Instance extends libLink.Link{ let { name, whitelisted } = message.data; let command = whitelisted ? `/whitelist add ${name}` : `/whiteliste remove ${name}`; - await this.server.sendRcon(command); + await this.sendRcon(command); } @@ -719,7 +737,7 @@ class Instance extends libLink.Link{ } async sendRconRequestHandler(message) { - let result = await this.server.sendRcon(message.data.command); + let result = await this.sendRcon(message.data.command); return { result }; } @@ -751,15 +769,16 @@ class Instance extends libLink.Link{ * Searches for instances in the provided directory * * Looks through all sub-dirs of the provided directory for valid - * instance definitions and updates the mapping of instance id to + * instance definitions and return a mapping of instance id to * instanceInfo objects. * - * @param {Map} instanceInfos - - * mapping between instance id and information about this instance. * @param {string} instancesDir - Directory containing instances * @param {Object} logger - console like logging interface. + * @returns {Map} + * mapping between instance id and information about this instance. */ -async function discoverInstances(instanceInfos, instancesDir, logger) { +async function discoverInstances(instancesDir, logger) { + let instanceInfos = new Map(); for (let entry of await fs.readdir(instancesDir, { withFileTypes: true })) { if (entry.isDirectory()) { let instanceConfig = new libConfig.InstanceConfig(); @@ -777,11 +796,6 @@ async function discoverInstances(instanceInfos, instancesDir, logger) { continue; } - // Ignore instances we have already discovered before. - if (instanceInfos.has(instanceConfig.get("instance.id"))) { - continue; - } - let instancePath = path.join(instancesDir, entry.name); logger.log(`found instance ${instanceConfig.get("instance.name")} in ${instancePath}`); instanceInfos.set(instanceConfig.get("instance.id"), { @@ -790,6 +804,7 @@ async function discoverInstances(instanceInfos, instancesDir, logger) { }); } } + return instanceInfos; } class InstanceConnection extends libLink.Link { @@ -815,7 +830,7 @@ class InstanceConnection extends libLink.Link { let instanceConnection = this.slave.instanceConnections.get(instanceId); if (!instanceConnection) { // Instance is probably on another slave - return await request.send(this.slave, message.data); + return await this.forwardRequestToMaster(message, request); } if (request.plugin && !instanceConnection.plugins.has(request.plugin)) { @@ -831,7 +846,7 @@ class InstanceConnection extends libLink.Link { let instanceConnection = this.slave.instanceConnections.get(instanceId); if (!instanceConnection) { // Instance is probably on another slave - await this.slave.forwardEventToMaster(message, event); + await this.forwardEventToMaster(message, event); return; } if (event.plugin && !instanceConnection.plugins.has(event.plugin)) { return; } @@ -911,6 +926,7 @@ class Slave extends libLink.Link { this.config = slaveConfig; this.instanceConnections = new Map(); + this.discoveredInstanceInfos = new Map(); this.instanceInfos = new Map(); this.adminlist = new Set(); @@ -922,6 +938,7 @@ class Slave extends libLink.Link { this.serverPlugins = new Map(Object.entries(data.plugins)); }); + this._startup = true; this._disconnecting = false; this._shuttingDown = false; @@ -1120,20 +1137,28 @@ class Slave extends libLink.Link { console.log(`Clusterio | Updated config for ${instanceInfo.path}`); } else { - let instanceConfig = new libConfig.InstanceConfig(); - await instanceConfig.load(serialized_config); + instanceInfo = this.discoveredInstanceInfos.get(instance_id); + if (instanceInfo) { + instanceInfo.config.update(serialized_config, true); - // XXX: race condition on multiple simultanious calls - let instanceDir = await this._findNewInstanceDir(instanceConfig.get("instance.name")); + } else { + let instanceConfig = new libConfig.InstanceConfig(); + await instanceConfig.load(serialized_config); + + // XXX: race condition on multiple simultanious calls + let instanceDir = await this._findNewInstanceDir(instanceConfig.get("instance.name")); + + await Instance.create(instanceDir, this.config.get("slave.factorio_directory")); + instanceInfo = { + path: instanceDir, + config: instanceConfig, + }; + + this.discoveredInstanceInfos.set(instance_id, instanceInfo); + } - await Instance.create(instanceDir, this.config.get("slave.factorio_directory")); - instanceInfo = { - path: instanceDir, - config: instanceConfig, - }; this.instanceInfos.set(instance_id, instanceInfo); - console.log(`Clusterio | assigned instance ${instanceConfig.get("instance.name")}`); - + console.log(`Clusterio | assigned instance ${instanceInfo.config.get("instance.name")}`); } // Somewhat hacky, but in the event of a lost session the status is @@ -1156,6 +1181,20 @@ class Slave extends libLink.Link { ); } + async unassignInstanceRequestHandler(message) { + let instanceId = message.data.instance_id; + let instanceInfo = this.instanceInfos.get(instanceId); + if (instanceInfo) { + let instanceConnection = this.instanceConnections.get(instanceId); + if (instanceConnection && ["starting", "running"].includes(instanceConnection.status)) { + await libLink.messages.stopInstance.send(instanceConnection, { instance_id: instanceId }); + } + + this.instanceInfos.delete(instanceId); + console.log(`Clusterio | unassigned instance ${instanceInfo.config.get("instance.name")}`); + } + } + /** * Initialize and connect an unloaded instance * @@ -1245,13 +1284,14 @@ class Slave extends libLink.Link { throw new libErrors.RequestError(`Instance with ID ${instanceId} is running`); } - let instanceInfo = this.instanceInfos.get(instanceId); + let instanceInfo = this.discoveredInstanceInfos.get(instanceId); if (!instanceInfo) { throw new libErrors.RequestError(`Instance with ID ${instanceId} does not exist`); } - await fs.remove(instanceInfo.path); + this.discoveredInstanceInfos.delete(instanceId); this.instanceInfos.delete(instanceId); + await fs.remove(instanceInfo.path); } /** @@ -1261,16 +1301,36 @@ class Slave extends libLink.Link { * the slave and master server with the new list of instances. */ async updateInstances() { - await discoverInstances(this.instanceInfos, this.config.get("slave.instances_directory"), console); + this.discoveredInstanceInfos = await discoverInstances(this.config.get("slave.instances_directory"), console); let list = []; - for (let [instanceId, instanceInfo] of this.instanceInfos) { + for (let [instanceId, instanceInfo] of this.discoveredInstanceInfos) { let instanceConnection = this.instanceConnections.get(instanceId); list.push({ serialized_config: instanceInfo.config.serialize(), status: instanceConnection ? instanceConnection.status : "stopped", }); } - libLink.messages.updateInstances.send(this, { instances: list }); + await libLink.messages.updateInstances.send(this, { instances: list }); + + // Handle configured auto startup instances + if (this._startup) { + this._startup = false; + + for (let [instanceId, instanceInfo] of this.instanceInfos) { + if (instanceInfo.config.get("instance.auto_start")) { + try { + let instanceConnection = await this._connectInstance(instanceId); + await libLink.messages.startInstance.send(instanceConnection, { + instance_id: instanceId, + save: null, + }); + } catch (err) { + console.error(`Error during auto startup for ${instanceInfo.config.get("instance.name")}:`); + console.error(err); + } + } + } + } } async prepareDisconnectRequestHandler(message, request) { diff --git a/packages/web_ui/src/util/websocket.jsx b/packages/web_ui/src/util/websocket.jsx index 61e7cdc6..86ea67ba 100644 --- a/packages/web_ui/src/util/websocket.jsx +++ b/packages/web_ui/src/util/websocket.jsx @@ -49,6 +49,10 @@ export class Control extends libLink.Link { } this.instanceOutputHandlers = new Map(); + + this.connector.on("connect", () => { + this.updateInstanceOutputSubscriptions().catch(console.error); + }); } async instanceOutputEventHandler(message) { diff --git a/plugins/global_chat/instance.js b/plugins/global_chat/instance.js index 65e994fc..ff996788 100644 --- a/plugins/global_chat/instance.js +++ b/plugins/global_chat/instance.js @@ -33,7 +33,7 @@ class InstancePlugin extends libPlugin.BaseInstancePlugin { async chatEventHandler(message) { // TODO check if cross server chat is enabled let content = `[${message.data.instance_name}] ${removeTags(message.data.content)}`; - await this.instance.server.sendRcon(`/sc game.print('${libLuaTools.escapeString(content)}')`, true); + await this.sendRcon(`/sc game.print('${libLuaTools.escapeString(content)}')`, true); } sendChat(message) { diff --git a/plugins/research_sync/instance.js b/plugins/research_sync/instance.js index c8bef0fb..e495db3c 100644 --- a/plugins/research_sync/instance.js +++ b/plugins/research_sync/instance.js @@ -29,7 +29,7 @@ class InstancePlugin extends libPlugin.BaseInstancePlugin { async progressEventHandler(message) { let techsJson = libLuaTools.escapeString(JSON.stringify(message.data.technologies)); - await this.instance.server.sendRcon(`/sc research_sync.update_progress("${techsJson}")`, true); + await this.sendRcon(`/sc research_sync.update_progress("${techsJson}")`, true); } async researchFinished(tech) { @@ -38,13 +38,13 @@ class InstancePlugin extends libPlugin.BaseInstancePlugin { async finishedEventHandler(message) { let { name, level, researched } = message.data; - await this.instance.server.sendRcon( + await this.sendRcon( `/sc research_sync.research_technology("${libLuaTools.escapeString(name)}", ${level})`, true ); } async onStart() { - let dumpJson = await this.instance.server.sendRcon("/sc research_sync.dump_technologies()"); + let dumpJson = await this.sendRcon("/sc research_sync.dump_technologies()"); let techsToSend = []; let instanceTechs = new Map(); for (let tech of JSON.parse(dumpJson)) { @@ -74,7 +74,7 @@ class InstancePlugin extends libPlugin.BaseInstancePlugin { if (techsToSync.length) { let syncJson = libLuaTools.escapeString(JSON.stringify(techsToSync)); - await this.instance.server.sendRcon(`/sc research_sync.sync_technologies("${syncJson}")`, true); + await this.sendRcon(`/sc research_sync.sync_technologies("${syncJson}")`, true); } } } diff --git a/plugins/statistics_exporter/instance.js b/plugins/statistics_exporter/instance.js index 8a3d63be..002e51f6 100644 --- a/plugins/statistics_exporter/instance.js +++ b/plugins/statistics_exporter/instance.js @@ -1,6 +1,7 @@ "use strict"; const libPlugin = require("@clusterio/lib/plugin"); const { Gauge } = require("@clusterio/lib/prometheus"); +const libHelpers = require("@clusterio/lib/helpers"); const instancePlayerCount = new Gauge( @@ -28,8 +29,8 @@ class InstancePlugin extends libPlugin.BaseInstancePlugin { } } - async onMetrics() { - let string = await this.instance.server.sendRcon("/sc statistics_exporter.export()"); + async gatherMetrics() { + let string = await this.sendRcon("/sc statistics_exporter.export()"); let stats; try { stats = JSON.parse(string); @@ -61,6 +62,14 @@ class InstancePlugin extends libPlugin.BaseInstancePlugin { } } } + + async onMetrics() { + // Wait up to 5 seconds for the metrics to be collected. It may + // take a long time for the command to go through if the command + // stream is overloaded. Should this take more than 5 seconds the + // previous values for the metrics will end up being sent to master. + await libHelpers.timeout(this.gatherMetrics(), 5e3); + } } module.exports = { diff --git a/plugins/subspace_storage/instance.js b/plugins/subspace_storage/instance.js index 7b5daaba..6250517d 100644 --- a/plugins/subspace_storage/instance.js +++ b/plugins/subspace_storage/instance.js @@ -22,7 +22,7 @@ class InstancePlugin extends libPlugin.BaseInstancePlugin { async onStart() { this.pingId = setInterval(() => { - this.instance.server.sendRcon( + this.sendRcon( "/sc __subspace_storage__ global.ticksSinceMasterPinged = 0", true ).catch(unexpectedError); }, 5000); @@ -30,7 +30,7 @@ class InstancePlugin extends libPlugin.BaseInstancePlugin { let response = await this.info.messages.getStorage.send(this.instance); // TODO Diff with dump of invdata produce minimal command to sync let itemsJson = libLuaTools.escapeString(JSON.stringify(response.items)); - await this.instance.server.sendRcon(`/sc __subspace_storage__ UpdateInvData("${itemsJson}", true)`, true); + await this.sendRcon(`/sc __subspace_storage__ UpdateInvData("${itemsJson}", true)`, true); } onExit() { @@ -68,7 +68,7 @@ class InstancePlugin extends libPlugin.BaseInstancePlugin { } let itemsJson = libLuaTools.escapeString(JSON.stringify(response.items)); - await this.instance.server.sendRcon(`/sc __subspace_storage__ Import("${itemsJson}")`, true); + await this.sendRcon(`/sc __subspace_storage__ Import("${itemsJson}")`, true); } // combinator signals --------------------------------------------------------- @@ -87,7 +87,7 @@ class InstancePlugin extends libPlugin.BaseInstancePlugin { items.push(["signal-unixtime", Math.floor(Date.now()/1000)]); let itemsJson = libLuaTools.escapeString(JSON.stringify(items)); - await this.instance.server.sendRcon(`/sc __subspace_storage__ UpdateInvData("${itemsJson}")`, true); + await this.sendRcon(`/sc __subspace_storage__ UpdateInvData("${itemsJson}")`, true); } } diff --git a/test/lib/helpers.js b/test/lib/helpers.js index e5b83697..b9671a93 100644 --- a/test/lib/helpers.js +++ b/test/lib/helpers.js @@ -1,5 +1,6 @@ "use strict"; const assert = require("assert").strict; +const hrtime = process.hrtime.bigint; const libHelpers = require("@clusterio/lib/helpers"); @@ -15,4 +16,30 @@ describe("lib/helpers", function() { assert.equal(libHelpers.basicType({ a: 1 }), "object"); }); }); + + describe("wait", function() { + it("should wait the approximate amount of time given", async function() { + let startNs = hrtime(); + await libHelpers.wait(100); + let duration = Number(hrtime() - startNs) / 1e6; + + assert(duration > 90 && duration < 110, `duration waited (${duration}ms) 10% of expected (100ms)`); + }); + }); + + describe("timeout", function() { + it("should return the result of an already resolved promise", async function() { + assert.equal(await libHelpers.timeout(Promise.resolve("value"), 10, "timeout"), "value"); + }); + it("should return the result of the promise if received before timeout", async function() { + assert.equal(await libHelpers.timeout( + (async () => { await libHelpers.wait(10); return "value"; })(), 20, "timeout" + ), "value"); + }); + it("should return the timeoutResult if not received before timeout", async function() { + assert.equal(await libHelpers.timeout( + (async () => { await libHelpers.wait(20); return "value"; })(), 10, "timeout" + ), "timeout"); + }); + }); }); diff --git a/test/lib/prometheus.js b/test/lib/prometheus.js index adef36c5..6507c197 100644 --- a/test/lib/prometheus.js +++ b/test/lib/prometheus.js @@ -200,7 +200,194 @@ describe("lib/prometheus", function() { }); }); + describe("class Histogram", function() { + let histogram; + let labeledHistogram; + beforeEach(function() { + histogram = new prometheus.Histogram( + "test", "Help", { register: false, buckets: [1, 2, 3, 4, Infinity] } + ); + labeledHistogram = new prometheus.Histogram( + "test", "Help", { register: false, buckets: [1, Infinity], labels: ["a", "b"] } + ); + }); + + describe("constructor", function() { + it("should have buckets, sum and count initialized to 0", function() { + assert.deepEqual( + histogram.buckets, + new Map([[1, 0], [2, 0], [3, 0], [4, 0], [Infinity, 0]]) + ); + assert.equal(histogram.sum, 0); + assert.equal(histogram.count, 0); + }); + it("should provide a default for buckets", function() { + histogram = new prometheus.Histogram( + "test", "help", { register: false } + ); + assert(histogram._buckets.length, "no default buckets were made"); + }); + it("should add inifinity bucket if not present", function() { + histogram = new prometheus.Histogram( + "test", "help", { buckets: [1], register: false } + ); + assert.deepEqual(histogram._buckets, [1, Infinity]); + }); + it("should not mutate bucket option passed to it", function() { + let buckets = [1]; + histogram = new prometheus.Histogram( + "test", "help", { buckets, register: false } + ); + assert.deepEqual(buckets, [1]); + }); + it("should throw on unrecoginzed option", function() { + assert.throws( + () => new prometheus.Histogram("test", "help", { invalid: 1 }), + new Error("Unrecognized option 'invalid'") + ); + }); + it("should throw on using le as label", function() { + assert.throws(() => { + let test = new prometheus.Histogram( + "test", "Help", { register: false, labels: ["le", "test"] } + ); + }, new Error("Reserved label 'le'")); + }); + }); + + describe(".observe()", function() { + it("should increment the appropriate bucket", function() { + histogram.observe(2); + assert.deepEqual( + histogram.buckets, + new Map([[1, 0], [2, 1], [3, 1], [4, 1], [Infinity, 1]]) + ); + assert.equal(histogram.sum, 2); + assert.equal(histogram.count, 1); + + histogram.observe(4); + assert.deepEqual( + histogram.buckets, + new Map([[1, 0], [2, 1], [3, 1], [4, 2], [Infinity, 2]]) + ); + assert.equal(histogram.sum, 6); + assert.equal(histogram.count, 2); + }); + }); + + describe(".startTimer()", function() { + it("should add observations when closure is called", function() { + let observeDuration = histogram.startTimer(); + assert.equal(histogram.count, 0); + observeDuration(); + assert.equal(histogram.count, 1); + observeDuration(); + assert.equal(histogram.count, 2); + }); + }); + + describe("static .linear()", function() { + it("should give back the appropriate buckets", function() { + let linear = prometheus.Histogram.linear; + assert.deepEqual(linear(1, 2, 3), [1, 3, 5]); + assert.deepEqual(linear(-1, 4, 3), [-1, 3, 7]); + assert.deepEqual(linear(1, 1, 4), [1, 2, 3, 4]); + }); + }); + + describe("static .exponential()", function() { + it("should give back the appropriate buckets", function() { + let exponential = prometheus.Histogram.exponential; + assert.deepEqual(exponential(1, 2, 3), [1, 2, 4]); + assert.deepEqual(exponential(2, 10, 3), [2, 20, 200]); + assert.deepEqual(exponential(1, 1.5, 4), [1, 1.5, 2.25, 3.375]); + }); + }); + + describe(".labels()", function() { + it("should initialize a value for the label set", function() { + labeledHistogram.labels({ a: "1", b: "2" }); + assert(labeledHistogram._bucketValues.has('a="1",b="2",le="1"'), "bucket value not set"); + assert(labeledHistogram._bucketValues.has('a="1",b="2",le="+Inf"'), "bucket value not set"); + assert(labeledHistogram._sumValues.has('a="1",b="2"'), "sum value not set"); + assert(labeledHistogram._countValues.has('a="1",b="2"'), "count value not set"); + + labeledHistogram.labels("4", "5"); + assert(labeledHistogram._bucketValues.has('a="4",b="5",le="1"'), "bucket value not set"); + assert(labeledHistogram._bucketValues.has('a="4",b="5",le="+Inf"'), "bucket value not set"); + assert(labeledHistogram._sumValues.has('a="4",b="5"'), "sum value not set"); + assert(labeledHistogram._countValues.has('a="4",b="5"'), "count value not set"); + }); + it("should return a child supporting the histogram methods", function() { + let child = labeledHistogram.labels({ a: "1", b: "2" }); + void child.buckets; + void child.sum; + void child.count; + child.observe(2); + let observeDuration = child.startTimer(); + observeDuration(); + child = labeledHistogram.labels("7", "8"); + void child.buckets; + void child.sum; + void child.count; + child.observe(2); + observeDuration = child.startTimer(); + observeDuration(); + }); + it("should return the same child for the same labels", function() { + let child = labeledHistogram.labels("1", "2"); + assert.equal(labeledHistogram.labels("1", "2"), child); + assert.equal(labeledHistogram.labels({ a: "1", b: "2" }), child); + }); + }); + describe(".remove()", function() { + it("should remove a label set created with .labels()", function() { + labeledHistogram.labels("1", "2"); + labeledHistogram.remove("1", "2"); + assert(!labeledHistogram._bucketValues.has('a="1",b="2",le="1"'), "bucket value not removed"); + assert(!labeledHistogram._bucketValues.has('a="1",b="2",le="+Inf"'), "bucket value not removed"); + assert(!labeledHistogram._sumValues.has('a="1",b="2"'), "sum value not removed"); + assert(!labeledHistogram._countValues.has('a="1",b="2"'), "count value not removed"); + }); + }); + describe(".removeAll()", function() { + it("should remove a label set created with .labels()", function() { + labeledHistogram.labels("1", "2"); + labeledHistogram.labels("1", "3"); + labeledHistogram.labels("4", "6"); + labeledHistogram.removeAll({ a: "1" }); + assert.deepEqual( + labeledHistogram._bucketValues, + new Map([['a="4",b="6",le="1"', 0], ['a="4",b="6",le="+Inf"', 0]]) + ); + assert.deepEqual(labeledHistogram._sumValues, new Map([['a="4",b="6"', 0]])); + assert.deepEqual(labeledHistogram._countValues, new Map([['a="4",b="6"', 0]])); + }); + }); + describe(".clear()", function() { + it("should remove all label sets created with .labels()", function() { + labeledHistogram.labels("1", "2"); + labeledHistogram.labels("1", "3"); + labeledHistogram.labels("4", "6"); + labeledHistogram.clear(); + assert(labeledHistogram._bucketValues.size === 0, "buckets were not cleared"); + assert(labeledHistogram._sumValues.size === 0, "sums were not cleared"); + assert(labeledHistogram._countValues.size === 0, "counts were not cleared"); + }); + }); + }); + describe("class ValueCollector", function() { + describe(".get()", function() { + it("should return the current value", function() { + let gauge = new prometheus.Gauge("test", "Help", { register: false }); + gauge.set(11); + assert.equal(gauge.get(), 11); + }); + }); + }); + + describe("class LabeledCollector", function() { describe("constructor", function() { it("should throw on invalid metric name", function() { for (let name of ["", " ", "0a", "$a"]) { @@ -242,14 +429,6 @@ describe("lib/prometheus", function() { }); }); - describe(".get()", function() { - it("should return the current value", function() { - let gauge = new prometheus.Gauge("test", "Help", { register: false }); - gauge.set(11); - assert.equal(gauge.get(), 11); - }); - }); - describe(".labels()", function() { let gauge; beforeEach(function() { @@ -310,6 +489,7 @@ describe("lib/prometheus", function() { it("should return the same child for the same labels", function() { let child = gauge.labels("1", "2"); assert.equal(gauge.labels("1", "2"), child); + assert.equal(gauge.labels({ a: "1", b: "2" }), child); }); }); describe(".remove()", function() { @@ -393,8 +573,9 @@ describe("lib/prometheus", function() { "collector did not give exactly one result" ); + let value = results[0].samples.get("").get(""); assert( - Math.abs(results[0].samples.get("") - Date.now() / 1000) < 1, + Math.abs(value - Date.now() / 1000) < 1, "value set is not close to the current unix time" ); }); @@ -413,7 +594,7 @@ describe("lib/prometheus", function() { "collector did not give exactly one result" ); - let value = results[0].samples.get(""); + let value = results[0].samples.get("").get(""); assert( value < 5 && value > 0, `process time is not between 0 and 5 seconds (${value})` @@ -434,7 +615,7 @@ describe("lib/prometheus", function() { "collector did not give exactly one result" ); - let value = results[0].samples.get(""); + let value = results[0].samples.get("").get(""); assert( value < 200e6 && value > 0, `resident memory is not between 0 and 200 MB (${value})` @@ -455,7 +636,7 @@ describe("lib/prometheus", function() { "collector did not give exactly one result" ); - let value = results[0].samples.get(""); + let value = results[0].samples.get("").get(""); assert( value < 200e6 && value > 0, `heap is not between 0 and 200 MB (${value})` @@ -500,6 +681,59 @@ describe("lib/prometheus", function() { 'test{a="x",b="y"} 0\n' ); }); + it("should format histogram metrics", async function() { + let registry = new prometheus.CollectorRegistry(); + let histogram = new prometheus.Histogram( + "test", "Help", { register: false, buckets: [1, 2, 3, 4] } + ); + for (let value of [2, 2, 5, 2]) { + histogram.observe(value); + } + registry.register(histogram); + let exposition = await prometheus.exposition(registry.collect()); + assert.equal( + exposition, + "# HELP test Help\n" + + "# TYPE test histogram\n" + + 'test_bucket{le="1"} 0\n' + + 'test_bucket{le="2"} 3\n' + + 'test_bucket{le="3"} 3\n' + + 'test_bucket{le="4"} 3\n' + + 'test_bucket{le="+Inf"} 4\n' + + "test_sum 11\n" + + "test_count 4\n" + ); + }); + it("should format histogram metrics with labels", async function() { + let registry = new prometheus.CollectorRegistry(); + let histogram = new prometheus.Histogram( + "test", "Help", { register: false, buckets: [1], labels: ["l"] } + ); + for (let [value, label] of [[1, "a"], [2, "b"], [3, "b"]]) { + histogram.labels(label).observe(value); + } + histogram.labels("c"); + + registry.register(histogram); + let exposition = await prometheus.exposition(registry.collect()); + assert.equal( + exposition, + "# HELP test Help\n" + + "# TYPE test histogram\n" + + 'test_bucket{l="a",le="1"} 1\n' + + 'test_bucket{l="a",le="+Inf"} 1\n' + + 'test_bucket{l="b",le="1"} 0\n' + + 'test_bucket{l="b",le="+Inf"} 2\n' + + 'test_bucket{l="c",le="1"} 0\n' + + 'test_bucket{l="c",le="+Inf"} 0\n' + + 'test_sum{l="a"} 1\n' + + 'test_sum{l="b"} 5\n' + + 'test_sum{l="c"} 0\n' + + 'test_count{l="a"} 1\n' + + 'test_count{l="b"} 2\n' + + 'test_count{l="c"} 0\n' + ); + }); it("should format Infinity and NaN", async function() { let registry = new prometheus.CollectorRegistry(); let gauge = new prometheus.Gauge( @@ -523,38 +757,64 @@ describe("lib/prometheus", function() { let serializedResult = { metric: { type: "type", name: "name", help: "Help", labels: ["a", "b"] }, - samples: [['a="1",b="2"', 1], ['a="3",b="4"', 5]], + samples: [["", [['a="1",b="2"', 1], ['a="3",b="4"', 5]]]], }; let deserializedResult = { metric: new prometheus.Metric("type", "name", "Help", ["a", "b"]), - samples: new Map([['a="1",b="2"', 1], ['a="3",b="4"', 5]]), + samples: new Map([["", new Map([['a="1",b="2"', 1], ['a="3",b="4"', 5]])]]), }; describe("serializeResult()", function() { - it("should convert a result into plain object", function() { + it("should convert real results into the reference serialization", async function() { + let counter = new prometheus.Counter("a_counter", "Counter", { register: false }); + let gauge = new prometheus.Gauge("a_gauge", "Gauge", { register: false, labels: ["a", "b"] }); + let registry = new prometheus.CollectorRegistry(); + registry.register(counter); + registry.register(gauge); + + counter.inc(3); + gauge.labels("1", "2").set(1); + gauge.labels("3", "4").set(5); + + let results = []; + for await (let result of registry.collect()) { + results.push(result); + } + + let referenceSerialization = [ + { + metric: { type: "counter", name: "a_counter", help: "Counter", labels: [] }, + samples: [["", [["", 3]]]], + }, + { + metric: { type: "gauge", name: "a_gauge", help: "Gauge", labels: ["a", "b"] }, + samples: [["", [['a="1",b="2"', 1], ['a="3",b="4"', 5]]]], + }, + ]; + assert.deepEqual( - prometheus.serializeResult(deserializedResult), - serializedResult + results.map(result => prometheus.serializeResult(result)), + referenceSerialization ); }); it("should support adding labels", function() { assert.deepEqual( prometheus.serializeResult({ metric: new prometheus.Metric("type", "name", "Help"), - samples: new Map([["", 2]]), + samples: new Map([["name", new Map([["", 2]])]]), }, { addLabels: { c: "8" }}), { metric: { type: "type", name: "name", help: "Help", labels: ["c"] }, - samples: [['c="8"', 2]], + samples: [["name", [['c="8"', 2]]]], } ); assert.deepEqual( prometheus.serializeResult({ metric: new prometheus.Metric("type", "name", "Help", ["a"]), - samples: new Map([['a="4"', 3]]), + samples: new Map([["", new Map([['a="4"', 3]])]]), }, { addLabels: { c: "8" }}), { metric: { type: "type", name: "name", help: "Help", labels: ["a", "c"] }, - samples: [['a="4",c="8"', 3]], + samples: [["", [['a="4",c="8"', 3]]]], } ); }); @@ -562,21 +822,21 @@ describe("lib/prometheus", function() { assert.deepEqual( prometheus.serializeResult({ metric: new prometheus.Metric("type", "name", "Help"), - samples: new Map([["", 2]]), + samples: new Map([["", new Map([["", 2]])]]), }, { metricName: "new" }), { metric: { type: "type", name: "new", help: "Help", labels: [] }, - samples: [["", 2]], + samples: [["", [["", 2]]]], } ); assert.deepEqual( prometheus.serializeResult({ metric: new prometheus.Metric("type", "name", "Help"), - samples: new Map([["", 2]]), + samples: new Map([["", new Map([["", 2]])]]), }, { metricHelp: "old" }), { metric: { type: "type", name: "name", help: "old", labels: [] }, - samples: [["", 2]], + samples: [["", [["", 2]]]], } ); }); @@ -584,7 +844,7 @@ describe("lib/prometheus", function() { assert.throws( () => prometheus.serializeResult({ metric: new prometheus.Metric("type", "name", "Help"), - samples: new Map([["", 3]]), + samples: new Map([["", new Map([["", 3]])]]), }, { invalid: true }), new Error("Unrecognized option 'invalid'") ); @@ -600,6 +860,34 @@ describe("lib/prometheus", function() { }); }); + describe("integration", function() { + it("should round trip serialize histogram results", async function() { + let registry = new prometheus.CollectorRegistry(); + let histogram = new prometheus.Histogram( + "test", "Help", { register: false, buckets: [1], labels: ["l"] } + ); + for (let [value, label] of [[1, "a"], [2, "b"], [3, "b"]]) { + histogram.labels(label).observe(value); + } + histogram.labels("c"); + + registry.register(histogram); + let results = []; + for await (let result of registry.collect()) { + results.push(result); + } + let exposition = await prometheus.exposition(results); + + let serialized = results.map(res => prometheus.serializeResult(res)); + let jsonRoundtrip = JSON.parse(JSON.stringify(serialized)); + let deserialized = jsonRoundtrip.map(res => prometheus.deserializeResult(res)); + assert.deepEqual(deserialized, results); + let roundTripExposition = await prometheus.exposition(deserialized); + + assert.equal(roundTripExposition, exposition); + }); + }); + describe("post test checks", function() { it("should not have any collectors in the default registry", function() { this.skip(); // XXX Doesn't work in Clusterio's shared testing env diff --git a/test/mock.js b/test/mock.js index c8cddfbb..8d0c0812 100644 --- a/test/mock.js +++ b/test/mock.js @@ -92,6 +92,10 @@ class MockInstance extends libLink.Link { }, }; } + + async sendRcon(command, expectEmpty, plugin) { + return await this.server.sendRcon(command); + } } class MockSlave extends libLink.Link { diff --git a/test/slave.js b/test/slave.js index a14c40f1..47ff8f53 100644 --- a/test/slave.js +++ b/test/slave.js @@ -152,8 +152,7 @@ describe("Slave testing", function() { it("should discover test instance", async function() { let logger = { log: () => {}, error: () => {} }; let instancePath = path.join("test", "file", "instances"); - let instanceInfos = new Map(); - await slave._discoverInstances(instanceInfos, instancePath, logger); + let instanceInfos = await slave._discoverInstances(instancePath, logger); let referenceConfig = new libConfig.InstanceConfig(); await referenceConfig.init();