Merge branch 'dev/monitoring'

nix/tests/test-bundle-starts.nix

@@ -73,6 +73,8 @@ in
     };
   };
 
+  # https://nixos.org/manual/nixos/stable/index.html#sec-nixos-tests
+  # https://nixos.org/manual/nixpkgs/unstable/#tester-runNixOSTest
   testScript = ''
     machine.wait_for_unit("spacebar-api")
     machine.wait_for_unit("spacebar-cdn")
@@ -82,7 +84,13 @@ in
     machine.wait_for_open_port(3001)
     machine.wait_for_open_port(3002)
     machine.wait_for_open_port(3003)
-    # If well known works, its probably fine(tm)?
+
+    # this should be working
     machine.succeed("curl -f http://api.sb.localhost/.well-known/spacebar/client")
+
+    # check if metrics endpoint works on all services
+    machine.succeed("curl -f http://api.sb.localhost/metrics")
+    machine.succeed("curl -f http://gateway.sb.localhost/metrics")
+    machine.succeed("curl -f http://cdn.sb.localhost/metrics")
   '';
 }

package.json

@@ -115,6 +115,7 @@
         "pg-query-stream": "^4.15.0",
         "picocolors": "^1.1.1",
         "probe-image-size": "^7.3.0",
+        "prom-client": "^15.1.3",
         "reflect-metadata": "^0.2.2",
         "tslib": "^2.8.1",
         "typeorm": "^0.3.30",

src/api/Server.ts

@@ -26,6 +26,7 @@ import { Authentication, CORS, ImageProxy, BodyParser, ErrorHandler, initRateLim
 import { initInstance } from "./util/handlers/Instance";
 import { route } from "./util";
 import { ProcessLifecycle } from "../util/util/ProcessLifecycle";
+import { Monitoring } from "../util/monitoring/Monitoring";
 
 const ASSETS_FOLDER = path.join(__dirname, "..", "..", "assets");
 const PUBLIC_ASSETS_FOLDER = path.join(ASSETS_FOLDER, "public");
@@ -51,6 +52,8 @@ export class SpacebarServer extends Server {
     }
 
     async start() {
+        await Monitoring.init();
+        Monitoring.attach(this.app);
         await initDatabase();
         await Config.init();
         await initEvent();

src/bundle/Server.ts

@@ -28,6 +28,7 @@ import * as Webrtc from "@spacebar/webrtc";
 import { CDNServer } from "@spacebar/cdn";
 import { Config, initDatabase } from "@spacebar/util";
 import { ProcessLifecycle } from "../util/util/ProcessLifecycle";
+import { Monitoring } from "../util/monitoring/Monitoring";
 
 const app = express();
 const server = http.createServer();
@@ -54,6 +55,7 @@ ProcessLifecycle.eventEmitter.on("stopping", async () => {
 });
 
 async function main() {
+    await Monitoring.init();
     await initDatabase();
     await Config.init();
 

src/cdn/Server.ts

@@ -24,6 +24,7 @@ import { CORS, BodyParser } from "@spacebar/api";
 import guildProfilesRoute from "./routes/guild-profiles";
 import { storage } from "./util";
 import { ProcessLifecycle } from "../util/util/ProcessLifecycle";
+import { Monitoring } from "../util/monitoring/Monitoring";
 
 export type CDNServerOptions = ServerOptions;
 
@@ -35,6 +36,8 @@ export class CDNServer extends Server {
     }
 
     async start() {
+        await Monitoring.init();
+        Monitoring.attach(this.app);
         await initDatabase();
         await Config.init();
 

src/gateway/Server.ts

@@ -22,8 +22,9 @@ import ws from "ws";
 import { checkToken, Config, initDatabase, initEvent, Rights } from "@spacebar/util";
 import { randomString } from "@spacebar/api"; // TODO: move to util
 import { Connection, openConnections } from "./events/Connection";
-import { cleanupOnStartup, OPCODES, Send } from "./util";
+import { cleanupOnStartup } from "./util";
 import { ProcessLifecycle } from "../util/util/ProcessLifecycle";
+import { Monitoring } from "../util/monitoring/Monitoring";
 
 export class Server {
     public ws: ws.Server;
@@ -67,7 +68,9 @@ export class Server {
                     res.setHeader("Set-Cookie", `__sb_sessid=${randomString(32)}; Secure; HttpOnly; SameSite=None; Path=/`);
                 }
                 const requestUrl = new URL(`http://${req.headers.host}${req.url}`);
-                if (requestUrl.pathname === "/_spacebar/gateway/admin/introspect") {
+                if (requestUrl.pathname === "/metrics") {
+                    return await Monitoring.handleRawRequest(req, res);
+                } else if (requestUrl.pathname === "/_spacebar/gateway/admin/introspect") {
                     if (!req.headers.authorization) {
                         return res.writeHead(401).end("Unauthorized");
                     } else {
@@ -169,6 +172,7 @@ export class Server {
     }
 
     async start(): Promise<void> {
+        await Monitoring.init();
         await initDatabase();
         await Config.init();
         await initEvent();

src/gateway/events/Connection.ts

@@ -30,6 +30,8 @@ import { URL } from "node:url";
 import { Config } from "@spacebar/util";
 import { Decoder, Encoder } from "@toondepauw/node-zstd";
 import { ProcessLifecycle } from "../../util/util/ProcessLifecycle";
+import { Monitoring } from "../../util/monitoring/Monitoring";
+import { Gauge } from "prom-client";
 
 // TODO: check rate limit
 // TODO: specify rate limit in config
@@ -37,11 +39,21 @@ import { ProcessLifecycle } from "../../util/util/ProcessLifecycle";
 
 export const openConnections: WebSocket[] = [];
 
+const openConnectionCount = Monitoring.attachMetric(
+    "spacebar_gateway_open_connection_count",
+    new Gauge({
+        name: "spacebar_gateway_open_connection_count",
+        help: "The total number of HTTP requests received",
+    }),
+);
+
 export async function Connection(this: WS.Server, socket: WebSocket, request: IncomingMessage) {
     openConnections.push(socket);
+    openConnectionCount.set(openConnections.length);
     socket.on("close", () => {
         const index = openConnections.indexOf(socket);
         if (index !== -1) openConnections.splice(index, 1);
+        openConnectionCount.set(openConnections.length);
     });
 
     const onShutdown = async () => {

src/util/monitoring/Monitoring.ts

@@ -0,0 +1,85 @@
+/*
+	Spacebar: A FOSS re-implementation and extension of the Discord.com backend.
+	Copyright (C) 2026 Spacebar and Spacebar Contributors
+	
+	This program is free software: you can redistribute it and/or modify
+	it under the terms of the GNU Affero General Public License as published
+	by the Free Software Foundation, either version 3 of the License, or
+	(at your option) any later version.
+	
+	This program is distributed in the hope that it will be useful,
+	but WITHOUT ANY WARRANTY; without even the implied warranty of
+	MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+	GNU Affero General Public License for more details.
+	
+	You should have received a copy of the GNU Affero General Public License
+	along with this program.  If not, see <https://www.gnu.org/licenses/>.
+*/
+
+import { IncomingMessage, ServerResponse } from "node:http";
+import * as client from "prom-client";
+import { Application, Router } from "express";
+import { Metric } from "prom-client";
+
+export class Monitoring {
+    static isInitialised = false;
+    public static async init() {
+        if (Monitoring.isInitialised) return;
+        console.log("[Monitoring] Initialising prometheus metrics");
+        client.collectDefaultMetrics({ prefix: "spacebar_" });
+        Monitoring.isInitialised = true;
+    }
+
+    public static attachMetric<T extends Metric>(name: string, metric: T): T {
+        const existingMetric = client.register.getSingleMetric(name);
+        // TODO: is there any way to *ensure* the metric is T? We're assuming that there's no conflicting definitions across the app...
+        if (existingMetric) return existingMetric as T;
+        client.register.registerMetric(metric);
+        return metric;
+    }
+
+    public static attach(app: Application) {
+        const http_request_total = new client.Counter({
+            name: "spacebar_http_request_total",
+            help: "The total number of HTTP requests received",
+            labelNames: ["path", "method", "status_code"],
+        });
+        client.register.registerMetric(http_request_total);
+
+        const http_response_rate_histogram = new client.Histogram({
+            name: "spacebar_http_duration",
+            labelNames: ["path", "method", "status_code"],
+            help: "The duration of HTTP requests in seconds",
+            buckets: [0.0, 0.05, 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9, 1.0, 1.1, 1.2, 1.3, 1.4, 1.5, 1.6, 1.7, 1.8, 1.9, 2.0, 2.5, 3.0, 3.5, 4.0, 4.5, 5.0, 10],
+        });
+        client.register.registerMetric(http_response_rate_histogram);
+
+        app.use((req, res, next) => {
+            const endTimer = http_response_rate_histogram.startTimer();
+            res.on("finish", () => {
+                const path = (res.locals.lambertRouteBase ?? req.baseUrl ?? "") + req.route?.path;
+                if (!req.route?.path && req.method !== "OPTIONS") {
+                    console.log("[Monitoring] Request route path was undefined? Request path:", req.path, "Request route:", req.route);
+                }
+                endTimer({ method: req.method, path, status_code: res.statusCode });
+
+                // OPTIONS requests don't set path due to not being routed... discard unhandled ones
+                if (!path && req.method === "OPTIONS") return;
+
+                http_request_total.inc({ method: req.method, path, status_code: res.statusCode });
+            });
+            next();
+        });
+
+        app.get("/metrics", async (req, res) => {
+            res.setHeader("Content-Type", client.register.contentType);
+            const metrics = await client.register.metrics();
+            res.send(metrics);
+        });
+    }
+
+    static async handleRawRequest(req: IncomingMessage, res: ServerResponse) {
+        const metrics = await client.register.metrics();
+        res.setHeader("Content-Type", client.register.contentType).writeHead(200).end(metrics);
+    }
+}

src/util/util/extensions/Array.ts

@@ -74,3 +74,9 @@ export function arrayDistributeSequentially<T>(array: T[], count: number): T[][]
 
     return groups;
 }
+
+//region Numerics
+export function arraySum(array: number[]) {
+    return array.reduce((prev, curr) => prev + curr, 0);
+}
+//endregion

src/util/util/ipc/Event.ts

@@ -125,7 +125,7 @@ export async function listenEvent(event: string, callback: (event: EventOpts) =>
         }
 
         if (!listener) {
-            listener = listener = new UnixSocketListener(path.join(process.env.EVENT_SOCKET_PATH, `${process.pid}.sock`));
+            listener = new UnixSocketListener(path.join(process.env.EVENT_SOCKET_PATH, `${process.pid}.sock`));
             await listener.init();
         }
         return await listener.listen(event, callback);

src/util/util/ipc/listener/RabbitMqSingleListener.ts

@@ -19,20 +19,34 @@
 import EventEmitter from "node:events";
 import { randomUUID } from "node:crypto";
 import { BaseEventListener } from "./BaseEventListener";
-import { EVENT, Event, EventOpts, sleep } from "@spacebar/util";
+import { arraySum, EVENT, Event, EventOpts, sleep } from "@spacebar/util";
 import amqp, { Channel, ChannelModel } from "amqplib";
 import { ProcessLifecycle } from "../../ProcessLifecycle";
+import { Monitoring } from "../../../monitoring/Monitoring";
+import { Gauge } from "prom-client";
 
 export class RabbitMqSingleListener extends BaseEventListener {
+    static openListenersMetric: Gauge;
     private readonly host: string;
     private connection?: ChannelModel;
     private channel?: Channel;
     eventEmitter: EventEmitter;
+    openListenersMetric: Gauge.Internal<string>;
 
     constructor(host: string) {
         super();
         this.eventEmitter = new EventEmitter();
         this.host = host;
+
+        RabbitMqSingleListener.openListenersMetric = Monitoring.attachMetric(
+            "spacebar_ipc_unix_listener_open_listener_count",
+            new Gauge({
+                name: "spacebar_ipc_rabbitmqsingle_listener_open_listener_count",
+                help: "Amount of open listeners on unix socket",
+                labelNames: ["host"],
+            }),
+        );
+        this.openListenersMetric = RabbitMqSingleListener.openListenersMetric.labels({ host });
     }
 
     async init() {
@@ -93,6 +107,7 @@ export class RabbitMqSingleListener extends BaseEventListener {
         this.channel = undefined;
         await this.connection?.close();
         this.connection = undefined;
+        RabbitMqSingleListener.openListenersMetric.remove({ host: this.host });
     }
 
     async listen(event: string, callback: (event: EventOpts) => unknown): Promise<() => Promise<void>> {
@@ -104,10 +119,12 @@ export class RabbitMqSingleListener extends BaseEventListener {
         };
 
         this.eventEmitter.addListener(event, listener);
+        this.openListenersMetric.set(arraySum(this.eventEmitter.eventNames().map((e) => this.eventEmitter.listeners(e).length)));
 
         const cancel = async () => {
             this.eventEmitter.removeListener(event, listener);
             this.eventEmitter.setMaxListeners(this.eventEmitter.getMaxListeners() - 1);
+            this.openListenersMetric.set(arraySum(this.eventEmitter.eventNames().map((e) => this.eventEmitter.listeners(e).length)));
         };
 
         this.eventEmitter.setMaxListeners(this.eventEmitter.getMaxListeners() + 1);

src/util/util/ipc/listener/UnixSocketListener.ts

@@ -20,19 +20,46 @@ import EventEmitter from "node:events";
 import fs from "node:fs";
 import net, { Server } from "node:net";
 import { BaseEventListener } from "./BaseEventListener";
-import { EVENT, Event, EventOpts } from "@spacebar/util";
+import { arraySum, EVENT, Event, EventOpts } from "@spacebar/util";
 import { ProcessLifecycle } from "../../ProcessLifecycle";
+import { Gauge } from "prom-client";
+import { Monitoring } from "../../../monitoring/Monitoring";
 
 export class UnixSocketListener extends BaseEventListener {
+    static openConnectionsMetric?: Gauge;
+    static openListenersMetric?: Gauge;
+
     eventEmitter: EventEmitter;
     socketPath: string;
     server: Server;
     isInitialized = false;
+    openConnectionsMetric: Gauge.Internal<string>;
+    openListenersMetric: Gauge.Internal<string>;
 
     constructor(socketPath: string) {
         super();
         this.eventEmitter = new EventEmitter();
         this.socketPath = socketPath;
+
+        UnixSocketListener.openConnectionsMetric = Monitoring.attachMetric(
+            "spacebar_ipc_unix_listener_open_connection_count",
+            new Gauge({
+                name: "spacebar_ipc_unix_listener_open_connection_count",
+                help: "Amount of open inbound connections on unix socket",
+                labelNames: ["path"],
+            }),
+        );
+        this.openConnectionsMetric = UnixSocketListener.openConnectionsMetric.labels({ path: socketPath });
+
+        UnixSocketListener.openListenersMetric = Monitoring.attachMetric(
+            "spacebar_ipc_unix_listener_open_listener_count",
+            new Gauge({
+                name: "spacebar_ipc_unix_listener_open_listener_count",
+                help: "Amount of open listeners on unix socket",
+                labelNames: ["path"],
+            }),
+        );
+        this.openListenersMetric = UnixSocketListener.openListenersMetric.labels({ path: socketPath });
     }
 
     async init() {
@@ -50,6 +77,7 @@ export class UnixSocketListener extends BaseEventListener {
         this.server = net.createServer((socket) => {
             socket.on("connect", () => {
                 console.log("[UnixSocketListener] Unix socket client connected, now at", this.server.connections, "connections...");
+                this.openConnectionsMetric.set(this.server.connections);
             });
             let buffer = Buffer.alloc(0);
             socket.on("data", (data: Buffer) => {
@@ -72,6 +100,7 @@ export class UnixSocketListener extends BaseEventListener {
             });
             socket.on("close", () => {
                 console.log("[UnixSocketListener] Unix socket client disconnected");
+                this.openConnectionsMetric.set(this.server.connections ?? 0);
             });
         });
 
@@ -90,6 +119,7 @@ export class UnixSocketListener extends BaseEventListener {
 
         console.log("[UnixSocketListener] Closing unix socket server");
         this.server.close();
+        UnixSocketListener.openConnectionsMetric?.remove({ path: this.socketPath });
 
         // clean up socket file
         try {
@@ -109,10 +139,12 @@ export class UnixSocketListener extends BaseEventListener {
         };
 
         this.eventEmitter.addListener(event, listener);
+        this.openListenersMetric.set(arraySum(this.eventEmitter.eventNames().map((e) => this.eventEmitter.listeners(e).length)));
 
         const cancel = async () => {
             this.eventEmitter.removeListener(event, listener);
             this.eventEmitter.setMaxListeners(this.eventEmitter.getMaxListeners() - 1);
+            this.openListenersMetric.set(arraySum(this.eventEmitter.eventNames().map((e) => this.eventEmitter.listeners(e).length)));
         };
 
         this.eventEmitter.setMaxListeners(this.eventEmitter.getMaxListeners() + 1);

src/util/util/ipc/writer/UnixSocketWriter.ts

@@ -23,8 +23,12 @@ import { red } from "picocolors";
 import { BaseEventWriter } from "./BaseEventWriter";
 import { Event, Stopwatch } from "@spacebar/util";
 import { ProcessLifecycle } from "../../ProcessLifecycle";
+import { Monitoring } from "../../../monitoring/Monitoring";
+import { Gauge } from "prom-client";
 
 export class UnixSocketWriter extends BaseEventWriter {
+    private static openConnectionsMetric: Gauge;
+
     socketPath: string;
     clients: { [key: string]: Socket } = {};
     watcher?: FSWatcher;
@@ -32,10 +36,21 @@ export class UnixSocketWriter extends BaseEventWriter {
     broadcastLock: Promise<void> = Promise.resolve();
     replayLock: Promise<void> = Promise.resolve();
     isInitializing = true;
+    openConnectionsMetric: Gauge.Internal<string>;
 
     constructor(socketPath: string) {
         super();
         this.socketPath = socketPath;
+
+        UnixSocketWriter.openConnectionsMetric = Monitoring.attachMetric(
+            "spacebar_ipc_unix_writer_open_connection_count",
+            new Gauge({
+                name: "spacebar_ipc_unix_writer_open_connection_count",
+                help: "Amount of open outbound connections on unix socket",
+                labelNames: ["path"],
+            }),
+        );
+        this.openConnectionsMetric = UnixSocketWriter.openConnectionsMetric.labels({ path: socketPath });
     }
 
     async init() {
@@ -80,6 +95,7 @@ export class UnixSocketWriter extends BaseEventWriter {
             try {
                 this.clients[fullPath] = net.createConnection(fullPath, () => {
                     console.log("[UnixSocketWriter] Unix socket client connected to", fullPath);
+                    this.openConnectionsMetric.set(Object.entries(this.clients).length);
                 });
 
                 this.clients[fullPath].on("error", (err) => {
@@ -94,6 +110,7 @@ export class UnixSocketWriter extends BaseEventWriter {
                 this.clients[fullPath].on("close", () => {
                     console.log("[UnixSocketWriter] Unix socket client closed:", fullPath);
                     delete this.clients[fullPath];
+                    this.openConnectionsMetric.set(Object.entries(this.clients).length);
                 });
             } catch (e) {
                 console.error("[UnixSocketWriter] Failed to create connection to", fullPath, ":", e);
@@ -224,6 +241,7 @@ export class UnixSocketWriter extends BaseEventWriter {
             }
         }
         this.clients = {};
+        UnixSocketWriter.openConnectionsMetric.remove({ path: this.socketPath });
     }
 }
 

src/util/util/lambert-server/Server.ts

@@ -54,7 +54,15 @@ export class Server {
             if (router.default) router = router.default;
             if (!router || router?.prototype?.constructor?.name !== "router") throw `File doesn't export any default router`;
 
-            this.app.use(path, <Router>router);
+            this.app.use(
+                path,
+                // TODO: I wish this middleware wasn't nessecary to preserve base path param names for monitoring...
+                (_, res, next) => {
+                    res.locals.lambertRouteBase = path;
+                    next();
+                },
+                <Router>router,
+            );
 
             if (this.options.serverInitLogging && process.env.LOG_ROUTES !== "false") console.log(`[Server] Route ${path} registered`);
 

src/webrtc/Server.ts

@@ -23,6 +23,7 @@ import { Config, initDatabase, initEvent } from "@spacebar/util";
 import { Connection } from "./events/Connection";
 import { loadWebRtcLibrary, mediaServer, WRTC_PORT_MAX, WRTC_PORT_MIN, WRTC_PUBLIC_IP } from "./util";
 import { ProcessLifecycle } from "../util/util/ProcessLifecycle";
+import { Monitoring } from "../util/monitoring/Monitoring";
 
 export class Server {
     public ws: ws.Server;
@@ -36,8 +37,11 @@ export class Server {
 
         if (server) this.server = server;
         else {
-            this.server = http.createServer(function (req, res) {
-                res.writeHead(200).end("Online");
+            this.server = http.createServer(async (req, res) => {
+                const requestUrl = new URL(`http://${req.headers.host}${req.url}`);
+                if (requestUrl.pathname === "/metrics") {
+                    return await Monitoring.handleRawRequest(req, res);
+                } else res.writeHead(200).end("Online");
             });
         }
 
@@ -59,6 +63,7 @@ export class Server {
     }
 
     async start(): Promise<void> {
+        await Monitoring.init();
         await initDatabase();
         await Config.init();
         await initEvent();