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/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/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/lib/factorio/server.js b/packages/lib/factorio/server.js index 69bdc04d..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; @@ -637,7 +641,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 }); } } @@ -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,17 +981,25 @@ class FactorioServer extends events.EventEmitter { this._check(["running"]); let hanged = true; + let stopped = false; function setAlive() { hanged = false; } 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(() => {}); } 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/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/packages/master/master.js b/packages/master/master.js index c2c2b0f7..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({ @@ -190,8 +193,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()) { @@ -495,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`); @@ -1981,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}`); } @@ -2001,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/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(` +--------------------------------------------------------------+ 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/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/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"); + }); + }); +}); 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 + }); + }); }); }); 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, }, }); 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); + }); + } + }); +}); 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`); }, }; }