From c4f4500fe5b5b393de5f3ced9c696f008a6b3a4b Mon Sep 17 00:00:00 2001 From: kenedybokephraim-boop Date: Wed, 22 Jul 2026 16:20:58 +0100 Subject: [PATCH] feat: add co-located HTTP server with health checks, Prometheus metrics, and WebSocket upgrade detection on shared port Implement an HTTP server as the primary listener using Node.js built-in http module, with WebSocketServer running in noServer: true mode. Incoming requests are inspected for the Upgrade header to correctly route WebSocket upgrades vs plain HTTP requests. Endpoints: - GET /healthz: Liveness probe returning 200 with uptime (503 during shutdown) - GET /readyz: Readiness probe returning 200 with connection/room counts - GET /metrics: Prometheus exposition format with 7 metrics (connections, rooms, messages, rate limit rejections, auth failures, heap usage, event loop lag) Features: - Protocol detection on shared port (HTTP vs WebSocket upgrade) - Event loop lag measurement via setTimeout(0) drift every 5 seconds - Atomic metrics counters for message types, auth failures, and rate limits - Graceful shutdown integration: markShuttingDown sets probes to 503, then closes both WSS and HTTP server - wss.address() patched to return httpServer.address() for test compatibility Closes #199 --- src/index.js | 18 ++- src/server.js | 124 +++++++++++++++++-- tests/http-endpoints.test.js | 223 +++++++++++++++++++++++++++++++++++ 3 files changed, 352 insertions(+), 13 deletions(-) create mode 100644 tests/http-endpoints.test.js diff --git a/src/index.js b/src/index.js index dceaf00..fded7a2 100644 --- a/src/index.js +++ b/src/index.js @@ -32,8 +32,10 @@ if (isNaN(config.maxPayloadBytes) || config.maxPayloadBytes < 1) { } let wss; +let httpServer; +let markShuttingDown; try { - ({ wss } = createServer(config)); + ({ wss, httpServer, markShuttingDown } = createServer(config)); } catch (err) { logger.error("Failed to start server", { error: err.message }); process.exit(1); @@ -42,7 +44,7 @@ try { logger.info("Gateway started", config); /** - * Initiates a graceful shutdown of the WebSocket server. + * Initiates a graceful shutdown of the server. * * Closes the server and waits for existing connections to finish. If the * server does not close within 5 seconds, a forced exit is triggered. @@ -63,8 +65,16 @@ export function shutdown(server, signal) { }, 5000); } -process.on("SIGTERM", () => shutdown(wss, "SIGTERM")); -process.on("SIGINT", () => shutdown(wss, "SIGINT")); +process.on("SIGTERM", () => { + markShuttingDown(); + wss.close(); + shutdown(httpServer, "SIGTERM"); +}); +process.on("SIGINT", () => { + markShuttingDown(); + wss.close(); + shutdown(httpServer, "SIGINT"); +}); process.on("uncaughtException", (err) => { logger.error("Uncaught exception", { error: err.message }); diff --git a/src/server.js b/src/server.js index 6f0b84f..dab01da 100644 --- a/src/server.js +++ b/src/server.js @@ -1,3 +1,4 @@ +import http from "node:http"; import { WebSocketServer } from "ws"; import { v4 as uuid } from "uuid"; import { RoomManager } from "./room-manager.js"; @@ -7,16 +8,94 @@ import { logger } from "./logger.js"; import { createConnRateLimiter } from "./conn-rate-limiter.js"; export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit, maxConnectionsPerIp } = {}) { - const wss = new WebSocketServer({ - port: port ?? 8080, - maxPayload: maxPayloadBytes ?? 1024, - }); - const rooms = new RoomManager(); const connRateLimiter = createConnRateLimiter(connRateLimit); const ipConnectionCount = new Map(); const MAX_CONNS_PER_IP = maxConnectionsPerIp ?? (Number(process.env.MAX_CONNECTIONS_PER_IP) || 10); + const metrics = { + messages: { location_update: 0, join_room: 0, leave_room: 0 }, + authFailures: 0, + rateLimitRejections: { connection: 0 }, + eventLoopLagMs: 0, + }; + + let isReady = false; + let isShuttingDown = false; + + const wss = new WebSocketServer({ + noServer: true, + maxPayload: maxPayloadBytes ?? 1024, + }); + + const httpServer = http.createServer((req, res) => { + if (req.method !== "GET") { + res.writeHead(405); + res.end("Method Not Allowed"); + return; + } + + const pathname = new URL(req.url, `http://${req.headers.host ?? "localhost"}`).pathname; + + if (pathname === "/healthz") { + if (isShuttingDown) { + res.writeHead(503, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ status: "shutting down" })); + return; + } + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ status: "ok", uptime: process.uptime() })); + } else if (pathname === "/readyz") { + if (isShuttingDown) { + res.writeHead(503, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ status: "not ready", reason: "server is shutting down" })); + return; + } + if (!isReady) { + res.writeHead(503, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ status: "not ready", reason: "initializing" })); + return; + } + res.writeHead(200, { "Content-Type": "application/json" }); + res.end(JSON.stringify({ + status: "ready", + connections: wss.clients.size, + rooms: rooms.roomCount, + })); + } else if (pathname === "/metrics") { + const mem = process.memoryUsage(); + const lines = [ + "# TYPE gateway_connections_active gauge", + `gateway_connections_active ${wss.clients.size}`, + "# TYPE gateway_rooms_active gauge", + `gateway_rooms_active ${rooms.roomCount}`, + "# TYPE gateway_messages_total counter", + `gateway_messages_total{type="location_update"} ${metrics.messages.location_update}`, + `gateway_messages_total{type="join_room"} ${metrics.messages.join_room}`, + `gateway_messages_total{type="leave_room"} ${metrics.messages.leave_room}`, + "# TYPE gateway_rate_limit_rejections_total counter", + `gateway_rate_limit_rejections_total{kind="connection"} ${metrics.rateLimitRejections.connection}`, + "# TYPE gateway_auth_failures_total counter", + `gateway_auth_failures_total ${metrics.authFailures}`, + "# TYPE gateway_heap_used_bytes gauge", + `gateway_heap_used_bytes ${mem.heapUsed}`, + "# TYPE gateway_event_loop_lag_ms gauge", + `gateway_event_loop_lag_ms ${metrics.eventLoopLagMs}`, + ]; + res.writeHead(200, { "Content-Type": "text/plain; version=0.0.4; charset=utf-8" }); + res.end(lines.join("\n") + "\n"); + } else { + res.writeHead(404); + res.end("Not Found"); + } + }); + + httpServer.on("upgrade", (req, socket, head) => { + wss.handleUpgrade(req, socket, head, (ws) => { + wss.emit("connection", ws, req); + }); + }); + function heartbeat() { this.isAlive = true; } @@ -29,6 +108,7 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit if (!connRateLimiter.check(ip)) { logger.warn("Connection rate limit exceeded", { ip }); + metrics.rateLimitRejections.connection++; ws.close(4029, "Connection rate limit exceeded"); return; } @@ -36,6 +116,7 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit const currentCount = ipConnectionCount.get(ip) ?? 0; if (currentCount >= MAX_CONNS_PER_IP) { logger.warn("Max connections per IP exceeded", { ip }); + metrics.rateLimitRejections.connection++; ws.close(4029, "Too many connections from this IP"); return; } @@ -56,6 +137,7 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit if (!authResult.ok) { logger.warn("Authentication failed", { clientId, reason: authResult.error }); + metrics.authFailures++; ws.close(4001, authResult.error); return; } @@ -79,17 +161,20 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit switch (msg.type) { case "join_room": { rooms.join(actualClientId, msg.roomId, ws); + metrics.messages.join_room++; logger.info("Client joined room", { clientId: actualClientId, roomId: msg.roomId }); ws.send(JSON.stringify({ type: "room_joined", payload: { roomId: msg.roomId } })); break; } case "leave_room": { rooms.leave(actualClientId, msg.roomId); + metrics.messages.leave_room++; logger.info("Client left room", { clientId: actualClientId, roomId: msg.roomId }); ws.send(JSON.stringify({ type: "room_left", payload: { roomId: msg.roomId } })); break; } case "location_update": { + metrics.messages.location_update++; const roomIds = rooms.getClientRooms(actualClientId); for (const roomId of roomIds) { rooms.broadcast(roomId, { @@ -125,7 +210,7 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit }); }); - const interval = setInterval(() => { + const heartbeatInterval = setInterval(() => { wss.clients.forEach((ws) => { if (ws.isAlive === false) { logger.warn("Terminating zombie connection", { clientId: ws._clientId ?? "unknown" }); @@ -136,9 +221,30 @@ export function createServer({ port, heartbeatMs, maxPayloadBytes, connRateLimit }); }, heartbeatMs ?? 30000); - wss.on("close", () => { - clearInterval(interval); + function measureLag() { + const start = Date.now(); + setTimeout(() => { + metrics.eventLoopLagMs = Date.now() - start; + }, 0); + } + const lagInterval = setInterval(measureLag, 5000); + measureLag(); + + httpServer.on("close", () => { + clearInterval(heartbeatInterval); + clearInterval(lagInterval); }); - return { wss, rooms, ipConnectionCount }; + httpServer.listen(port ?? 8080); + + wss.address = () => httpServer.address(); + + function markShuttingDown() { + isShuttingDown = true; + isReady = false; + } + + isReady = true; + + return { wss, httpServer, rooms, metrics, ipConnectionCount, markShuttingDown }; } diff --git a/tests/http-endpoints.test.js b/tests/http-endpoints.test.js new file mode 100644 index 0000000..1a7c5f1 --- /dev/null +++ b/tests/http-endpoints.test.js @@ -0,0 +1,223 @@ +import { describe, it, expect, beforeEach, afterEach } from "vitest"; +import http from "node:http"; +import WebSocket from "ws"; +import jwt from "jsonwebtoken"; +import { createServer } from "../src/server.js"; + +const TEST_SECRET = "test-secret-key"; + +function makeToken(clientId) { + return jwt.sign({ sub: clientId }, TEST_SECRET, { expiresIn: 60 }); +} + +function httpGet(port, path) { + return new Promise((resolve, reject) => { + const req = http.get(`http://localhost:${port}${path}`, (res) => { + const chunks = []; + res.on("data", (chunk) => chunks.push(chunk)); + res.on("end", () => { + resolve({ status: res.statusCode, headers: res.headers, body: Buffer.concat(chunks).toString() }); + }); + }); + req.on("error", reject); + }); +} + +describe("HTTP endpoints", () => { + let server; + let port; + + beforeEach(() => { + process.env.AUTH_SECRET = TEST_SECRET; + server = createServer({ port: 0, heartbeatMs: 60000, maxPayloadBytes: 4096 }); + port = server.httpServer.address().port; + }); + + afterEach(async () => { + for (const client of server.wss.clients) { + client.terminate(); + } + server.wss.close(); + await new Promise((resolve) => server.httpServer.close(resolve)); + delete process.env.AUTH_SECRET; + }); + + describe("GET /healthz", () => { + it("returns 200 with status ok and uptime", async () => { + const res = await httpGet(port, "/healthz"); + expect(res.status).toBe(200); + expect(res.headers["content-type"]).toContain("application/json"); + const body = JSON.parse(res.body); + expect(body.status).toBe("ok"); + expect(typeof body.uptime).toBe("number"); + expect(body.uptime).toBeGreaterThanOrEqual(0); + }); + + it("returns 503 during shutdown", async () => { + server.markShuttingDown(); + const res = await httpGet(port, "/healthz"); + expect(res.status).toBe(503); + const body = JSON.parse(res.body); + expect(body.status).toBe("shutting down"); + }); + }); + + describe("GET /readyz", () => { + it("returns 200 with ready status when initialized", async () => { + const res = await httpGet(port, "/readyz"); + expect(res.status).toBe(200); + const body = JSON.parse(res.body); + expect(body.status).toBe("ready"); + expect(typeof body.connections).toBe("number"); + expect(typeof body.rooms).toBe("number"); + }); + + it("returns 503 during shutdown", async () => { + server.markShuttingDown(); + const res = await httpGet(port, "/readyz"); + expect(res.status).toBe(503); + const body = JSON.parse(res.body); + expect(body.status).toBe("not ready"); + expect(body.reason).toBe("server is shutting down"); + }); + + it("reports active connections and rooms", async () => { + const token = makeToken("ready-client"); + const ws = new WebSocket(`ws://localhost:${port}/?token=${token}`); + await new Promise((resolve) => ws.once("open", resolve)); + + const joinMsg = JSON.stringify({ type: "join_room", roomId: "ready-room" }); + await new Promise((resolve) => { + ws.once("message", resolve); + ws.send(joinMsg); + }); + + const res = await httpGet(port, "/readyz"); + const body = JSON.parse(res.body); + expect(body.connections).toBeGreaterThanOrEqual(1); + expect(body.rooms).toBeGreaterThanOrEqual(1); + + ws.close(); + await new Promise((resolve) => ws.once("close", resolve)); + }); + }); + + describe("GET /metrics", () => { + it("returns 200 with Prometheus content type", async () => { + const res = await httpGet(port, "/metrics"); + expect(res.status).toBe(200); + expect(res.headers["content-type"]).toBe("text/plain; version=0.0.4; charset=utf-8"); + }); + + it("contains all required metric names", async () => { + const res = await httpGet(port, "/metrics"); + expect(res.body).toContain("gateway_connections_active"); + expect(res.body).toContain("gateway_rooms_active"); + expect(res.body).toContain("gateway_messages_total"); + expect(res.body).toContain("gateway_rate_limit_rejections_total"); + expect(res.body).toContain("gateway_auth_failures_total"); + expect(res.body).toContain("gateway_heap_used_bytes"); + expect(res.body).toContain("gateway_event_loop_lag_ms"); + }); + + it("contains valid Prometheus TYPE declarations", async () => { + const res = await httpGet(port, "/metrics"); + expect(res.body).toContain("# TYPE gateway_connections_active gauge"); + expect(res.body).toContain("# TYPE gateway_rooms_active gauge"); + expect(res.body).toContain("# TYPE gateway_messages_total counter"); + expect(res.body).toContain("# TYPE gateway_rate_limit_rejections_total counter"); + expect(res.body).toContain("# TYPE gateway_auth_failures_total counter"); + expect(res.body).toContain("# TYPE gateway_heap_used_bytes gauge"); + expect(res.body).toContain("# TYPE gateway_event_loop_lag_ms gauge"); + }); + + it("tracks active connections", async () => { + const res0 = await httpGet(port, "/metrics"); + expect(res0.body).toContain("gateway_connections_active 0"); + + const token = makeToken("metrics-client"); + const ws = new WebSocket(`ws://localhost:${port}/?token=${token}`); + await new Promise((resolve) => ws.once("open", resolve)); + + const res1 = await httpGet(port, "/metrics"); + expect(res1.body).toContain("gateway_connections_active 1"); + + ws.close(); + await new Promise((resolve) => ws.once("close", resolve)); + }); + + it("tracks message counts", async () => { + const token = makeToken("metrics-msg-client"); + const ws = new WebSocket(`ws://localhost:${port}/?token=${token}`); + await new Promise((resolve) => ws.once("open", resolve)); + + const joinMsg = JSON.stringify({ type: "join_room", roomId: "metrics-room" }); + await new Promise((resolve) => { + ws.once("message", resolve); + ws.send(joinMsg); + }); + + const res = await httpGet(port, "/metrics"); + expect(res.body).toContain('gateway_messages_total{type="join_room"} 1'); + expect(res.body).toContain('gateway_messages_total{type="leave_room"} 0'); + expect(res.body).toContain('gateway_messages_total{type="location_update"} 0'); + + ws.close(); + await new Promise((resolve) => ws.once("close", resolve)); + }); + + it("tracks heap usage", async () => { + const res = await httpGet(port, "/metrics"); + const match = res.body.match(/gateway_heap_used_bytes (\d+)/); + expect(match).not.toBeNull(); + const heapUsed = parseInt(match[1], 10); + expect(heapUsed).toBeGreaterThan(0); + }); + }); + + describe("WebSocket upgrade on shared port", () => { + it("allows WebSocket connections on the same port as HTTP", async () => { + const token = makeToken("shared-port-client"); + const ws = new WebSocket(`ws://localhost:${port}/?token=${token}`); + await new Promise((resolve) => ws.once("open", resolve)); + expect(ws.readyState).toBe(WebSocket.OPEN); + ws.close(); + await new Promise((resolve) => ws.once("close", resolve)); + }); + + it("does not intercept HTTP requests as WebSocket upgrades", async () => { + const httpRes = await httpGet(port, "/healthz"); + expect(httpRes.status).toBe(200); + const body = JSON.parse(httpRes.body); + expect(body.status).toBe("ok"); + + const token = makeToken("coexist-client"); + const ws = new WebSocket(`ws://localhost:${port}/?token=${token}`); + await new Promise((resolve) => ws.once("open", resolve)); + expect(ws.readyState).toBe(WebSocket.OPEN); + + ws.close(); + await new Promise((resolve) => ws.once("close", resolve)); + }); + }); + + describe("HTTP error handling", () => { + it("returns 405 for non-GET requests", async () => { + const res = await new Promise((resolve, reject) => { + const req = http.request(`http://localhost:${port}/healthz`, { method: "POST" }, (r) => { + const chunks = []; + r.on("data", (c) => chunks.push(c)); + r.on("end", () => resolve({ status: r.statusCode, body: Buffer.concat(chunks).toString() })); + }); + req.on("error", reject); + req.end(); + }); + expect(res.status).toBe(405); + }); + + it("returns 404 for unknown paths", async () => { + const res = await httpGet(port, "/unknown"); + expect(res.status).toBe(404); + }); + }); +});