Merge pull request #284 from Hornwitser/small-things-2

Small things 2

CHANGELOG.md

@@ -37,6 +37,8 @@ Version 2.0.0
   allowing it to be proxied behind web-server.
 - Added WebSocket usage statistics.
 - Added commands sent statistic.
+- Added Factorio server CPU and memory usage statistic.
+- Added Factorio autosave size statistic.
 
 ### Changes
 

README.md

@@ -10,8 +10,6 @@ Discord for development/support/play: https://discord.gg/5XuDkje
 
 ## Important notice
 
-**WARNING: The master branch is currently broken, do not use it!**
-
 This is the development branch for factorioClusterio 2.0 which is currently undergoing heavy
 restructuring and refactoring.  Expect plugins and existing installations to frequently break when
 using this branch.  If you don't want to be an alpha tester for 2.0 please use the stable
@@ -243,15 +241,13 @@ The basic operations to start a new instance is the following
 
     node clusterctl create-instance --name "My instance"
     node clusterctl assign-instance --instance "My instance" --slave "Local"
-    node clusterctl create-save --instance "My instance"
     node clusterctl start-instance --instance "My instance"
 
 The first line creates the instance configuration on the master server.
 The second assigns the instance to a slave which creates the instance
 directory and files needed to run the instance on the given slave.  The
-third line creates a new savegame for the instance.  You could also
-upload your own save to the instance directory instead.  And finally the
-fourth line starts the instance.
+third line starts the instance, which creates a new save if there are no
+save games present.
 
 There are many more commands available with clusterctl.  See
 `node clusterctl --help` for a full list of them.

clusterctl.js

@@ -251,6 +251,7 @@ commands.push(new Command({
 	definition: ['start-instance', "Start instance", (yargs) => {
 		yargs.options({
 			'instance': { describe: "Instance to start", nargs: 1, type: 'string', demandOption: true },
+			'save': { describe: "Save load, defaults to latest", nargs: 1, type: 'string' },
 		});
 	}],
 	handler: async function(args, control) {
@@ -258,6 +259,7 @@ commands.push(new Command({
 		await link.messages.setInstanceOutputSubscriptions.send(control, { instance_ids: [instanceId] });
 		let response = await link.messages.startInstance.send(control, {
 			instance_id: instanceId,
+			save: args.save || null,
 		});
 	},
 }));
@@ -308,6 +310,14 @@ commands.push(new Command({
 	},
 }));
 
+commands.push(new Command({
+	definition: ["debug-dump-ws", "Dump WebSocket messages sent and received by master", (yargs) => { }],
+	handler: async function(args, control) {
+		await link.messages.debugDumpWs.send(control);
+		return new Promise(() => {});
+	},
+}));
+
 // Convert to mapping from name to command instance
 commands = new Map([...commands.map(command => [command.name, command])]);
 
@@ -350,10 +360,22 @@ class Control extends link.Link {
 		console.log(formatOutputColored(output));
 	}
 
+	async debugWsMessageEventHandler(message) {
+		console.log("WS", message.data.direction, message.data.content);
+	}
+
 	async shutdown() {
-		await link.messages.shutdownConnection.send(this);
-		this.connector.close(1001, "Control Quit")
-		await events.once(this.connector, "close");
+		this.connector.setTimeout(30);
+
+		try {
+			await link.messages.prepareDisconnect.send(this);
+		} catch (err) {
+			if (!(err instanceof SessionLost)) {
+				throw err;
+			}
+		}
+
+		await this.connector.close(1001, "Control Quit")
 	}
 }
 
@@ -499,6 +521,14 @@ async function startControl() {
 		throw err;
 	}
 
+	process.on("SIGINT", () => {
+		console.log("Caught interrupt signal, closing connection");
+		control.shutdown().catch(err => {
+			console.error(err);
+			process.exit(1);
+		});
+	});
+
 	if (commands.has(commandName)) {
 		command = commands.get(commandName);
 
@@ -522,8 +552,6 @@ async function startControl() {
 			await control.shutdown();
 		}
 	}
-
-	//XXX control.close("done");
 }
 
 module.exports = {

docs/developing-for-clusterio.md

@@ -383,7 +383,5 @@ inventory.
 Clusterio Plugins
 -----------------
 
-The plugin interface is undergoing a major rewrite, and will be
-documented once that is done.  See
-[issue #220](https://github.com/clusterio/factorioClusterio/issues/220)
-for details.
+See the [Writing Plugins](writing-plugins.md) document for details on
+the plugin interface and how to write plugins for Clusterio.

docs/socket.md

@@ -253,28 +253,29 @@ requests and their contents.
 - error - string - Human readable text message describing the error.
 
 
-Graceful shutdown
------------------
+Graceful Disconnect
+-------------------
 
 To ensure no data is lost when closing the link it's important to
-send connection shutdown request and wait for it to complete before
+send a prepare disconnect request and wait for it to complete before
 closing the WebSocket.  Either party of the link can send this request
 and the party sending the request should be the one that initates the
 close of the WebSocket.
 
-### `connection_shutdown_request`
+### `prepare_disconnect_request`
 
-Request graceful shutdown of the connection.  The receiving party should
+Request graceful disconnect of the connection.  The receiving party should
 stop tasks that send events and requests over the link and wait for
 pending requests it has over the link to complete, but it should not
 stop responding to requests sent to it.  Only after this process is done
 should the response be sent.
 
-### `connection_shutdown_response`
+### `prepare_disconnect_response`
 
-Signals the shutdown is complete and that the connection can be closed.
-The party that initiatied the shutdown request may still send other
-requests/events over the link to finish its part of the shutdown.
+Signals the preparation for disconnect is complete and that the
+connection can be closed.  The party that initiatied the prepare
+disconnect request may still send other requests/events over the link to
+finish its part of the disconnection proceedure.
 
 If the error response variant is received it should be logged if
 possible and treated as a success.

lib/config/definitions.js

@@ -65,6 +65,13 @@ MasterGroup.define({
 	optional: true,
 	initial_value: "database/certificates/cert.key",
 });
+MasterGroup.define({
+	name: "tls_bits",
+	title: "TLS Bits",
+	description: "Number of bits to use for auto generated TLS certificate.",
+	type: "number",
+	initial_value: 2048,
+});
 MasterGroup.define({
 	name: "auth_secret",
 	title: "Master Authentication Secret",
@@ -87,6 +94,13 @@ MasterGroup.define({
 	type: "number",
 	initial_value: 60,
 });
+MasterGroup.define({
+	name: "connector_shutdown_timeout",
+	title: "Connector Shutdown Timeout",
+	description: "Timeout in seconds for connectors to properly disconnect on shutdown",
+	type: "number",
+	initial_value: 30,
+});
 MasterGroup.finalize();
 
 class MasterConfig extends classes.Config { }

lib/errors.js

@@ -14,6 +14,9 @@ class InvalidMessage extends Error {
 	}
 }
 
+// Thrown from requests when the session it was sent on was lost.
+class SessionLost extends Error { }
+
 // Thrown when WebSocket authentication failed
 class AuthenticationFailed extends Error { }
 
@@ -36,6 +39,7 @@ module.exports = {
 	CommandError,
 	RequestError,
 	InvalidMessage,
+	SessionLost,
 	AuthenticationFailed,
 	StartupError,
 	EnvironmentError,

lib/factorio/server.js

@@ -349,6 +349,18 @@ const outputHeuristics = [
 		}
 	},
 
+	// Message indicating the server has started an autosave
+	{
+		filter: {
+			type: 'log',
+			message: /^Saving to _autosave\d+ \((non-)?blocking\)\.$/,
+		},
+		action: function(output) {
+			let name = /^Saving to (_autosave\d+) /.exec(output.message)[1];
+			this.emit('_autosave', name);
+		},
+	},
+
 	// Message indicating the server has finished saving
 	{
 		filter: {
@@ -419,6 +431,8 @@ const outputHeuristics = [
  * output - invoked with parsed output from Factorio
  * rcon-ready - invoked when the RCON client has connected
  * game-ready - invoked when the server is finished starting up
+ * autosave-start - invoked when the server starts an autosave
+ * autosave-fnished - invoked when the autosave finished
  * exit - invoked when the sterver has exited
  */
 class FactorioServer extends events.EventEmitter {
@@ -454,6 +468,20 @@ class FactorioServer extends events.EventEmitter {
 
 		this._stdout = new LineSplitter((line) => { this._handleOutput(line, 'stdout'); });
 		this._stderr = new LineSplitter((line) => { this._handleOutput(line, 'stderr'); });
+
+		// Track autosaving
+		this._runningAutosave = null;
+		this.on("_autosave", function(name) {
+			this.emit("autosave-start", name);
+			this._runningAutosave = name;
+		})
+
+		this.on("_saved", function() {
+			if (this._runningAutosave) {
+				this.emit("autosave-finished", this._runningAutosave);
+				this._runningAutosave = null;
+			}
+		})
 	}
 
 	_check(expectedStates) {
@@ -590,6 +618,16 @@ class FactorioServer extends events.EventEmitter {
 		return this._version;
 	}
 
+	/**
+	 * The pid of the server process, or null if not running
+	 */
+	get pid() {
+		if (!this._server) {
+			return null;
+		}
+		return this._server.pid;
+	}
+
 	_attachStdio() {
 		this._server.stdout.on('data', chunk => { this._stdout.data(chunk); });
 		this._server.stdout.on('close', () => { this._stdout.end(); });
@@ -635,6 +673,7 @@ class FactorioServer extends events.EventEmitter {
 			this._rconReady = false;
 			this._gameReady = false;
 			this._unexpected = [];
+			this._runningAutosave = null;
 
 			this.emit("exit");
 		});

lib/link/connectors.js

@@ -129,6 +129,8 @@ class WebSocketClientConnector extends WebSocketBaseConnector {
 		this._sessionToken = null;
 		this._connected = false;
 		this._reconnectDelay = reconnectDelay;
+		this._timeout = 15 * 60;
+		this._startedReconnect = null;
 	}
 
 	_check(expectedStates) {
@@ -139,6 +141,15 @@ class WebSocketClientConnector extends WebSocketBaseConnector {
 		}
 	}
 
+	setTimeout(timeout) {
+		this._timeout = timeout;
+
+		if (this._reconnectId && this.timeout < this._reconnectDelay / 2) {
+			clearTimeout(this._reconnectId);
+			this.reconnect();
+		}
+	}
+
 	/**
 	 * Send a handshake message over the socket
 	 *
@@ -155,7 +166,7 @@ class WebSocketClientConnector extends WebSocketBaseConnector {
 	 *
 	 * Sends a close frame and disconnects the connector.
 	 */
-	close(code, reason) {
+	async close(code, reason) {
 		if (this._state === "new") {
 			return;
 		}
@@ -168,6 +179,7 @@ class WebSocketClientConnector extends WebSocketBaseConnector {
 		this.stopHeartbeat();
 		this._state = "closing";
 		this._socket.close(code, reason);
+		await events.once(this, "close");
 	}
 
 	/**
@@ -207,6 +219,24 @@ class WebSocketClientConnector extends WebSocketBaseConnector {
 			return;
 		}
 
+		if (this._startedReconnect === null) {
+			this._startedReconnect = Date.now();
+		}
+
+		if (this._startedReconnect + this._timeout * 1000 < Date.now()) {
+			console.log("SOCKET | Timed out trying to reconnect");
+			this._heartbeatInterval = null;
+			this._lastReceivedSeq = null;
+			this._sessionToken = null;
+			this._sendBuffer.length = 0;
+			if (this._connected) {
+				this._connected = false;
+				this.emit("close");
+			}
+			this.emit("invalidate");
+			return;
+		}
+
 		let delay = Math.random() * this._reconnectDelay;
 		console.log(`SOCKET | waiting ${delay.toLocaleString("en", { maximumFractionDigits: 2 })} seconds for reconnect`)
 		this._reconnectId = setTimeout(() => {
@@ -231,18 +261,14 @@ class WebSocketClientConnector extends WebSocketBaseConnector {
 				this._state = "new";
 
 				if (this._connected) {
-					this.emit("close");
 					this._connected = false;
+					this.emit("close");
 				}
 
 			} else {
 				this._state = "handshake";
 				this.reconnect();
-
-				if (this._connected) {
-					this.emit("dropped");
-					this._connected = false;
-				}
+				this.emit("dropped");
 			}
 
 			this.stopHeartbeat();

lib/link/link.js

@@ -45,7 +45,7 @@ class Link {
 		connector.on("invalidate", () => {
 			for (let waiterType of this._waiters.values()) {
 				for (let waiter of waiterType) {
-					waiter.reject(new Error("Session lost"));
+					waiter.reject(new errors.SessionLost("Session Lost"));
 				}
 			}
 
@@ -208,14 +208,14 @@ class Link {
 	async pingRequestHandler() { }
 
 	/**
-	 * Prepare connection for close
+	 * Prepare connection for disconnect
 	 *
 	 * Waits for all pending requests on the connection to resolve.
 	 * Sub-classes should implement this handler with code that prevents
 	 * additional requests to be sent out before calling the super class
 	 * method.
 	 */
-	async shutdownConnectionRequestHandler() {
+	async prepareDisconnectRequestHandler() {
 		let promises = [];
 		for (let waiterType of this._waiters.values()) {
 			for (let waiter of waiterType) {
@@ -224,7 +224,6 @@ class Link {
 		}
 
 		await Promise.all(promises);
-		this.connector.setClosing();
 	}
 }
 

lib/link/messages.js

@@ -185,8 +185,8 @@ let messages = {}
 
 // Connection requests
 let wsLinks = ["master-control", "control-master", "master-slave", "slave-master"];
-messages.shutdownConnection = new Request({
-	type: "shutdown_connection",
+messages.prepareDisconnect = new Request({
+	type: "prepare_disconnect",
 	links: wsLinks,
 });
 
@@ -195,6 +195,12 @@ messages.ping = new Request({
 	links: wsLinks,
 });
 
+messages.debugDumpWs = new Request({
+	type: "debug_dump_ws",
+	links: ["control-master"],
+});
+
+
 // Management requests
 messages.listSlaves = new Request({
 	type: 'list_slaves',
@@ -298,7 +304,7 @@ messages.startInstance = new Request({
 	type: 'start_instance',
 	links: ['control-master', 'master-slave', 'slave-instance'],
 	requestProperties: {
-		"instance_id": { type: "integer" },
+		"save": { type: ["string", "null"] },
 	},
 	forwardTo: 'instance',
 });
@@ -484,6 +490,15 @@ class Event {
 	}
 }
 
+messages.debugWsMessage = new Event({
+	type: "debug_ws_message",
+	links: ["master-control"],
+	eventProperties: {
+		"direction": { type: "string" },
+		"content": { type: "string" },
+	},
+});
+
 messages.instanceStopped = new Event({
 	type: 'instance_stopped',
 	links: ['instance-slave'],

lib/plugin.js

@@ -127,7 +127,7 @@ class BaseMasterPlugin {
 	/**
 	 * Called when the master server is shutting down
 	 */
-	async onExit() { }
+	async onShutdown() { }
 }
 
 /**

lib/prometheus.js

@@ -400,7 +400,8 @@ function deserializeResult(serializedResult) {
 
 let defaultCollectors = {};
 defaultCollectors.processCpuSecondsTotal = new Gauge(
-	'process_cpu_seconds', "Total user and system CPU time spent in seconds.",
+	'process_cpu_seconds_total',
+	"Total user and system CPU time spent in seconds.",
 	{
 		callback: async function() {
 			let usage = process.cpuUsage();
@@ -409,6 +410,26 @@ defaultCollectors.processCpuSecondsTotal = new Gauge(
 	},
 );
 
+defaultCollectors.processResidentMemoryBytes = new Gauge(
+	'process_resident_memory_bytes',
+	"Resident memory size in bytes.",
+	{
+		callback: function() {
+			this.set(process.memoryUsage().rss);
+		},
+	},
+);
+
+defaultCollectors.processHeapBytes = new Gauge(
+	'process_heap_bytes',
+	"Process heap size in bytes.",
+	{
+		callback: function() {
+			this.set(process.memoryUsage().heapUsed);
+		},
+	},
+);
+
 defaultCollectors.processStartTimeSeconds = new Gauge(
 	'process_start_time_seconds',
 	"Start time of the process since unix epoch in seconds."

master.js

@@ -29,6 +29,8 @@ const version = require("./package").version;
 // ugly globals
 let masterConfig;
 let masterConfigPath;
+let stopAcceptingNewSessions = false;
+let debugEvents = new events.EventEmitter();
 
 // homebrew modules
 const generateSSLcert = require("lib/generateSSLcert");
@@ -249,31 +251,44 @@ async function shutdown() {
 
 		for (let [pluginName, masterPlugin] of masterPlugins) {
 			let startTime = Date.now();
-			await masterPlugin.onExit();
-			console.log(`Plugin ${pluginName} exited in ${Date.now() - startTime}ms`);
+			await masterPlugin.onShutdown();
+			console.log(`Plugin ${pluginName} shutdown in ${Date.now() - startTime}ms`);
 		}
 
+		stopAcceptingNewSessions = true;
+
+		let disconnectTasks = [];
 		for (let controlConnection of controlConnections) {
-			controlConnection.connector.close(1001, "Server Quit");
+			controlConnection.connector.setTimeout(masterConfig.get("master.connector_shutdown_timeout"));
+			disconnectTasks.push(controlConnection.disconnect(1001, "Server Quit"));
 		}
 
 		for (let slaveConnection of slaveConnections.values()) {
-			slaveConnection.connector.close(1001, "Server Quit");
+			slaveConnection.connector.setTimeout(masterConfig.get("master.connector_shutdown_timeout"));
+			disconnectTasks.push(slaveConnection.disconnect(1001, "Server Quit"));
+		}
+
+		console.log(`Waiting for ${disconnectTasks.length} connectors to close`);
+		for (let task of disconnectTasks) {
+			try {
+				await task;
+			} catch (err) {
+				if (!(err instanceof errors.SessionLost)) {
+					console.log("Unexpected error disconnecting connector");
+					console.log(err);
+				}
+			}
 		}
 
 		for (let socket of pendingSockets) {
 			socket.close(1001, "Server Quit");
 		}
 
-		if (httpServer) {
-			console.log("Stopping HTTP server");
-			await new Promise(resolve => httpServer.close(resolve));
-		}
-
-		if (httpsServer) {
-			console.log("Stopping HTTPS server");
-			await new Promise(resolve => httpsServer.close(resolve));
-		}
+		let stopTasks = [];
+		console.log("Stopping HTTP(S) server");
+		if (httpServer) { stopTasks.push(new Promise(resolve => httpServer.close(resolve))); }
+		if (httpsServer) { stopTasks.push(new Promise(resolve => httpsServer.close(resolve))); }
+		await Promise.all(stopTasks);
 
 		console.log(`Clusterio cleanly exited in ${Date.now() - exitStartTime}ms`);
 
@@ -380,10 +395,22 @@ class BaseConnection extends link.Link {
 		}
 	}
 
-	async shutdown(code, reason) {
-		await link.messages.shutdownConnection.send(this);
-		this.connector.close(code, reason);
-		await events.once(this.connector, "close");
+	async prepareDisconnectRequestHandler(message, request) {
+		this.connector.setClosing();
+		return await super.prepareDisconnectRequestHandler(message, request);
+	}
+
+	async disconnect(code, reason) {
+		try {
+			await link.messages.prepareDisconnect.send(this);
+		} catch (err) {
+			if (!(err instanceof errors.SessionLost)) {
+				console.error("Unexpected error preparing disconnect");
+				console.error(err);
+			}
+		}
+
+		await this.connector.close(code, reason);
 	}
 }
 
@@ -403,6 +430,16 @@ class ControlConnection extends BaseConnection {
 		});
 
 		this.instanceOutputSubscriptions = new Set();
+
+		this.ws_dumper = null;
+		this.connector.on("connected", () => {
+			this.connector._socket.clusterio_ignore_dump = !!this.ws_dumper;
+		});
+		this.connector.on("close", () => {
+			if (this.ws_dumper) {
+				debugEvents.off("message", this.ws_dumper);
+			}
+		});
 	}
 
 	async listSlavesRequestHandler(message) {
@@ -525,6 +562,14 @@ class ControlConnection extends BaseConnection {
 	async setInstanceOutputSubscriptionsRequestHandler(message) {
 		this.instanceOutputSubscriptions = new Set(message.data.instance_ids);
 	}
+
+	async debugDumpWsRequestHandler(message) {
+		this.ws_dumper = data => {
+			link.messages.debugWsMessage.send(this, data);
+		};
+		this.connector._socket.clusterio_ignore_dump = true;
+		debugEvents.on("message", this.ws_dumper)
+	}
 }
 
 var slaveConnections = new Map();
@@ -617,12 +662,13 @@ class WebSocketServerConnector extends link.WebSocketBaseConnector {
 		this._sessionId = sessionId;
 
 		// The following states are used in the server connector
-		// handshake: Wating for client to (re)connect.
+		// handshake: Waiting for client to (re)connect.
 		// connected: Connection is online
 		// closing: Connection is in the process of being closed.
 		// closed: Connection has been closed
 		this._state = "handshake";
 		this._connected = false;
+		this._timeout = 15 * 60;
 	}
 
 	_check(expectedStates) {
@@ -665,6 +711,11 @@ class WebSocketServerConnector extends link.WebSocketBaseConnector {
 		this._socket.terminate();
 		this._socket = socket;
 
+		if (this._timeoutId) {
+			clearTimeout(this._timeoutId);
+			this._timeoutId = null;
+		}
+
 		this._heartbeatInterval = masterConfig.get("master.heartbeat_interval")
 		this._socket.send(JSON.stringify({
 			seq: null,
@@ -685,6 +736,26 @@ class WebSocketServerConnector extends link.WebSocketBaseConnector {
 		this.emit("connected");
 	}
 
+	setTimeout(timeout) {
+		if (this._timeoutId) {
+			clearTimeout(this._timeoutId);
+		}
+
+		this._timeoutId = setTimeout(() => { this._timedOut() }, timeout * 1000);
+		this._timeout = timeout;
+	}
+
+	_timedOut() {
+		console.log("SOCKET | Connection timed out");
+		this._timeoutId = null;
+		this._lastReceivedSeq = null;
+		this._sendBuffer.length = 0;
+		this._state = "closed";
+		activeConnectors.delete(this._sessionId);
+		this.emit("close");
+		this.emit("invalidate");
+	}
+
 	_attachSocketHandlers() {
 		this.startHeartbeat();
 
@@ -697,17 +768,19 @@ class WebSocketServerConnector extends link.WebSocketBaseConnector {
 				activeConnectors.delete(this._sessionId);
 
 				if (this._connected) {
-					this.emit("close");
 					this._connected = false;
+					this.emit("close");
+				}
+
+				if (this._timeoutId) {
+					clearTimeout(this._timeoutId);
+					this._timeoutId = null;
 				}
 
 			} else {
 				this._state = "handshake";
-
-				if (this._connected) {
-					this.emit("dropped");
-					this._connected = false;
-				}
+				this.emit("dropped");
+				this._timeoutId = setTimeout(() => { this._timedOut() }, this._timeout * 1000);
 			}
 
 			this.stopHeartbeat();
@@ -754,14 +827,15 @@ class WebSocketServerConnector extends link.WebSocketBaseConnector {
 	 *
 	 * Sends a close frame and disconnects the connector.
 	 */
-	close(code, reason) {
-		if (this._state === "new") {
+	async close(code, reason) {
+		if (this._state === "closed") {
 			return;
 		}
 
 		this.stopHeartbeat();
 		this._state = "closing";
 		this._socket.close(code, reason);
+		await events.once(this, "close")
 	}
 }
 
@@ -773,10 +847,18 @@ wss.on("connection", function (socket, req) {
 
 	// Track statistics
 	wsConnectionsCounter.inc();
-	socket.on("message", () => { wsMessageCounter.labels("in").inc(); });
+	socket.on("message", (message) => {
+		wsMessageCounter.labels("in").inc();
+		if (!socket.clusterio_ignore_dump) {
+			debugEvents.emit("message", { direction: "in", content: message });
+		}
+	});
 	let originalSend = socket.send;
 	socket.send = (...args) => {
 		wsMessageCounter.labels("out").inc();
+		if (typeof args[0] === "string" && !socket.clusterio_ignore_dump) {
+			debugEvents.emit("message", { direction: "out", content: args[0] });
+		}
 		return originalSend.call(socket, ...args);
 	};
 
@@ -827,14 +909,14 @@ async function handleHandshake(message, socket, req, attachHandler) {
 	} catch (err) {
 		console.log(`SOCKET | closing ${req.socket.remoteAddress} after receiving invalid JSON`);
 		wsRejectedConnectionsCounter.inc();
-		socket.close(1001, "Invalid JSON");
+		socket.close(1002, "Invalid JSON");
 		return;
 	}
 
 	if (!schema.clientHandshake(message)) {
 		console.log(`SOCKET | closing ${req.socket.remoteAddress} after receiving invalid handshake:`, message);
 		wsRejectedConnectionsCounter.inc();
-		socket.close(1001, "Bad handshake");
+		socket.close(1002, "Bad handshake");
 		return;
 	}
 
@@ -864,6 +946,13 @@ async function handleHandshake(message, socket, req, attachHandler) {
 		return
 	}
 
+	if (stopAcceptingNewSessions) {
+		console.log(`SOCKET | closing ${req.socket.remoteAddress}, server is shutting down`);
+		wsRejectedConnectionsCounter.inc();
+		socket.close(1001, "Shutting down");
+		return;
+	}
+
 	let result = await authenticate.check(data.token);
 
 	if (!result.ok) {
@@ -882,7 +971,7 @@ async function handleHandshake(message, socket, req, attachHandler) {
 		let connection = slaveConnections.get(data.id);
 		if (connection) {
 			console.log(`SOCKET | disconnecting existing connection for slave ${data.id}`);
-			await connection.shutdown();
+			await connection.disconnect(1008, "Registered from another connection");
 		}
 
 		console.log(`SOCKET | registered slave ${data.id} version ${data.version}`);
@@ -1066,6 +1155,7 @@ async function startServer() {
 	// Create a self signed certificate if the certificate files doesn't exist
 	if (httpsPort && !await fs.exists(tls_cert) && !await fs.exists(tls_key)) {
 		await generateSSLcert({
+			bits: masterConfig.get("master.tls_bits"),
 			sslCertPath: tls_cert,
 			sslPrivKeyPath: tls_key,
 			doLogging: true,

modules/clusterio/impl.lua

@@ -18,6 +18,9 @@ impl.events[api.events.on_server_startup] = function()
             instance_name = nil,
         }
     end
+
+    -- Replay will just desync after the save is patched
+    game.disable_replay()
 end
 
 -- Internal API

package.json

@@ -54,6 +54,7 @@
     "needle": "^2.4.0",
     "node-factorio-api": "^0.3.8",
     "node-forge": "^0.7.1",
+    "pidusage": "^2.0.18",
     "rcon-client": "^4.2.0",
     "request": "^2.83.0",
     "sanitizer": "^0.1.3",

plugins/research_sync/master.js

@@ -12,7 +12,7 @@ class MasterPlugin extends plugin.BaseMasterPlugin {
 		this.progressToBroadcast = new Set();
 	}
 
-	async onExit() {
+	async onShutdown() {
 		if (this.progressBroadcastId) {
 			clearTimeout(this.progressBroadcastId);
 		}

plugins/subspace_storage/master.js

@@ -177,7 +177,7 @@ class MasterPlugin extends plugin.BaseMasterPlugin {
 		}
 	}
 
-	async onExit() {
+	async onShutdown() {
 		clearInterval(this.autosaveId);
 		clearInterval(this.doleMagicId);
 		await saveDatabase(this.master.config, this.items);

slave.js

@@ -2,6 +2,7 @@ const fs = require('fs-extra');
 const path = require('path');
 const yargs = require("yargs");
 const events = require("events");
+const pidusage = require("pidusage");
 const version = require("./package").version;
 
 // internal libraries
@@ -22,6 +23,24 @@ const instanceRconCommandsCounter = new prometheus.Counter(
 	{ labels: ["instance_id"] }
 );
 
+const instanceFactorioCpuTime = new prometheus.Gauge(
+	"clusterio_instance_factorio_cpu_time_total",
+	"Factorio CPU time spent in seconds.",
+	{ labels: ["instance_id"] }
+);
+
+const instanceFactorioMemoryUsage = new prometheus.Gauge(
+	"clusterio_instance_factorio_resident_memory_bytes",
+	"Factorio resident memory size in bytes.",
+	{ labels: ["instance_id"] }
+);
+
+const instanceFactorioAutosaveSize = new prometheus.Gauge(
+	"clusterio_instance_factorio_autosave_bytes",
+	"Size of Factorio server autosave in bytes.",
+	{ labels: ["instance_id"] }
+);
+
 /**
  * Keeps track of the runtime parameters of an instance
  */
@@ -41,6 +60,7 @@ class Instance extends link.Link{
 			rconPassword: this.config.get('factorio.rcon_password'),
 		};
 
+		this._running = false;
 		this.server = new factorio.FactorioServer(
 			factorioDir, this._dir, serverOptions
 		);
@@ -65,16 +85,38 @@ class Instance extends link.Link{
 			console.log(`Error in instance ${this.name}:`, err);
 		});
 
-		this.server.on("exit", err => {
-			link.messages.instanceStopped.send(this);
-
-			// Notify plugins of exit
-			for (let pluginInstance of this.plugins.values()) {
-				pluginInstance.onExit();
-			}
+		this.server.on("autosave-finished", name => {
+			this._autosave(name).catch(err => {
+				console.error("Error handling autosave-finished:", err);
+			});
 		});
 	}
 
+	async _autosave(name) {
+		let stat = await fs.stat(this.path("saves", `${name}.zip`));
+		instanceFactorioAutosaveSize.labels(String(this.config.get("instance.id"))).set(stat.size);
+	}
+
+	notifyExit() {
+		this._running = false;
+		link.messages.instanceStopped.send(this);
+
+		// Clear metrics this instance is exporting
+		for (let collector of prometheus.defaultRegistry.collectors) {
+			if (
+				collector instanceof prometheus.ValueCollector
+				&& collector.metric.labels.includes('instance_id')
+			) {
+				collector.removeAll({ instance_id: String(this.config.get("instance.id")) });
+			}
+		}
+
+		// Notify plugins of exit
+		for (let pluginInstance of this.plugins.values()) {
+			pluginInstance.onExit();
+		}
+	}
+
 	async init(pluginInfos, slave) {
 		await this.server.init();
 
@@ -131,7 +173,6 @@ class Instance extends link.Link{
 	 * @param {String} instanceDir -
 	 *     Directory to create the new instance into.
 	 * @param {String} factorioDir - Path to factorio installation.
-	 * @param {Object} options - Options for new instance.
 	 */
 	static async create(instanceConfig, instanceDir, factorioDir) {
 		console.log(`Clusterio | Creating ${instanceDir}`);
@@ -140,7 +181,18 @@ class Instance extends link.Link{
 		await fs.ensureDir(path.join(instanceDir, "saves"));
 	}
 
-	async start(saveName, slaveConfig, socket) {
+	/**
+	 * Prepare a save for starting
+	 *
+	 * Writes server settings, links mods, creates a new save if no save is
+	 * passed, and patches it with modules.
+	 *
+	 * @param {String|null} saveName -
+	 *     Save to prepare from the instance saves directory.  Creates a new
+	 *     save if null.
+	 * @returns {String} Name of the save prepared.
+	 */
+	async prepare(saveName) {
 		console.log("Clusterio | Writing server-settings.json");
 		await this.writeServerSettings();
 
@@ -158,6 +210,20 @@ class Instance extends link.Link{
 
 		await symlinkMods(this, "sharedMods", console);
 
+		// Use latest save if no save was specified
+		if (saveName === null) {
+			saveName = await fileOps.getNewestFile(
+				this.path("saves"), (name) => !name.endsWith('.tmp.zip')
+			);
+		}
+
+		// Create save if no save was found.
+		if (saveName === null) {
+			console.log("Clusterio | Creating new save")
+			await this.server.create("world.zip");
+			saveName = "world.zip";
+		}
+
 		// Patch save with lua modules from plugins
 		console.log("Clusterio | Patching save");
 
@@ -218,11 +284,22 @@ class Instance extends link.Link{
 		}
 
 		await factorio.patch(this.path("saves", saveName), [...modules.values()]);
+		return saveName;
+	}
 
+	/**
+	 * Start Factorio server
+	 *
+	 * Launches the Factorio server for this instance with the given save.
+	 *
+	 * @param {String} saveName - Name of save game to load.
+	 */
+	async start(saveName) {
 		this.server.on('rcon-ready', () => {
 			console.log("Clusterio | RCON connection established");
 		});
 
+		this.server.on("exit", () => this.notifyExit());
 		await this.server.start(saveName);
 		await this.server.disableAchievements()
 		await this.updateInstanceData();
@@ -230,6 +307,8 @@ class Instance extends link.Link{
 		for (let pluginInstance of this.plugins.values()) {
 			await pluginInstance.onStart();
 		}
+
+		this._running = true;
 	}
 
 	/**
@@ -245,6 +324,8 @@ class Instance extends link.Link{
 	 * Stop the instance
 	 */
 	async stop() {
+		this._running = false;
+
 		// XXX this needs more thought to it
 		if (this.server._state === "running") {
 			for (let pluginInstance of this.plugins.values()) {
@@ -253,53 +334,62 @@ class Instance extends link.Link{
 
 			await this.server.stop();
 		}
-
-		// Clear metrics this instance is exporting
-		for (let collector of prometheus.defaultRegistry.collectors) {
-			if (
-				collector instanceof prometheus.ValueCollector
-				&& collector.metric.labels.includes('instance_id')
-			) {
-				collector.removeAll({ instance_id: String(this.config.get("instance.id")) });
-			}
-		}
 	}
 
 	async getMetricsRequestHandler() {
 		let results = []
-		for (let pluginInstance of this.plugins.values()) {
-			let pluginResults = await pluginInstance.onMetrics();
-			if (pluginResults !== undefined) {
-				for await (let result of pluginResults) {
-					results.push(prometheus.serializeResult(result))
+		if (this._running) {
+			for (let pluginInstance of this.plugins.values()) {
+				let pluginResults = await pluginInstance.onMetrics();
+				if (pluginResults !== undefined) {
+					for await (let result of pluginResults) {
+						results.push(prometheus.serializeResult(result))
+					}
 				}
 			}
 		}
 
+		let pid = this.server.pid;
+		if (pid) {
+			let stats = await pidusage(pid);
+			instanceFactorioCpuTime.labels(String(this.config.get("instance.id"))).set(stats.ctime / 1000);
+			instanceFactorioMemoryUsage.labels(String(this.config.get("instance.id"))).set(stats.memory);
+		}
+
 		return { results };
 	}
 
-	async startInstanceRequestHandler() {
-		// Find save to start
-		let latestSave = await fileOps.getNewestFile(
-			this.path("saves"), (name) => !name.endsWith('.tmp.zip')
-		);
-		if (latestSave === null) {
-			throw new errors.RequestError(
-				"No savefile was found to start the instance with."
-			);
+	async startInstanceRequestHandler(message) {
+		let saveName = message.data.save;
+		try {
+			saveName = await this.prepare(saveName)
+		} catch (err) {
+			this.notifyExit();
+			throw err;
 		}
 
-		await this.start(latestSave, this.config, this.socket);
+		try {
+			await this.start(saveName, this.config, this.socket);
+		} catch (err) {
+			await this.stop();
+			throw err;
+		}
 	}
 
 	async createSaveRequestHandler() {
-		console.log("Clusterio | Writing server-settings.json");
-		await this.writeServerSettings();
+		try {
+			console.log("Clusterio | Writing server-settings.json");
+			await this.writeServerSettings();
 
-		console.log("Creating save .....");
-		await symlinkMods(this, "sharedMods", console);
+			console.log("Creating save .....");
+			await symlinkMods(this, "sharedMods", console);
 
+		} catch (err) {
+			this.notifyExit();
+			throw err;
+		}
+
+		this.server.on("exit", () => this.notifyExit());
 		await this.server.create("world");
 		console.log("Clusterio | Successfully created save");
 	}
@@ -413,7 +503,6 @@ class InstanceConnection extends link.Link {
 				continue; // Do not broadcast back to the source
 			}
 
-			console.log("Broadcasting event");
 			event.send(instanceConnection, message.data);
 		}
 	}
@@ -600,13 +689,7 @@ class Slave extends link.Link {
 	async startInstanceRequestHandler(message, request) {
 		let instanceId = message.data.instance_id;
 		let instanceConnection = await this._connectInstance(instanceId);
-		try {
-			return await request.send(instanceConnection, message.data);
-
-		} catch (err) {
-			await this.stopInstance(instanceId);
-			throw err;
-		}
+		return await request.send(instanceConnection, message.data);
 	}
 
 	async createSaveRequestHandler(message, request) {
@@ -656,13 +739,28 @@ class Slave extends link.Link {
 		await this.updateInstances();
 	}
 
+	/**
+	 * Stops all instances and closes the connection
+	 */
 	async shutdown() {
-		await link.messages.shutdownConnection.send(this);
+		this.connector.setTimeout(30);
+
+		try {
+			await link.messages.prepareDisconnect.send(this);
+		} catch (err) {
+			if (!(err instanceof errors.SessionLost)) {
+				console.error("Unexpected error preparing disconnect");
+				console.error(err);
+			}
+		}
+
 		for (let instanceId of this.instanceConnections.keys()) {
 			await this.stopInstance(instanceId);
 		}
-		this.connector.close(1001, "Slave Shutdown");
-		await events.once(this.connector, "close");
+		await this.connector.close(1001, "Slave Shutdown");
+
+		// Clear silly interval in pidfile library.
+		pidusage.clear();
 	}
 }
 

test/integration/index.js

@@ -77,7 +77,11 @@ before(async function() {
 	await exec(`node master --config ${masterConfigPath} config set master.auth_secret TestSecretDoNotUse`);
 	await exec(`node master --config ${masterConfigPath} config set master.http_port 8880`);
 	await exec(`node master --config ${masterConfigPath} config set master.https_port 4443`);
+	await exec(`node master --config ${masterConfigPath} config set master.tls_certificate ${path.join(databaseDir, "cert.crt")}`);
+	await exec(`node master --config ${masterConfigPath} config set master.tls_private_key ${path.join(databaseDir, "cert.key")}`);
+	await exec(`node master --config ${masterConfigPath} config set master.tls_bits 1024`);
 	await exec(`node master --config ${masterConfigPath} config set master.heartbeat_interval 0.25`);
+	await exec(`node master --config ${masterConfigPath} config set master.connector_shutdown_timeout 2`);
 
 	await exec(`node slave --config ${slaveConfigPath} config set slave.id 4`);
 	await exec(`node slave --config ${slaveConfigPath} config set slave.name slave`);
@@ -106,6 +110,11 @@ after(async function() {
 	}
 });
 
+// Ensure the test processes are stopped.
+process.on("exit", () => {
+	if (slaveProcess) { slaveProcess.kill(); }
+	if (masterProcess) { masterProcess.kill(); }
+});
 
 
 module.exports = {

test/integration/link.js

@@ -12,7 +12,8 @@ class TestControl extends link.Link {
 		link.attachAllMessages(this);
 	}
 
-	async shutdownConnectionRequestHandler() { }
+	async debugWsMessageEventHandler() { }
+	async prepareDisconnectRequestHandler() { }
 	async instanceOutputEventHandler() { }
 }
 

test/lib/factorio/lines.js

@@ -109,6 +109,28 @@ let testLines = new Map([
 			message: "Error while running event level::on_nth_tick(1)",
 		}
 	],
+	[
+		"4802.763 Info AppManager.cpp:287: Saving to _autosave4 (blocking).",
+		{
+			format: "seconds",
+			time: "4802.763",
+			type: "log",
+			level: "Info",
+			file: "AppManager.cpp:287",
+			message: "Saving to _autosave4 (blocking).",
+		}
+	],
+	[
+		"4202.780 Info AppManagerStates.cpp:1802: Saving finished",
+		{
+			format: "seconds",
+			time: "4202.780",
+			type: "log",
+			level: "Info",
+			file: "AppManagerStates.cpp:1802",
+			message: "Saving finished",
+		}
+	],
 ])
 
 module.exports = {

test/lib/plugin.js

@@ -16,6 +16,7 @@ describe("lib/plugin", function() {
 			await instancePlugin.init();
 		})
 		it("should define defaults for hooks", async function() {
+			await instancePlugin.onMetrics();
 			await instancePlugin.onStart();
 			await instancePlugin.onStop();
 			instancePlugin.onExit();
@@ -30,7 +31,8 @@ describe("lib/plugin", function() {
 			await masterPlugin.init();
 		})
 		it("should define defaults for hooks", async function() {
-			await masterPlugin.onExit();
+			await masterPlugin.onMetrics();
+			await masterPlugin.onShutdown();
 		})
 	});
 

test/lib/prometheus.js

@@ -312,7 +312,7 @@ describe("lib/prometheus", function() {
 			});
 		});
 		describe(".remove()", function() {
-			it("should remove a label set created with .label()", function() {
+			it("should remove a label set created with .labels()", function() {
 				let gauge = new prometheus.Gauge(
 					'test', "Help", { register: false, labels: ['a', 'b'] }
 				);
@@ -328,7 +328,7 @@ describe("lib/prometheus", function() {
 			});
 		});
 		describe(".removeAll()", function() {
-			it("should remove a label set created with .label()", function() {
+			it("should remove a label set created with .labels()", function() {
 				let gauge = new prometheus.Gauge(
 					'test', "Help", { register: false, labels: ['a', 'b'] }
 				);
@@ -358,7 +358,7 @@ describe("lib/prometheus", function() {
 			});
 		});
 		describe(".clear()", function() {
-			it("should remove all label sets created with .label()", function() {
+			it("should remove all label sets created with .labels()", function() {
 				let gauge = new prometheus.Gauge(
 					'test', "Help", { register: false, labels: ['a', 'b'] }
 				);
@@ -419,6 +419,48 @@ describe("lib/prometheus", function() {
 				);
 			});
 		});
+		describe("processResidentMemoryBytes", function() {
+			it("should give a low value", async function() {
+				this.skip(); // XXX Doesn't work in Clusterio's shared testing env
+				let results = []
+				let collector = prometheus.defaultCollectors.processResidentMemoryBytes;
+				for await (let result of collector.collect()) {
+					results.push(result);
+				}
+
+				assert(
+					results.length == 1,
+					"collector did not give exactly one result"
+				);
+
+				let value = results[0].samples.get('')
+				assert(
+					value < 200e6 && value > 0,
+					`resident memory is not between 0 and 200 MB (${value})`
+				);
+			});
+		});
+		describe("processHeapBytes", function() {
+			it("should give a low value", async function() {
+				this.skip(); // XXX Doesn't work in Clusterio's shared testing env
+				let results = []
+				let collector = prometheus.defaultCollectors.processHeapBytes;
+				for await (let result of collector.collect()) {
+					results.push(result);
+				}
+
+				assert(
+					results.length == 1,
+					"collector did not give exactly one result"
+				);
+
+				let value = results[0].samples.get('')
+				assert(
+					value < 200e6 && value > 0,
+					`heap is not between 0 and 200 MB (${value})`
+				);
+			});
+		});
 	});
 
 	describe("exposition()", function() {