From d7edae99c5932b8f14c5fc210810efdc42138168 Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Mon, 22 Jan 2024 19:29:36 +0100 Subject: [PATCH 1/3] Add metrics for pending request Track the number of requests currently awaiting a response from a request handler both from interally issued requests and requests forwarded from other parties. --- CHANGELOG.md | 3 ++- packages/controller/controller.ts | 18 ++++++++++++++++++ packages/host/host.ts | 31 ++++++++++++++++++++++++++----- packages/lib/src/link/link.ts | 7 +++++++ 4 files changed, 53 insertions(+), 6 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ca5d4e2f..4dd4be7c 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 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..c7365190 100644 --- a/packages/lib/src/link/link.ts +++ b/packages/lib/src/link/link.ts @@ -141,6 +141,13 @@ 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); From 123148e2777ad90dca275676e58438f84497d62e Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Mon, 22 Jan 2024 19:54:41 +0100 Subject: [PATCH 2/3] Fix pending request not deleted on response Fix memory leak due to the recorded _pendingRequests entry for a request not being deleted when the response for the request has been received. --- CHANGELOG.md | 1 + packages/lib/src/link/link.ts | 2 ++ test/lib/link/link.js | 4 ++++ 3 files changed, 7 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4dd4be7c..ca5d9a34 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ - 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. ## Version 2.0.0-alpha.14 diff --git a/packages/lib/src/link/link.ts b/packages/lib/src/link/link.ts index c7365190..4793008b 100644 --- a/packages/lib/src/link/link.ts +++ b/packages/lib/src/link/link.ts @@ -428,6 +428,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) { @@ -444,6 +445,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"); From 96bd5fb1c2879f50bfa0371b3c196f68a1266b27 Mon Sep 17 00:00:00 2001 From: Hornwitser Date: Mon, 22 Jan 2024 20:33:46 +0100 Subject: [PATCH 3/3] Handle virtual links in Link._clearPendingRequests When a request is forwarded from a virtual link to a WebSocket link as is the case when an instance sends a request to the controller the origin for the request is a virtual link. Handle this case in _clearPendingRequests instead of throwing an assertion error. --- CHANGELOG.md | 1 + packages/lib/src/link/link.ts | 16 +++++++++++----- 2 files changed, 12 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index ca5d9a34..bfa1c742 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -14,6 +14,7 @@ - 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/lib/src/link/link.ts b/packages/lib/src/link/link.ts index 4793008b..14b7d1ec 100644 --- a/packages/lib/src/link/link.ts +++ b/packages/lib/src/link/link.ts @@ -153,12 +153,18 @@ export class Link { 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();