diff --git a/CHANGELOG.md b/CHANGELOG.md index b5bbfb88..59365897 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/README.md b/README.md index 116f9b94..9298a408 100644 --- a/README.md +++ b/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. diff --git a/clusterctl.js b/clusterctl.js index 89a0509a..6f38905e 100644 --- a/clusterctl.js +++ b/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 = { diff --git a/docs/developing-for-clusterio.md b/docs/developing-for-clusterio.md index 5e2273fe..24734fc2 100644 --- a/docs/developing-for-clusterio.md +++ b/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. diff --git a/docs/socket.md b/docs/socket.md index 597ef535..87d4718a 100644 --- a/docs/socket.md +++ b/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. diff --git a/lib/config/definitions.js b/lib/config/definitions.js index 95f3f737..9c1dbc14 100644 --- a/lib/config/definitions.js +++ b/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 { } diff --git a/lib/errors.js b/lib/errors.js index 2945d13d..19c9d27f 100644 --- a/lib/errors.js +++ b/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, diff --git a/lib/factorio/server.js b/lib/factorio/server.js index 0a765ddc..8f7a6540 100644 --- a/lib/factorio/server.js +++ b/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"); }); diff --git a/lib/link/connectors.js b/lib/link/connectors.js index 276d2922..2b94c0e5 100644 --- a/lib/link/connectors.js +++ b/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(); diff --git a/lib/link/link.js b/lib/link/link.js index 20377652..d36d8b56 100644 --- a/lib/link/link.js +++ b/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(); } } diff --git a/lib/link/messages.js b/lib/link/messages.js index 05305053..365bfa6a 100644 --- a/lib/link/messages.js +++ b/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'], diff --git a/lib/plugin.js b/lib/plugin.js index abd97ad6..41661db3 100644 --- a/lib/plugin.js +++ b/lib/plugin.js @@ -127,7 +127,7 @@ class BaseMasterPlugin { /** * Called when the master server is shutting down */ - async onExit() { } + async onShutdown() { } } /** diff --git a/lib/prometheus.js b/lib/prometheus.js index 2f1386c8..8913f52a 100644 --- a/lib/prometheus.js +++ b/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." diff --git a/master.js b/master.js index 876d9bac..a144945f 100644 --- a/master.js +++ b/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, diff --git a/modules/clusterio/impl.lua b/modules/clusterio/impl.lua index 3ae9593f..52303963 100644 --- a/modules/clusterio/impl.lua +++ b/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 diff --git a/package.json b/package.json index 8646ff7b..b0693d9a 100644 --- a/package.json +++ b/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", diff --git a/plugins/research_sync/master.js b/plugins/research_sync/master.js index deaf10c4..84a00db4 100644 --- a/plugins/research_sync/master.js +++ b/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); } diff --git a/plugins/subspace_storage/master.js b/plugins/subspace_storage/master.js index 3170af23..21e44938 100644 --- a/plugins/subspace_storage/master.js +++ b/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); diff --git a/slave.js b/slave.js index 8b748680..a08d56bd 100644 --- a/slave.js +++ b/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(); } } diff --git a/test/integration/index.js b/test/integration/index.js index 8173bf6e..49546554 100644 --- a/test/integration/index.js +++ b/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 = { diff --git a/test/integration/link.js b/test/integration/link.js index f1b1dba5..5836d507 100644 --- a/test/integration/link.js +++ b/test/integration/link.js @@ -12,7 +12,8 @@ class TestControl extends link.Link { link.attachAllMessages(this); } - async shutdownConnectionRequestHandler() { } + async debugWsMessageEventHandler() { } + async prepareDisconnectRequestHandler() { } async instanceOutputEventHandler() { } } diff --git a/test/lib/factorio/lines.js b/test/lib/factorio/lines.js index eb6159ba..2e316ac0 100644 --- a/test/lib/factorio/lines.js +++ b/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 = { diff --git a/test/lib/plugin.js b/test/lib/plugin.js index 4072aebf..92cd772a 100644 --- a/test/lib/plugin.js +++ b/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(); }) }); diff --git a/test/lib/prometheus.js b/test/lib/prometheus.js index afecec92..f959556b 100644 --- a/test/lib/prometheus.js +++ b/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() {