Merge pull request #340 from Hornwitser/rcon-performance

Improvements from investigating RCON performance

.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",

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
 

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
+
+
 ### <plugin_name>.enabled
 
 Whether to enable the given plugin on the instance.  Note that plugins

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')"
         );
 

packages/ctl/ctl.js

@@ -203,13 +203,13 @@ instanceConfigCommands.add(new libCommand.Command({
 instanceCommands.add(instanceConfigCommands);
 
 instanceCommands.add(new libCommand.Command({
-	definition: ["assign <instance> <slave>", "Assign instance to a slave", (yargs) => {
+	definition: ["assign <instance> [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,

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();
 
 /**

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
 	 *

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

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"],

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);
+	}
 }
 
 /**

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;

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<integer, Object>} 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<integer, Object>}
+ *     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) {

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) {

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) {

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

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 = {

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

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");
+		});
+	});
 });

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

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 {

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();