diff --git a/nix/tests/test-bundle-starts.nix b/nix/tests/test-bundle-starts.nix index 2e088f6..852b171 100644 --- a/nix/tests/test-bundle-starts.nix +++ b/nix/tests/test-bundle-starts.nix @@ -73,6 +73,8 @@ }; }; + # 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 @@ 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") ''; } diff --git a/package-lock.json b/package-lock.json index 2b96765..659cd6a 100644 --- a/package-lock.json +++ b/package-lock.json @@ -47,6 +47,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", @@ -954,6 +955,15 @@ "url": "https://paulmillr.com/funding/" } }, + "node_modules/@opentelemetry/api": { + "version": "1.9.1", + "resolved": "https://registry.npmjs.org/@opentelemetry/api/-/api-1.9.1.tgz", + "integrity": "sha512-gLyJlPHPZYdAk1JENA9LeHejZe1Ti77/pTeFm/nMXmQH/HFZlcS/O2XJB+L8fkbrNSqhdtlvjBVjxwUYanNH5Q==", + "license": "Apache-2.0", + "engines": { + "node": ">=8.0.0" + } + }, "node_modules/@peculiar/asn1-schema": { "version": "2.7.0", "resolved": "https://registry.npmjs.org/@peculiar/asn1-schema/-/asn1-schema-2.7.0.tgz", @@ -2325,6 +2335,12 @@ "integrity": "sha512-nbE1WxOTTrUWIfsfZ4aHGYu5DOuNkbxGokjV6Z2kxfJK3uaAb8zNK1muzOeipoLHZjInT4Br88BHpzevc681xA==", "license": "CC0-1.0" }, + "node_modules/bintrees": { + "version": "1.0.2", + "resolved": "https://registry.npmjs.org/bintrees/-/bintrees-1.0.2.tgz", + "integrity": "sha512-VOMgTMwjAaUG580SXn3LacVgjurrbMme7ZZNYGSSV7mmtY6QQRh0Eg3pwIcntQ77DErK1L0NxkbetjcoXzVwKw==", + "license": "MIT" + }, "node_modules/bmp-ts": { "version": "1.0.9", "resolved": "https://registry.npmjs.org/bmp-ts/-/bmp-ts-1.0.9.tgz", @@ -5876,6 +5892,19 @@ "stream-parser": "~0.3.1" } }, + "node_modules/prom-client": { + "version": "15.1.3", + "resolved": "https://registry.npmjs.org/prom-client/-/prom-client-15.1.3.tgz", + "integrity": "sha512-6ZiOBfCywsD4k1BN9IX0uZhF+tJkV8q8llP64G5Hajs4JOeVLPCwpPVcpXy3BwYiUGgyJzsJJQeOIv7+hDSq8g==", + "license": "Apache-2.0", + "dependencies": { + "@opentelemetry/api": "^1.4.0", + "tdigest": "^0.1.1" + }, + "engines": { + "node": "^16 || ^18 || >=20" + } + }, "node_modules/proxy-addr": { "version": "2.0.7", "resolved": "https://registry.npmjs.org/proxy-addr/-/proxy-addr-2.0.7.tgz", @@ -6551,6 +6580,15 @@ "node": ">=8" } }, + "node_modules/tdigest": { + "version": "0.1.2", + "resolved": "https://registry.npmjs.org/tdigest/-/tdigest-0.1.2.tgz", + "integrity": "sha512-+G0LLgjjo9BZX2MfdvPfH+MKLCrxlXSYec5DaPYP1fe6Iyhf0/fSmJ0bFiZ1F8BT6cGXl2LpltQptzjXKWEkKA==", + "license": "MIT", + "dependencies": { + "bintrees": "1.0.2" + } + }, "node_modules/thirty-two": { "version": "1.0.2", "resolved": "https://registry.npmjs.org/thirty-two/-/thirty-two-1.0.2.tgz", diff --git a/package.json b/package.json index dc3fe06..f20e31d 100644 --- a/package.json +++ b/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", diff --git a/src/api/Server.ts b/src/api/Server.ts index a1eb171..f9f1c70 100644 --- a/src/api/Server.ts +++ b/src/api/Server.ts @@ -26,6 +26,7 @@ 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 @@ } async start() { + await Monitoring.init(); + Monitoring.attach(this.app); await initDatabase(); await Config.init(); await initEvent(); diff --git a/src/bundle/Server.ts b/src/bundle/Server.ts index f743497..c396d10 100644 --- a/src/bundle/Server.ts +++ b/src/bundle/Server.ts @@ -28,6 +28,7 @@ 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 @@ }); async function main() { + await Monitoring.init(); await initDatabase(); await Config.init(); diff --git a/src/cdn/Server.ts b/src/cdn/Server.ts index 5d87a86..8964161 100644 --- a/src/cdn/Server.ts +++ b/src/cdn/Server.ts @@ -24,6 +24,7 @@ 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 @@ } async start() { + await Monitoring.init(); + Monitoring.attach(this.app); await initDatabase(); await Config.init(); diff --git a/src/gateway/Server.ts b/src/gateway/Server.ts index 5ac94a8..4304197 100644 --- a/src/gateway/Server.ts +++ b/src/gateway/Server.ts @@ -22,8 +22,9 @@ 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 @@ 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 @@ } async start(): Promise { + await Monitoring.init(); await initDatabase(); await Config.init(); await initEvent(); diff --git a/src/gateway/events/Connection.ts b/src/gateway/events/Connection.ts index d795d23..8daa599 100644 --- a/src/gateway/events/Connection.ts +++ b/src/gateway/events/Connection.ts @@ -30,6 +30,8 @@ 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 @@ 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 () => { diff --git a/src/util/monitoring/Monitoring.ts b/src/util/monitoring/Monitoring.ts new file mode 100644 index 0000000..2479267 --- /dev/null +++ b/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 . +*/ + +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(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); + } +} diff --git a/src/util/util/extensions/Array.ts b/src/util/util/extensions/Array.ts index e3e6478..c95328b 100644 --- a/src/util/util/extensions/Array.ts +++ b/src/util/util/extensions/Array.ts @@ -74,3 +74,9 @@ return groups; } + +//region Numerics +export function arraySum(array: number[]) { + return array.reduce((prev, curr) => prev + curr, 0); +} +//endregion diff --git a/src/util/util/ipc/Event.ts b/src/util/util/ipc/Event.ts index e836c26..af7c832 100644 --- a/src/util/util/ipc/Event.ts +++ b/src/util/util/ipc/Event.ts @@ -125,7 +125,7 @@ } 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); diff --git a/src/util/util/ipc/listener/RabbitMqSingleListener.ts b/src/util/util/ipc/listener/RabbitMqSingleListener.ts index cd89cac..7e5867a 100644 --- a/src/util/util/ipc/listener/RabbitMqSingleListener.ts +++ b/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; 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 @@ 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> { @@ -104,10 +119,12 @@ }; 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); diff --git a/src/util/util/ipc/listener/UnixSocketListener.ts b/src/util/util/ipc/listener/UnixSocketListener.ts index be1b455..50ffe2b 100644 --- a/src/util/util/ipc/listener/UnixSocketListener.ts +++ b/src/util/util/ipc/listener/UnixSocketListener.ts @@ -20,19 +20,46 @@ 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; + openListenersMetric: Gauge.Internal; 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 @@ 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 @@ }); socket.on("close", () => { console.log("[UnixSocketListener] Unix socket client disconnected"); + this.openConnectionsMetric.set(this.server.connections ?? 0); }); }); @@ -90,6 +119,7 @@ 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 @@ }; 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); diff --git a/src/util/util/ipc/writer/UnixSocketWriter.ts b/src/util/util/ipc/writer/UnixSocketWriter.ts index 357034d..b50f53d 100644 --- a/src/util/util/ipc/writer/UnixSocketWriter.ts +++ b/src/util/util/ipc/writer/UnixSocketWriter.ts @@ -23,8 +23,12 @@ 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 @@ broadcastLock: Promise = Promise.resolve(); replayLock: Promise = Promise.resolve(); isInitializing = true; + openConnectionsMetric: Gauge.Internal; 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 @@ 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 @@ 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 @@ } } this.clients = {}; + UnixSocketWriter.openConnectionsMetric.remove({ path: this.socketPath }); } } diff --git a/src/util/util/lambert-server/Server.ts b/src/util/util/lambert-server/Server.ts index 617b279..4a73f7c 100644 --- a/src/util/util/lambert-server/Server.ts +++ b/src/util/util/lambert-server/Server.ts @@ -54,7 +54,15 @@ 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); + 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, + ); if (this.options.serverInitLogging && process.env.LOG_ROUTES !== "false") console.log(`[Server] Route ${path} registered`); diff --git a/src/webrtc/Server.ts b/src/webrtc/Server.ts index 8bc3402..f3e8e2c 100644 --- a/src/webrtc/Server.ts +++ b/src/webrtc/Server.ts @@ -23,6 +23,7 @@ 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 @@ 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 @@ } async start(): Promise { + await Monitoring.init(); await initDatabase(); await Config.init(); await initEvent();