From 29a3c26b4463336a0dba9b4d85f84cdf553e8269 Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Wed, 9 Dec 2020 04:02:02 +0100 Subject: [PATCH 1/7] Fix usage of logger in FactorioServer Fix incorrect usage of the global logger, and typo'd use of the _logger property in FactorioServer. --- packages/lib/factorio/server.js | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/packages/lib/factorio/server.js b/packages/lib/factorio/server.js index 69bdc04d..221429f1 100644 --- a/packages/lib/factorio/server.js +++ b/packages/lib/factorio/server.js @@ -637,7 +637,7 @@ class FactorioServer extends events.EventEmitter { } if (!this.emit(`ipc-${channel}`, content)) { - this.logger.warn(`Warning: Unhandled ipc-${channel}`, { content }); + this._logger.warn(`Warning: Unhandled ipc-${channel}`, { content }); } } @@ -976,7 +976,7 @@ class FactorioServer extends events.EventEmitter { let timeoutId = setTimeout(() => { if (hanged) { - logger.error("Factorio appears to have hanged, sending SIGKILL"); + this._logger.error("Factorio appears to have hanged, sending SIGKILL"); if (this._rconClient) { this._rconClient.end().catch(() => {}); } From b681ea72432b9b3a28508bf47b4d6529ea0509ad Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Thu, 3 Dec 2020 19:08:45 +0100 Subject: [PATCH 2/7] Add RateLimiter helper Add helper class for implementing rate limited operations. --- packages/lib/RateLimiter.js | 104 ++++++++++++++++++++++++++++++++ plugins/research_sync/master.js | 32 +++------- test/lib/RateLimiter.js | 64 ++++++++++++++++++++ 3 files changed, 177 insertions(+), 23 deletions(-) create mode 100644 packages/lib/RateLimiter.js create mode 100644 test/lib/RateLimiter.js diff --git a/packages/lib/RateLimiter.js b/packages/lib/RateLimiter.js new file mode 100644 index 00000000..69adf335 --- /dev/null +++ b/packages/lib/RateLimiter.js @@ -0,0 +1,104 @@ +"use strict"; + + +/** + * Helper class for rate limiting actions + * + * Provides a simple interface for implementing actions that are rate + * limited on invocations. + * + * @example + * const rateLimiter = RateLimiter({ + * action: () => { console.log("action"); }, + * maxRate: 1, + * }); + * + * for (let i = 0; i < 100; i++) { + * rateLimiter.activate(); + * // Prints action twice to the console, once for the first activation + * // and one more time 1 second later for 99 other activations that were + * // exceeding the rate limit. + * } + * @alias module:lib/RateLimiter + */ +class RateLimiter { + /** + * Construct RateLimiter helper + * + * The options passed may also be modified on the instance at runtime. + * + * @param {object=} options - Options for this rate limiter + * @param {number=} options.maxRate - + * Maximum rate of activations per second. + * @param {function()=} options.action - + * Callback invoked at maste `maxRate` per second on activation. + */ + constructor(options = {}) { + /* + * Maximum rate of activations per second. + * @type {number} + */ + this.maxRate = options.maxRate || 1; + + /** + * Callback invoked at most `maxRate` per second on activation. + * @type {function()} + */ + this.action = options.action || null; + + this._lastRun = 0; + this._runTimeout = null; + this.lastProgressBroadcast = Date.now(); + } + + /** + * Activate the rate limited action + * + * Checks if the invocation of the rate limited action now would fall + * within the rate limit. If it is within the limit then the action is + * invoked immediatly, otherwise a timeout is registered to invoke the + * action at next point in time where it will be within the rate limit. + * + * @returns {boolean} true if activation was within the rate limit. + */ + activate() { + let now = Date.now(); + if (now < this._lastRun + 1000 / this.maxRate) { + if (!this._runTimeout && this.action !== null) { + this._runTimeout = setTimeout(() => { + this._runTimeout = null; + this._lastRun = Date.now(); + this.action(); + }, this._lastRun + 1000 / this.maxRate - now); + } + return false; + } + + // We may end up activating before the timeout does. + if (this._runTimeout) { + clearTimeout(this._runTimeout); + this._runTimeout = null; + } + + this._lastRun = now; + if (this.action !== null) { + this.action(); + } + return true; + } + + /** + * Cancel pending action + * + * Cancel pending action invocation. This is necessary when shutting + * down the use of the rate limit as a timeout is registered to invoke + * the next action when activated faster than the rate limit allows. + */ + cancel() { + if (this._runTimeout) { + clearTimeout(this._runTimeout); + } + } +} + +module.exports = RateLimiter; diff --git a/plugins/research_sync/master.js b/plugins/research_sync/master.js index 1475680a..9d735ceb 100644 --- a/plugins/research_sync/master.js +++ b/plugins/research_sync/master.js @@ -3,41 +3,27 @@ const fs = require("fs-extra"); const path = require("path"); const libPlugin = require("@clusterio/lib/plugin"); +const RateLimiter = require("@clusterio/lib/RateLimiter"); class MasterPlugin extends libPlugin.BaseMasterPlugin { async init() { this.technologies = await loadTechnologies(this.master.config, this.logger); + this.progressRateLimiter = new RateLimiter({ + maxRate: 1, + action: () => this.broadcastProgress(), + }); + this.lastProgressBroadcast = Date.now(); this.progressBroadcastId = null; this.progressToBroadcast = new Set(); } async onShutdown() { - if (this.progressBroadcastId) { - clearTimeout(this.progressBroadcastId); - } + this.progressRateLimiter.cancel(); await saveTechnologies(this.master.config, this.technologies, this.logger); } - registerProgress() { - // Rate limit progress broadcasts to one per second - if (Date.now() < this.lastProgressBroadcast + 1000) { - if (!this.progressBroadcastId) { - this.progressBroadcastId = setTimeout(() => { - this.broadcastProgress(); - this.progressBroadcastId = null; - }, 2000); - } - - } else { - if (this.progressBroadcastId) { - clearTimeout(this.progressBroadcastId); - } - this.broadcastProgress(); - } - } - broadcastProgress() { let techs = []; for (let name of this.progressToBroadcast) { @@ -81,7 +67,7 @@ class MasterPlugin extends libPlugin.BaseMasterPlugin { if (newProgress < 1) { tech.progress = newProgress; this.progressToBroadcast.add(name); - this.registerProgress(); + this.progressRateLimiter.activate(); } else { tech.researched = true; @@ -148,7 +134,7 @@ class MasterPlugin extends libPlugin.BaseMasterPlugin { } } } - this.registerProgress(); + this.progressRateLimiter.activate(); let technologies = []; for (let [name, tech] of this.technologies) { diff --git a/test/lib/RateLimiter.js b/test/lib/RateLimiter.js new file mode 100644 index 00000000..d7448d3f --- /dev/null +++ b/test/lib/RateLimiter.js @@ -0,0 +1,64 @@ +"use strict"; +const assert = require("assert").strict; + +const { wait } = require("@clusterio/lib/helpers"); +const RateLimiter = require("@clusterio/lib/RateLimiter"); + + +describe("lib/RateLimiter", function() { + describe("class RateLimiter", function() { + let rateLimiter; + beforeEach(function() { + rateLimiter = new RateLimiter(); + }); + afterEach(function() { + rateLimiter.cancel(); + }); + + it("should rate limit activations", function() { + assert(rateLimiter.activate(), "fist invocation was rate limited"); + assert(!rateLimiter.activate(), "second invocation was not rate limited"); + }); + + it("should invoke action immediately when not rate limited", function() { + let invoked = false; + rateLimiter.action = () => { invoked = true; }; + rateLimiter.activate(); + assert(invoked, "action was not invoked"); + }); + + it("should invoke action at a later point in time if rate limited", async function() { + let invokedCount = 0; + rateLimiter.maxRate = 100; + rateLimiter.action = () => { invokedCount += 1; }; + rateLimiter.activate(); + rateLimiter.activate(); + assert.equal(invokedCount, 1, "expected one invocation"); + await wait(20); + assert.equal(invokedCount, 2, "expected two invocations"); + }); + + it("should invoke action before timeout if maxRate changed", async function() { + let invokedCount = 0; + rateLimiter.action = () => { invokedCount += 1; }; + rateLimiter.activate(); + rateLimiter.activate(); + assert.equal(invokedCount, 1, "expected one invocation"); + await wait(20); + rateLimiter.maxRate = 100; + rateLimiter.activate(); + assert.equal(invokedCount, 2, "expected two invocations"); + }); + + it("should cancel action if canceled before timeout", async function() { + let invokedCount = 0; + rateLimiter.action = () => { invokedCount += 1; }; + rateLimiter.activate(); + rateLimiter.activate(); + rateLimiter.cancel(); + assert.equal(invokedCount, 1, "expected one invocation"); + await wait(20); + assert.equal(invokedCount, 1, "expected one invocation"); + }); + }); +}); From 699e3994dfc25e5157d38a3ce02a6370dc276bd7 Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Wed, 16 Dec 2020 20:30:17 +0100 Subject: [PATCH 3/7] Add missing plugin error printing in slave.js Special case PluginError printing during startup in slave.js in the same way it's done in master.js. This makes it so that the stack of the error thrown by a plugin is printed instead of the stack trace of the place it was wrapped into a PluginError. --- packages/slave/slave.js | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/packages/slave/slave.js b/packages/slave/slave.js index b717228e..350ba6a7 100755 --- a/packages/slave/slave.js +++ b/packages/slave/slave.js @@ -1647,6 +1647,14 @@ I version of clusterio. Expect things to break. I if (err instanceof libErrors.AuthenticationFailed) { logger.fatal(err.message); + } else if (err instanceof libErrors.PluginError) { + logger.fatal(` +${err.pluginName} plugin threw an unexpected error +during startup, please report it to the plugin author. +------------------------------------------------------ +${err.original.stack}` + ); + } else { logger.fatal(` +--------------------------------------------------------------+ From 6eef4bd300d24d2cdd2598e56826099348f0822f Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Wed, 16 Dec 2020 20:33:17 +0100 Subject: [PATCH 4/7] Make metrics gathering timeout configurable When reducing the collection interval in Prometheus it may be necessary to to reduce the metrics gathering timeout to ensure it's less than the interval. Add config options to adjust the timeouts used for metrics gathering so that these can be set accordingly. --- CHANGELOG.md | 1 + docs/configuration.md | 9 +++++++++ packages/lib/config/definitions.js | 7 +++++++ packages/master/master.js | 3 ++- plugins/statistics_exporter/README.md | 17 ++++++++++++++++- plugins/statistics_exporter/info.js | 13 +++++++++++++ plugins/statistics_exporter/instance.js | 7 ++++--- plugins/statistics_exporter/test/plugin.js | 4 +++- test/mock.js | 11 ++++++++--- 9 files changed, 63 insertions(+), 9 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c1308051..57d2cf45 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -53,6 +53,7 @@ Version 2.0.0 - Added option to auto start instances on slave startup. - Added metric mapping of slave and instance ids to their names, allowing them to be displayed by name in queries that join with the mapping. +- Added option to configure timeout for exported Prometheus metrics. ### Changes diff --git a/docs/configuration.md b/docs/configuration.md index 4951fe5d..74409f5c 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -112,6 +112,15 @@ unavoidable. Defaults to 30. +### master.metrics_timeout + +Timeout in seconds before a call to gather metrics from a slave times +out. This should be less than both the timeout and collection interval +configured for the collection job in Prometheus. + +Defaults to 8. + + ### master.default_role_id Role to automatically grant to new users. If null no role is granted. diff --git a/packages/lib/config/definitions.js b/packages/lib/config/definitions.js index 2d342b9b..223f53f1 100644 --- a/packages/lib/config/definitions.js +++ b/packages/lib/config/definitions.js @@ -94,6 +94,13 @@ MasterGroup.define({ type: "number", initial_value: 30, }); +MasterGroup.define({ + name: "metrics_timeout", + title: "Metrics Timeout", + description: "Timeout in seconds for metrics gathering from slaves", + type: "number", + initial_value: 8, +}); MasterGroup.define({ name: "default_role_id", title: "Default role", diff --git a/packages/master/master.js b/packages/master/master.js index c2c2b0f7..14439ada 100755 --- a/packages/master/master.js +++ b/packages/master/master.js @@ -190,8 +190,9 @@ async function getMetrics(req, res, next) { } let requests = []; + let timeout = masterConfig.get("master.metrics_timeout") * 1000; for (let slaveConnection of slaveConnections.values()) { - requests.push(libHelpers.timeout(libLink.messages.getMetrics.send(slaveConnection), 8e3, null)); + requests.push(libHelpers.timeout(libLink.messages.getMetrics.send(slaveConnection), timeout, null)); } for await (let result of await libPrometheus.defaultRegistry.collect()) { diff --git a/plugins/statistics_exporter/README.md b/plugins/statistics_exporter/README.md index b22a9892..7dc08021 100644 --- a/plugins/statistics_exporter/README.md +++ b/plugins/statistics_exporter/README.md @@ -7,7 +7,7 @@ Clusterio. Installation -============ +------------ Run the following commands in the folder Clusterio is installed to: @@ -16,3 +16,18 @@ Run the following commands in the folder Clusterio is installed to: Substitute clusteriomaster with clusterioslave or clusterioctl if this a dedicate slave or ctl installation respectively. + + +Instance Configuration +---------------------- + +### statistics_exporter.command_timeout + +Timeout in seconds to wait for the statitics gathering command to the +Factorio server return the result. If the timeout is exceeded the +previous collected results are returned instead. Normally the command +returns immediatly with the data, but if the RCON interface is +overloaded then the command may take too long to timely answer the +metrics collection request. + +Defaults to `1`. diff --git a/plugins/statistics_exporter/info.js b/plugins/statistics_exporter/info.js index e79bbd89..c89f5c1e 100644 --- a/plugins/statistics_exporter/info.js +++ b/plugins/statistics_exporter/info.js @@ -1,8 +1,21 @@ "use strict"; +const libConfig = require("@clusterio/lib/config"); + +class InstanceConfigGroup extends libConfig.PluginConfigGroup {} +InstanceConfigGroup.groupName = "statistics_exporter"; +InstanceConfigGroup.define({ + name: "command_timeout", + title: "Command Timeout", + description: "Timeout in seconds of the metrics command before returning results from the previous invocation.", + type: "number", + initial_value: 1, +}); +InstanceConfigGroup.finalize(); module.exports = { name: "statistics_exporter", title: "Prometheus Statistics Export", description: "Exports statistics to Prometheus", instanceEntrypoint: "instance", + InstanceConfigGroup, }; diff --git a/plugins/statistics_exporter/instance.js b/plugins/statistics_exporter/instance.js index 001d96b6..0784f816 100644 --- a/plugins/statistics_exporter/instance.js +++ b/plugins/statistics_exporter/instance.js @@ -73,11 +73,12 @@ class InstancePlugin extends libPlugin.BaseInstancePlugin { } async onMetrics() { - // Wait up to 5 seconds for the metrics to be collected. It may + // Wait configured timeout 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 + // stream is overloaded. Should the timeout be exceeded the // previous values for the metrics will end up being sent to master. - await libHelpers.timeout(this.gatherMetrics(), 5e3); + let timeout = this.instance.config.get("statistics_exporter.command_timeout") * 1000; + await libHelpers.timeout(this.gatherMetrics(), timeout); } } diff --git a/plugins/statistics_exporter/test/plugin.js b/plugins/statistics_exporter/test/plugin.js index 52d18595..50a28c88 100644 --- a/plugins/statistics_exporter/test/plugin.js +++ b/plugins/statistics_exporter/test/plugin.js @@ -9,8 +9,10 @@ const info = require("../info"); describe("statistics_exporter plugin", function() { describe("class InstancePlugin", function() { let instancePlugin; + let mockInstance = new mock.MockInstance(); + mockInstance.mockConfigEntries.set("statistics_exporter.command_timeout", 1); before(async function() { - instancePlugin = new instance.InstancePlugin(info, new mock.MockInstance()); + instancePlugin = new instance.InstancePlugin(info, mockInstance); await instancePlugin.init(); }); diff --git a/test/mock.js b/test/mock.js index acdf3c96..89ada241 100644 --- a/test/mock.js +++ b/test/mock.js @@ -103,11 +103,16 @@ class MockInstance extends libLink.Link { this.logger = new MockLogger(); this.server = new MockServer(); this.name = "test"; + this.mockConfigEntries = new Map([ + ["instance.id", 7357], + ["factorio.enable_save_patching", true], + ]); this.config = { get: (name) => { - if (name === "instance.id") { return 7357; } - if (name === "factorio.enable_save_patching") { return true; } - throw Error("Not implemented"); + if (this.mockConfigEntries.has(name)) { + return this.mockConfigEntries.get(name); + } + throw Error(`mock for field ${name} is not implemented`); }, }; } From e8405cbf77d978d52c451cd77830bd68bb2b477c Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Mon, 28 Dec 2020 06:53:59 +0100 Subject: [PATCH 5/7] Fix message validation for request responses Fix the validation loging for sending responses to requests not functioning due to the message data property being passed incorrectly to the validator. --- packages/lib/link/messages.js | 5 +++-- test/lib/link/messages.js | 23 ++++++++++++++++++++++- 2 files changed, 25 insertions(+), 3 deletions(-) diff --git a/packages/lib/link/messages.js b/packages/lib/link/messages.js index 1eb28116..4e558ab6 100644 --- a/packages/lib/link/messages.js +++ b/packages/lib/link/messages.js @@ -207,12 +207,13 @@ class Request extends Message { response = {}; } - if (!this._responseValidator({ seq: 0, type: this.responseType, response })) { + let data = { ...response, seq: message.seq }; + if (!this._responseValidator({ seq: 0, type: this.responseType, data })) { logger.error(JSON.stringify(this._responseValidator.errors, null, 4)); throw new Error(`Validation failed responding to ${this.requestType}`); } - link.connector.send(this.responseType, { ...response, seq: message.seq }); + link.connector.send(this.responseType, data); }).catch(err => { if (!(err instanceof libErrors.RequestError)) { diff --git a/test/lib/link/messages.js b/test/lib/link/messages.js index 62342e94..97ad641e 100644 --- a/test/lib/link/messages.js +++ b/test/lib/link/messages.js @@ -20,6 +20,9 @@ describe("lib/link/messages", function() { requestProperties: { "test": { type: "string" }, }, + responseProperties: { + "test": { type: "string" }, + }, }); describe("constructor", function() { @@ -63,7 +66,24 @@ describe("lib/link/messages", function() { }); }); it("should implicitly send an empty object on empty return", async function() { - handlerResult = undefined; + let emptyRequest = new libLink.Request({ + type: "empty", + links: ["source-target"], + }); + emptyRequest.attach(testTargetLink, async(message) => {}); + testTargetLink._handlers.get("empty_request")({ seq: 2 }); + let result = await new Promise(resolve => { + testTargetLink.connector.send = (type, data) => resolve({ type, data }); + }); + assert.deepEqual(result, { + type: "empty_response", + data: { + seq: 2, + }, + }); + }); + it("should validate response from handler", async function() { + handlerResult = { test: 12 }; testTargetLink._handlers.get("test_request")({ seq: 2 }); let result = await new Promise(resolve => { testTargetLink.connector.send = (type, data) => resolve({ type, data }); @@ -71,6 +91,7 @@ describe("lib/link/messages", function() { assert.deepEqual(result, { type: "test_response", data: { + error: "Validation failed responding to test_request", seq: 2, }, }); From 16525ae07cf8a525ace30531b2ee168931417699 Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Mon, 28 Dec 2020 06:56:47 +0100 Subject: [PATCH 6/7] Add graceful closure of HTTP connections When closing down the HTTP and HTTPS servers existing connections are not closed by Node.js, and may live on forever if they are actively in use. A common case where this happens is if Prometheus is set up with a short interval for collections. To close down the active connections they need to be tracked and closed when the server is closed, add the HttpCloser class which encapsulates the tracking and closure process behind a simple interface and use this to close down the HTTP and HTTPS server on the master. --- packages/master/master.js | 9 +- packages/master/src/HttpCloser.js | 109 ++++++++++++++ test/master/HttpCloser.js | 236 ++++++++++++++++++++++++++++++ 3 files changed, 352 insertions(+), 2 deletions(-) create mode 100644 packages/master/src/HttpCloser.js create mode 100644 test/master/HttpCloser.js diff --git a/packages/master/master.js b/packages/master/master.js index 14439ada..a0b8af05 100755 --- a/packages/master/master.js +++ b/packages/master/master.js @@ -49,6 +49,7 @@ let db = { // homebrew modules const generateSslCert = require("./src/generate_ssl_cert"); +const HttpCloser = require("./src/HttpCloser"); const routes = require("./src/routes"); const libDatabase = require("@clusterio/lib/database"); const libSchema = require("@clusterio/lib/schema"); @@ -72,7 +73,9 @@ const bodyParser = require("body-parser"); const fileUpload = require("express-fileupload"); const app = express(); let httpServer; +let httpServerCloser; let httpsServer; +let httpsServerCloser; app.use(cookieParser()); app.use(bodyParser.json({ @@ -496,8 +499,8 @@ async function shutdown() { let stopTasks = []; logger.info("Stopping HTTP(S) server"); - if (httpServer) { stopTasks.push(new Promise(resolve => httpServer.close(resolve))); } - if (httpsServer) { stopTasks.push(new Promise(resolve => httpsServer.close(resolve))); } + if (httpServer) { stopTasks.push(httpServerCloser.close()); } + if (httpsServer) { stopTasks.push(httpsServerCloser.close()); } await Promise.all(stopTasks); logger.info(`Clusterio cleanly exited in ${Date.now() - exitStartTime}ms`); @@ -1982,6 +1985,7 @@ async function startServer() { // Only start listening for connections after all plugins have loaded if (httpPort) { httpServer = new (require("http").Server)(app); + httpServerCloser = new HttpCloser(httpServer); await listen(httpServer, httpPort, bindAddress); logger.info(`Listening for HTTP on port ${httpServer.address().port}`); } @@ -2002,6 +2006,7 @@ async function startServer() { key: privateKey, cert: certificate, }, app); + httpsServerCloser = new HttpCloser(httpsServer); await listen(httpsServer, httpsPort, bindAddress); logger.info(`Listening for HTTPS on port ${httpsServer.address().port}`); } diff --git a/packages/master/src/HttpCloser.js b/packages/master/src/HttpCloser.js new file mode 100644 index 00000000..8a488404 --- /dev/null +++ b/packages/master/src/HttpCloser.js @@ -0,0 +1,109 @@ +"use strict"; +const http = require("http"); +const https = require("https"); +const util = require("util"); + +/** + * Gracefully close a Node.js HTTP(S) server + * + * Helper class to properly close down a Node.js HTTP(S) server. This is + * necessary due to connections being able to linger around forever with the + * keep-alive logic in HTTP. + * + * @example + * // during server initialization + * const server = http.createServer(); + * const serverCloser = new HttpCloser(server); + * + * // in the shutdown logic + * await serverCloser.close(); + * + * @alias module:master/src/HttpCloser + */ +class HttpCloser { + /** + * Server to attach this closer to. + * @param {https.Server | http.Server} server - HTTP(S) to attach to + */ + constructor(server) { + this._server = server; + this._sockets = new Set(); + this._responses = new Set(); + + if (this._server instanceof http.Server) { + this._server.on("connection", this._addSocket.bind(this)); + } else if (this._server instanceof https.Server) { + this._server.on("secureConnection", this._addSocket.bind(this)); + } else { + throw Error("server must be an instance of either http.Server or https.Server."); + } + this._server.on("request", this._addRequest.bind(this)); + } + + _addSocket(socket) { + this._sockets.add(socket); + socket.once("close", () => { + this._sockets.delete(socket); + }); + } + + _addRequest(request, response) { + this._responses.add(response); + response.once("close", () => this._responses.delete(response)); + response.once("finish", () => this._responses.delete(response)); + } + + /** + * Gracefully close down the attached HTTP(S) server. + * + * Closes open connections that do not have a request in flight, signals + * connections to pending requests should close, and initiates the + * close operation on the attached server. + * + * If there's requests in progress that still haven't been completed + * after the timeout value passed they will be aborted. + * + * @param {number} timeout - + * Time in ms to wait for connections to close. + */ + async close(timeout = 5000) { + let activeSockets = new Set(); + for (let response of this._responses) { + if (!response.headersSent) { + response.setHeader("Connection", "close"); + } else { + let socket = response.socket; + response.once("finish", () => { + socket.end(); + }); + } + activeSockets.add(response.socket); + } + for (let socket of this._sockets) { + if (!activeSockets.has(socket)) { + socket.end(); + } + } + + return new Promise((resolve, reject) => { + let timer = setTimeout(() => { + for (let socket of this._sockets) { + // Ideally a RST would be sent here to indicate the the + // transfer failed, but we can't reliably send that from + // Node.js, see nodejs/node#27428. + socket.destroy(); + } + }, timeout); + this._server.close(err => { + clearTimeout(timer); + if (err) { + reject(err); + } else { + resolve(); + } + }); + }); + } +}; + +module.exports = HttpCloser; diff --git a/test/master/HttpCloser.js b/test/master/HttpCloser.js new file mode 100644 index 00000000..9b59de49 --- /dev/null +++ b/test/master/HttpCloser.js @@ -0,0 +1,236 @@ +"use strict"; +const assert = require("assert").strict; +const http = require("http"); +const https = require("https"); +const util = require("util"); +const tls = require("tls"); + +const forge = require("node-forge"); +const phin = require("phin"); + +const { wait } = require("@clusterio/lib/helpers"); +const HttpCloser = require("@clusterio/master/src/HttpCloser"); + +// Time to wait during async operations to ensure they happened in order. +const tick = 20; + +const origCreateSecureContext = tls.createSecureContext; +tls.createSecureContext = function(options) { + if (!options.cert || !options.key) { + return origCreateSecureContext(options); + } + + let lessOptions = { ...options }; + delete lessOptions.key; + delete lessOptions.cert; + let ctx = origCreateSecureContext(lessOptions); + ctx.context.setCert(options.cert); + ctx.context.setKey(options.key, undefined); + return ctx; +}; + +function serverSuite(proto) { + let api = proto === "http" ? http : https; + let server; + let serverOptions = {}; + let clientOptions = {}; + let closer; + let isClosed; + let addr; + + function setClosed() { + isClosed = true; + } + + async function get(url, options = {}) { + let core = { ...clientOptions, ...(options.core || {}) }; + return await phin({ ...options, url, core }); + } + + before(async function() { + if (proto === "https") { + // Generate TLS certificate + let keypair = forge.pki.rsa.generateKeyPair(512); + let cert = forge.pki.createCertificate(); + cert.publicKey = keypair.publicKey; + cert.serialNumber = "01"; + cert.validity.notBefore = new Date(); + cert.validity.notAfter = new Date(cert.validity.notBefore); + cert.validity.notAfter.setFullYear(cert.validity.notAfter.getFullYear() + 1); + let attrs = [{ name: "commonName", value: "localhost" }]; + cert.setSubject(attrs); + cert.setIssuer(attrs); + cert.sign(keypair.privateKey); + serverOptions.key = forge.pki.privateKeyToPem(keypair.privateKey); + let pem = forge.pki.certificateToPem(cert); + serverOptions.cert = pem; + serverOptions.ciphers = "DEFAULT@SECLEVEL=0"; + clientOptions.ca = pem; + clientOptions.ciphers = "DEFAULT@SECLEVEL=0"; + } + }); + beforeEach(async function() { + isClosed = false; + server = api.createServer(serverOptions); + server.unref(); + server.on("close", setClosed); + closer = new HttpCloser(server); + await util.promisify(server.listen.bind(server))(); + addr = `${proto}://localhost:${server.address().port}/`; + }); + + afterEach(function() { + server.off("close", setClosed); + }); + + it("should close the server", async function() { + await closer.close(); + assert(isClosed, "Server was not closed"); + }); + + it("should reject with error if the close fails", async function() { + await closer.close(); + await assert.rejects(closer.close(), { code: "ERR_SERVER_NOT_RUNNING" }); + }); + + it("should gracefully let pending requests finish", async function() { + server.on("request", (request, response) => { + setTimeout(() => { response.end("ok-1"); }, 3*tick); + }); + + let request = get(addr); + request.catch(() => {}); + await wait(tick); + await closer.close(); + + let response = await request; + assert.equal(response.statusCode, 200, "non-200 response status code"); + assert.equal(response.body.toString(), "ok-1", "response content does not match expected"); + }); + + it("should set Connection header to close for pending requests", async function() { + server.on("request", (request, response) => { + setTimeout(() => { response.end("ok-2"); }, 3*tick); + assert.equal(request.headers.connection, "keep-alive", "Connection not kept alive"); + }); + + let agent = new api.Agent({ keepAlive: true }); + let request = get(addr, { core: { agent }}); + request.catch(() => {}); + await wait(tick); + await closer.close(); + + let response = await request; + assert.equal(response.statusCode, 200, "non-200 response status code"); + assert.equal(response.body.toString(), "ok-2", "response content does not match expected"); + assert.equal(response.headers.connection, "close", "response Connection header not set to close"); + }); + + it("should gracefully close open connections", async function() { + server.on("request", (request, response) => { + response.end("ok-3"); + assert.equal(request.headers.connection, "keep-alive", "Connection not kept alive"); + }); + + let agent = new api.Agent({ keepAlive: true }); + let request = get(addr, { core: { agent }}); + request.catch(() => {}); + await wait(tick); + await closer.close(); + + let response = await request; + assert.equal(response.statusCode, 200, "Request replied with non-200 status code"); + assert.equal(response.body.toString(), "ok-3", "Request content does not match expected"); + }); + + it("should refuse connection if request sent after starting closure", async function() { + server.on("request", (request, response) => { + setTimeout(() => { response.end("ok-4"); }, 3*tick); + }); + + let request = get(addr); + request.catch(() => {}); + await wait(tick); + let close = closer.close(); + + let agent = new api.Agent({ keepAlive: true }); + await assert.rejects(get(addr, { core: { agent }}), { code: "ECONNREFUSED" }); + + let response = await request; + assert.equal(response.statusCode, 200, "non-200 response status code"); + assert.equal(response.body.toString(), "ok-4", "response content does not match expected"); + + await close; + }); + + it("should close connection after in-flight request is over", async function() { + server.on("request", (request, response) => { + response.write("ok"); + setTimeout(() => { response.end("-5"); }, 3*tick); + assert.equal(request.headers.connection, "keep-alive", "Connection not kept alive"); + }); + + let agent = new api.Agent({ keepAlive: true }); + let request = get(addr, { core: { agent }}); + request.catch(() => {}); + await wait(tick); + await closer.close(); + + let response = await request; + assert.equal(response.statusCode, 200, "non-200 response status code"); + assert.equal(response.body.toString(), "ok-5", "response content does not match expected"); + }); + + it("should cause connection reset for qeueud request after in-flight request", async function() { + server.on("request", (request, response) => { + response.write("ok"); + setTimeout(() => { response.end("-6"); }, 3*tick); + assert.equal(request.headers.connection, "keep-alive", "Connection not kept alive"); + }); + + let agent = new api.Agent({ keepAlive: true, maxSockets: 1 }); + let request = get(addr, { core: { agent }}); + request.catch(() => {}); + let droppedRequest = get(addr, { core: { agent }}); + droppedRequest.catch(() => {}); + await wait(tick); + await closer.close(); + + let response = await request; + assert.equal(response.statusCode, 200, "non-200 response status code"); + assert.equal(response.body.toString(), "ok-6", "response content does not match expected"); + await assert.rejects(droppedRequest, { code: "ECONNRESET" }); + }); + + it("should abort requests that take too long to respond", async function() { + server.on("request", (request, response) => { + response.write("ok"); + setTimeout(() => { response.end("-7"); }, 5*tick); + }); + + let request = get(addr); + request.catch(() => {}); + await wait(tick); + await closer.close(tick*2); + + // Due to a bug in phin [https://github.com/ethanent/phin/issues/59] + // the request Promise never settles. + // TODO: uncomment when bug is fixed in phin. + // await assert.rejects(request); + }); +} + +describe("master/src/HttpCloser.js", function() { + describe("class HttpCloser", function() { + describe("constructor()", function() { + it("should throw if not passed a Server instance", function() { + assert.throws(() => new HttpCloser({})); + }); + }); + for (let proto of ["http", "https"]) { + describe(`${proto} server`, function() { + serverSuite(proto); + }); + } + }); +}); From aa3dfdbc0aa5f095f11d58848d80a13742da12a4 Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Tue, 29 Dec 2020 03:29:10 +0100 Subject: [PATCH 7/7] Fix crash on race condition stopping Factorio server If the server decides to exit on its own during the stop operation to the FactorioServer class after the operation has started but before the signal is sent to the process then the exit handler will have wiped the server child process and cause the kill call to fail. This in turn would let the hang timer running which would bring down the Node.js process due to an unhandled error for the same reason. Fix by making sure the stop operation watches for the server exiting and returning early if the server already quit before getting to the send stop signal part. --- packages/lib/factorio/server.js | 29 ++++++++++++++++++++++++++--- test/lib/factorio/server.js | 27 +++++++++++++++++++++++++++ 2 files changed, 53 insertions(+), 3 deletions(-) diff --git a/packages/lib/factorio/server.js b/packages/lib/factorio/server.js index 221429f1..d9de5ca8 100644 --- a/packages/lib/factorio/server.js +++ b/packages/lib/factorio/server.js @@ -522,6 +522,8 @@ class FactorioServer extends events.EventEmitter { * @param {number} options.maxConcurrentCommands - * Maximum number of RCON commands transmitted in parallel on the * RCON connection. + * @param {number} options.hangTimeout - + * Timeout in ms to wait for a response to a shutdown command. */ constructor(factorioDir, writeDir, options) { super(); @@ -547,6 +549,8 @@ class FactorioServer extends events.EventEmitter { this.verboseLogging = options.verboseLogging; /** Maximum number of RCON commands transmitted in parallel on the RCON connection */ this.maxConcurrentCommands = options.maxConcurrentCommands || 5; + /** Timeout in ms to wait for a response to a shutdown command */ + this.hangTimeout = options.hangTimeout || 5000; this._state = "new"; this._server = null; this._rconClient = null; @@ -704,8 +708,6 @@ class FactorioServer extends events.EventEmitter { this._rconClient.on("authenticated", () => { this._rconReady = true; this.emit("rcon-ready"); }); this._rconClient.on("end", () => { this._rconClient = null; - this._rconReady = false; - this._startRcon().catch(() => { /* Ignore */ }); }); // XXX: Workaround to suppress bogus event listener warning if (this._rconClient.emitter instanceof events.EventEmitter) { @@ -802,6 +804,10 @@ class FactorioServer extends events.EventEmitter { } } + if (this._rconClient) { + this._rconClient.end().catch(() => {}); + } + // Reset server state this._state = "init"; this._server = null; @@ -954,6 +960,9 @@ class FactorioServer extends events.EventEmitter { if (!this._rconReady) { await events.once(this, "rcon-ready"); } + if (!this._rconClient) { + throw new Error("RCON connection lost"); + } let response = await this._rconClient.send(message); if (expectEmpty && response !== "") { @@ -972,6 +981,7 @@ class FactorioServer extends events.EventEmitter { this._check(["running"]); let hanged = true; + let stopped = false; function setAlive() { hanged = false; } let timeoutId = setTimeout(() => { @@ -982,7 +992,14 @@ class FactorioServer extends events.EventEmitter { } this._server.kill("SIGKILL"); } - }, 5000); + }, this.hangTimeout); + + // It's possible the server decides to exit on its own, in which + // case the stop operation has to be cancelled. + this._server.once("exit", () => { + stopped = true; + clearTimeout(timeoutId); + }); this._state = "stopping"; // On Windows we need to save the map as there's no graceful shutdown. @@ -1014,6 +1031,12 @@ class FactorioServer extends events.EventEmitter { await this._rconClient.end(); } + // The Factorio server may have decided to get ahead of us and + // stop by itself before we had the chance to signal it. + if (stopped) { + return; + } + // On linux this sends SIGTERM to the process. On windows this // will terminate the process with no cleanup. You can send your // complaint for the lack of graceful shutdown to the Factorio diff --git a/test/lib/factorio/server.js b/test/lib/factorio/server.js index f258d11a..c62186a3 100644 --- a/test/lib/factorio/server.js +++ b/test/lib/factorio/server.js @@ -5,6 +5,7 @@ const fs = require("fs-extra"); const path = require("path"); const libFactorio = require("@clusterio/lib/factorio"); +const { wait } = require("@clusterio/lib/helpers"); const { testLines } = require("./lines"); @@ -243,5 +244,31 @@ describe("lib/factorio/server", function() { assert(!await fs.pathExists(filePath), "File was not deleted"); }); }); + + describe(".stop()", function() { + it("should handle server quitting on its own during stop", async function() { + if (process.platform === "win32") { + this.skip(); + } + server.hangTimeout = 20; + server._server = new events.EventEmitter(); + server._server.kill = () => true; + server._state = "running"; + server._rconReady = true; + server._rconClient = { + async end() { + server._rconClient = null; + process.nextTick(() => { + server.emit("_quitting"); + server._server.emit("exit"); + }); + }, + }; + server._watchExit(); + + await server.stop(); + await wait(21); // Wait until after hang timeout + }); + }); }); });