diff --git a/CHANGELOG.md b/CHANGELOG.md index ca5d4e2f..bfa1c742 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,7 +4,8 @@ ### Features -- Added support for `color-settings` Mod settings (#568)[https://github.com/clusterio/clusterio/issues/568]. +- Added support for `color-settings` Mod settings [#568](https://github.com/clusterio/clusterio/issues/568). +- Added `clusterio_controller_pending_requests` and `clusterio_host_pending_requests` metrics. ### Changes @@ -12,6 +13,8 @@ - Fixed ctl not exporting types used for adding commands to ctl plugins. - Fixed controller only plugins throwing an error about missing a config field on startup out when added to hosts. - Fixed installer incorrectly claiming the options --factorio-dir and --no-download-headless conflicts with each other. +- Fixed a memory leak on every request sent over the WebSocket. +- Fixed assertion error on closing or invalidating a link with messages forwarded from a virtual link. ## Version 2.0.0-alpha.14 diff --git a/packages/controller/controller.ts b/packages/controller/controller.ts index 14b18e46..b4161f65 100644 --- a/packages/controller/controller.ts +++ b/packages/controller/controller.ts @@ -37,6 +37,24 @@ import { version } from "./package.json"; // globals let controller: Controller; +void new lib.Gauge( + "clusterio_controller_pending_requests", + "Count of pending link requests currently waiting in memory on the controller.", + { + callback: (gauge) => { + let count = 0; + if (controller) { + for (const host of controller.wsServer.hostConnections.values()) { + count += host.pendingRequestCount; + } + for (const control of controller.wsServer.controlConnections.values()) { + count += control.pendingRequestCount; + } + } + gauge.set(count); + }, + } +); void new lib.Gauge( "clusterio_controller_host_mapping", diff --git a/packages/host/host.ts b/packages/host/host.ts index 46827d85..fa195675 100755 --- a/packages/host/host.ts +++ b/packages/host/host.ts @@ -26,6 +26,27 @@ import { ConsoleTransport, levels, logger } from "@clusterio/lib"; import Host from "./src/Host"; +let host: Host | undefined; + +void new lib.Gauge( + "clusterio_host_pending_requests", + "Count of pending link requests currently waiting in memory on the host.", + { + labels: ["host_id"], + callback: (gauge) => { + if (!host) { + return; + } + + let count = host.pendingRequestCount; + for (const connection of host.instanceConnections.values()) { + count += connection.pendingRequestCount; + count += connection.instance.pendingRequestCount; + } + gauge.labels(String(host.config.get("host.id"))).set(count); + }, + } +); export class HostConnector extends lib.WebSocketClientConnector { constructor( @@ -205,7 +226,7 @@ async function startHost() { } let hostConnector = new HostConnector(hostConfig, tlsCa, pluginInfos); - let host = new Host( + host = new Host( hostConnector, args.config, hostConfig, @@ -227,7 +248,7 @@ async function startHost() { secondSigint = true; logger.info("Caught interrupt signal, shutting down"); - host.shutdown(); + host!.shutdown(); }); let secondSigterm = false; process.on("SIGTERM", () => { @@ -240,11 +261,11 @@ async function startHost() { secondSigterm = true; logger.info("Caught termination signal, shutting down"); - host.shutdown(); + host!.shutdown(); }); process.on("SIGHUP", () => { logger.info("Terminal closed, shutting down"); - host.shutdown(); + host!.shutdown(); }); try { @@ -255,7 +276,7 @@ async function startHost() { } hostConnector.once("connect", () => { - logger.add(new lib.LinkTransport({ link: host })); + logger.add(new lib.LinkTransport({ link: host! })); }); await hostConnector.connect(); diff --git a/packages/lib/src/link/link.ts b/packages/lib/src/link/link.ts index deabd028..14b7d1ec 100644 --- a/packages/lib/src/link/link.ts +++ b/packages/lib/src/link/link.ts @@ -141,17 +141,30 @@ export class Link { }); } + /** + * Count of requests currently waiting for a response on this link + */ + get pendingRequestCount() { + return this._pendingRequests.size + this._forwardedRequests.size; + } + _clearPendingRequests(err: Error & { code?: string }) { for (let pending of this._pendingRequests.values()) { pending.reject(err); } for (let pending of this._forwardedRequests.values()) { - assert(pending.origin.connector instanceof WebSocketBaseConnector); - if (pending.origin.connector.hasSession) { - pending.origin.connector.sendResponseError( - new libData.ResponseError(err.message, err.code), pending.src, pending.dst - ); + // Drop response if the origin is a WebSocket and the session + // has expired. The other end of the link will have sent an + // error response to the request in this case. + if ( + pending.origin.connector instanceof WebSocketBaseConnector + && !pending.origin.connector.hasSession + ) { + continue; } + pending.origin.connector.sendResponseError( + new libData.ResponseError(err.message, err.code), pending.src, pending.dst + ); } this._pendingRequests.clear(); this._forwardedRequests.clear(); @@ -421,6 +434,7 @@ export class Link { throw new libErrors.InvalidMessage(`Received reply from ${message.src} for message sent to ${pending.dst}`); } + this._pendingRequests.delete(message.dst.requestId!); try { pending.resolve(pending.request.responseFromJSON(message.data)); } catch (err: any) { @@ -437,6 +451,7 @@ export class Link { ); } + this._pendingRequests.delete(message.dst.requestId!); pending.reject(new libErrors.RequestError(message.data.message, message.data.code, message.data.stack)); } diff --git a/test/lib/link/link.js b/test/lib/link/link.js index 0c991031..c8bd06c4 100644 --- a/test/lib/link/link.js +++ b/test/lib/link/link.js @@ -16,6 +16,7 @@ describe("lib/link/link", function() { describe("class Link", function() { let testConnector; + /** @type {lib.Link} */ let testLink; let src = addr({ controlId: 1 }); let dst = addr("controller"); @@ -281,6 +282,7 @@ describe("lib/link/link", function() { let srcReq = new lib.Address(lib.Address.control, 1, 1); testConnector.emit("message", new lib.MessageResponse(1, dst, srcReq)); await request; + assert.deepEqual(testLink._pendingRequests, new Map()); }); it("should send request and reject with error when error response is received", async function() { let message = events.once(testConnector, "send"); @@ -297,6 +299,7 @@ describe("lib/link/link", function() { request, { message: "Error" } ); + assert.deepEqual(testLink._pendingRequests, new Map()); }); it("should send request and resolve with data when response is received", async function() { let message = events.once(testConnector, "send"); @@ -305,6 +308,7 @@ describe("lib/link/link", function() { let srcReq = new lib.Address(lib.Address.control, 1, 1); testConnector.emit("message", new lib.MessageResponse(1, dst, srcReq, 44)); assert.deepEqual(await request, new NumberRequest.Response(44)); + assert.deepEqual(testLink._pendingRequests, new Map()); }); it("should send event to the other side of the link", async function() { let message = events.once(testConnector, "send");