feat: open WebSocket and streams metrics

lib/Prometheus.ts

@@ -159,15 +159,40 @@ class Prometheus {
       }),
     );
 
-    const swapStatusCacheCount = new Gauge({
-      name: `${Prometheus.metric_prefix}swap_status_cache_count`,
-      help: 'swap status messages cached',
-    });
-    setInterval(
-      () => swapStatusCacheCount.set(this.api.swapInfos.cacheSize),
-      Prometheus.gaugeUpdateInterval,
+    this.registerGauge(
+      this.swapRegistry!,
+      'swap_status_cache_count',
+      'number of swap status messages cached',
+      () => this.api.swapInfos.cacheSize,
+    );
+
+    this.registerGauge(
+      this.swapRegistry!,
+      'pending_stream_count',
+      'pending SSE swap status stream count',
+      () => this.api.controller.pendingStreamCount,
+    );
+
+    this.registerGauge(
+      this.swapRegistry!,
+      'open_websocket_count',
+      'open WebSockets count',
+      () => this.api.websocket.openWebSocketCount,
     );
-    this.swapRegistry!.registerMetric(swapStatusCacheCount);
+  };
+
+  private registerGauge = (
+    registry: Registry,
+    name: string,
+    help: string,
+    cb: () => number,
+  ) => {
+    const gauge = new Gauge({
+      help,
+      name: `${Prometheus.metric_prefix}${name}`,
+    });
+    setInterval(() => gauge.set(cb()), Prometheus.gaugeUpdateInterval);
+    registry.registerMetric(gauge);
   };
 }
 

lib/api/Api.ts

@@ -12,12 +12,11 @@ import WebSocketHandler from './v2/WebSocketHandler';
 
 class Api {
   public readonly swapInfos: SwapInfos;
+  public readonly controller: Controller;
+  public readonly websocket: WebSocketHandler;
 
   private app: Application;
 
-  private readonly websocket: WebSocketHandler;
-  private readonly controller: Controller;
-
   constructor(
     private readonly logger: Logger,
     private readonly config: ApiConfig,

lib/api/Controller.ts

@@ -41,6 +41,10 @@ class Controller {
     });
   }
 
+  public get pendingStreamCount() {
+    return this.pendingSwapStreams.size;
+  }
+
   // Static files
   public serveFile = (fileName: string) => {
     return (_: Request, res: Response): void => {

lib/api/v2/WebSocketHandler.ts

@@ -58,6 +58,10 @@ class WebSocketHandler {
     this.listenSwapUpdates();
   }
 
+  public get openWebSocketCount() {
+    return this.socketToSwaps.size;
+  }
+
   public register = (server: http.Server) => {
     server.on('upgrade', (request, socket, head) => {
       this.ws.handleUpgrade(request, socket, head, (ws) => {