Merge pull request #8451 from sagemathinc/conat-router-metrics

conat/server: add prometheus metrics

Author: williamstein

src/packages/conat/core/server.ts

@@ -74,6 +74,7 @@ import { type SysConatServer, sysApiSubject, sysApi } from "./sys";
 import { forkedConatServer } from "./start-server";
 import { stickyChoice } from "./sticky";
 import { EventEmitter } from "events";
+import { Metrics } from "../types";
 
 const logger = getLogger("conat:core:server");
 
@@ -310,6 +311,10 @@ export class ConatServer extends EventEmitter {
     });
   };
 
+  public getUsage = (): Metrics => {
+    return this.usage.getMetrics();
+  };
+
   // this is for the Kubernetes health check -- I haven't
   // thought at all about what to do here, really.
   // Hopefully experience can teach us.
@@ -602,7 +607,9 @@ export class ConatServer extends EventEmitter {
       return;
     }
     if (!(await this.isAllowed({ user, subject, type: "sub" }))) {
-      const message = `permission denied subscribing to '${subject}' from ${JSON.stringify(user)}`;
+      const message = `permission denied subscribing to '${subject}' from ${JSON.stringify(
+        user,
+      )}`;
       this.log(message);
       throw new ConatError(message, {
         code: 403,
@@ -706,7 +713,9 @@ export class ConatServer extends EventEmitter {
     }
 
     if (!(await this.isAllowed({ user: from, subject, type: "pub" }))) {
-      const message = `permission denied publishing to '${subject}' from ${JSON.stringify(from)}`;
+      const message = `permission denied publishing to '${subject}' from ${JSON.stringify(
+        from,
+      )}`;
       this.log(message);
       throw new ConatError(message, {
         // this is the http code for permission denied, and having this
@@ -950,7 +959,9 @@ export class ConatServer extends EventEmitter {
           return;
         }
         if (!(await this.isAllowed({ user, subject, type: "pub" }))) {
-          const message = `permission denied waiting for interest in '${subject}' from ${JSON.stringify(user)}`;
+          const message = `permission denied waiting for interest in '${subject}' from ${JSON.stringify(
+            user,
+          )}`;
           this.log(message);
           respond({ error: message, code: 403 });
         }
@@ -1791,7 +1802,9 @@ export function updateSticky(update: StickyUpdate, sticky: Sticky): boolean {
 function getServerAddress(options: Options) {
   const port = options.port;
   const path = options.path?.slice(0, -"/conat".length) ?? "";
-  return `http${options.ssl || port == 443 ? "s" : ""}://${options.clusterIpAddress ?? "localhost"}:${port}${path}`;
+  return `http${options.ssl || port == 443 ? "s" : ""}://${
+    options.clusterIpAddress ?? "localhost"
+  }:${port}${path}`;
 }
 
 /*

src/packages/conat/monitor/usage.ts

@@ -1,8 +1,10 @@
-import json from "json-stable-stringify";
 import { EventEmitter } from "events";
-import type { JSONValue } from "@cocalc/util/types";
-import { ConatError } from "@cocalc/conat/core/client";
+import json from "json-stable-stringify";
+
 import { getLogger } from "@cocalc/conat/client";
+import { ConatError } from "@cocalc/conat/core/client";
+import type { JSONValue } from "@cocalc/util/types";
+import { Metrics } from "../types";
 
 const logger = getLogger("monitor:usage");
 
@@ -17,6 +19,9 @@ export class UsageMonitor extends EventEmitter {
   private options: Options;
   private total = 0;
   private perUser: { [user: string]: number } = {};
+  // metrics will be picked up periodically and exposed via e.g. prometheus
+  private countDeny = 0;
+  private metrics: Metrics = {};
 
   constructor(options: Options) {
     super();
@@ -38,27 +43,53 @@ export class UsageMonitor extends EventEmitter {
 
   private initLogging = () => {
     const { log } = this.options;
-    if (log == null) {
-      return;
-    }
+
+    // Record metrics for all events (even if logging is disabled)
     this.on("total", (total, limit) => {
-      log("usage", this.options.resource, { total, limit });
+      this.metrics["total:count"] = total;
+      this.metrics["total:limit"] = limit;
+      if (log) {
+        log("usage", this.options.resource, { total, limit });
+      }
     });
     this.on("add", (user, count, limit) => {
-      log("usage", this.options.resource, "add", { user, count, limit });
+      // this.metrics["add:count"] = count;
+      // this.metrics["add:limit"] = limit;
+      if (log) {
+        log("usage", this.options.resource, "add", { user, count, limit });
+      }
     });
     this.on("delete", (user, count, limit) => {
-      log("usage", this.options.resource, "delete", { user, count, limit });
+      // this.metrics["delete:count"] = count;
+      // this.metrics["delete:limit"] = limit;
+      if (log) {
+        log("usage", this.options.resource, "delete", { user, count, limit });
+      }
     });
     this.on("deny", (user, limit, type) => {
-      log("usage", this.options.resource, "not allowed due to hitting limit", {
-        type,
-        user,
-        limit,
-      });
+      this.countDeny += 1;
+      this.metrics["deny:count"] = this.countDeny;
+      this.metrics["deny:limit"] = limit;
+      if (log) {
+        log(
+          "usage",
+          this.options.resource,
+          "not allowed due to hitting limit",
+          {
+            type,
+            user,
+            limit,
+          },
+        );
+      }
     });
   };
 
+  // we return a copy
+  getMetrics = () => {
+    return { ...this.metrics };
+  };
+
   add = (user: JSONValue) => {
     const u = this.toJson(user);
     let count = this.perUser[u] ?? 0;

src/packages/conat/types.ts

@@ -9,3 +9,8 @@ export interface Location {
 
   path?: string;
 }
+
+type EventType = "total" | "add" | "delete" | "deny";
+type ValueType = "count" | "limit";
+type MetricKey = `${EventType}:${ValueType}`;
+export type Metrics = { [K in MetricKey]?: number };

src/packages/server/conat/socketio/health.ts

@@ -3,27 +3,23 @@ Health check for use in Kubernetes.
 */
 
 import type { ConatServer } from "@cocalc/conat/core/server";
-import { conatClusterHealthPort } from "@cocalc/backend/data";
-import { createServer } from "http";
+import type { IncomingMessage, ServerResponse } from "http";
 import { getLogger } from "@cocalc/backend/logger";
 
 const logger = getLogger("conat:socketio:health");
 
-export async function health(server: ConatServer) {
-  logger.debug(
-    `starting /health socketio server on port ${conatClusterHealthPort}`,
-  );
-  const healthServer = createServer();
-  healthServer.listen(conatClusterHealthPort);
-  healthServer.on("request", (req, res) => {
-    if (req.method === "GET") {
-      if (server.isHealthy()) {
-        res.statusCode = 200;
-        res.end("healthy");
-      } else {
-        res.statusCode = 500;
-        res.end("Unhealthy");
-      }
-    }
-  });
+export function handleHealth(
+  server: ConatServer,
+  _req: IncomingMessage,
+  res: ServerResponse,
+) {
+  const healthy = server.isHealthy();
+  logger.debug("/health reporting conat is healthy=${healthy}");
+  if (healthy) {
+    res.statusCode = 200;
+    res.end("healthy");
+  } else {
+    res.statusCode = 500;
+    res.end("Unhealthy");
+  }
 }

src/packages/server/conat/socketio/metrics.ts

@@ -0,0 +1,58 @@
+/*
+Metrics endpoint for use in Kubernetes.
+*/
+
+import { delay } from "awaiting";
+import type { IncomingMessage, ServerResponse } from "http";
+import { Gauge, register } from "prom-client";
+
+import { getLogger } from "@cocalc/backend/logger";
+import type { ConatServer } from "@cocalc/conat/core/server";
+import { Metrics } from "@cocalc/conat/types";
+
+const logger = getLogger("conat:socketio:metrics");
+
+const DELAY_MS = 10_000;
+
+const usageMetric = new Gauge({
+  name: "cocalc_conat_usage",
+  help: "Conat server usage metrics",
+  labelNames: ["event", "value"],
+});
+
+// periodically grab the metrics and set the Gauge, avoids an event emitter callback memory leak
+export async function initMetrics(server: ConatServer) {
+  logger.debug("metrics endpoint initialized");
+
+  await delay(DELAY_MS);
+  while (server.state != "closed") {
+    try {
+      const usage: Metrics = server.getUsage();
+      if (usage) {
+        for (const [key, val] of Object.entries(usage)) {
+          const [event, value] = key.split(":");
+          usageMetric.set({ event, value }, val);
+        }
+      }
+    } catch (err) {
+      logger.debug(`WARNING: error retrieving metrics -- ${err}`);
+    }
+    await delay(DELAY_MS);
+  }
+}
+
+export async function handleMetrics(
+  _req: IncomingMessage,
+  res: ServerResponse,
+) {
+  try {
+    const metricsData = await register.metrics();
+    res.setHeader("Content-Type", "text/plain; version=0.0.4; charset=utf-8");
+    res.statusCode = 200;
+    res.end(metricsData);
+  } catch (error) {
+    logger.error("Error getting metrics:", error);
+    res.statusCode = 500;
+    res.end("Internal server error");
+  }
+}

src/packages/server/conat/socketio/server.ts

@@ -31,27 +31,31 @@ Then in another terminal, make a client connected to each:
 
 */
 
+import { hostname } from "node:os";
+import { join } from "node:path";
+
+import basePath from "@cocalc/backend/base-path";
+import "@cocalc/backend/conat";
+import "@cocalc/backend/conat/persist"; // initializes context
 import {
-  init as createConatServer,
-  type Options,
-} from "@cocalc/conat/core/server";
-import { getUser, isAllowed } from "./auth";
-import { secureRandomString } from "@cocalc/backend/misc";
-import {
+  conatClusterName as clusterName,
+  conatClusterHealthPort,
+  conatClusterPort,
   conatPassword,
   conatSocketioCount,
-  conatClusterPort,
-  conatClusterName as clusterName,
 } from "@cocalc/backend/data";
-import basePath from "@cocalc/backend/base-path";
+import { getLogger } from "@cocalc/backend/logger";
+import { secureRandomString } from "@cocalc/backend/misc";
 import port from "@cocalc/backend/port";
-import { join } from "path";
-import "@cocalc/backend/conat";
-import "@cocalc/backend/conat/persist"; // initializes context
+import type { ConatServer } from "@cocalc/conat/core/server";
+import {
+  init as createConatServer,
+  type Options,
+} from "@cocalc/conat/core/server";
+import { getUser, isAllowed } from "./auth";
 import { dnsScan, localAddress, SCAN_INTERVAL } from "./dns-scan";
-import { health } from "./health";
-import { hostname } from "node:os";
-import { getLogger } from "@cocalc/backend/logger";
+import { handleHealth } from "./health";
+import { initMetrics, handleMetrics } from "./metrics";
 
 const logger = getLogger("conat-server");
 
@@ -95,9 +99,8 @@ export async function init(
     opts.forgetClusterNodeInterval = 4 * SCAN_INTERVAL;
     const server = createConatServer(opts);
     // enable dns scanner
-    dnsScan(server);
-    // enable health checks
-    health(server);
+    dnsScan(server); // we don't await it, it runs forever
+    await startVitalsServer(server);
     return server;
   }
 
@@ -115,3 +118,21 @@ export async function init(
     });
   }
 }
+
+async function startVitalsServer(server: ConatServer) {
+  // create shared HTTP server for health and metrics endpoints
+  const { createServer } = await import("http");
+  const vitalsServer = createServer((req, res) => {
+    if (req.method === "GET" && req.url === "/health") {
+      handleHealth(server, req, res);
+    } else if (req.method === "GET" && req.url === "/metrics") {
+      handleMetrics(req, res);
+    } else {
+      res.statusCode = 404;
+      res.end("Not Found");
+    }
+  });
+  vitalsServer.listen(conatClusterHealthPort);
+  logger.debug(`starting vitals server on port ${conatClusterHealthPort}`);
+  initMetrics(server); // start prometheus metrics collection
+}