Merge pull request #1574 from multiversx/add-ws-subscriptions-metrics

added metric per topics

Author: bogdan-rosianu

src/common/metrics/api.metrics.service.ts

@@ -23,6 +23,7 @@ export class ApiMetricsService {
   private static transactionsPendingResultsCounter: Counter<string>;
   private static batchUpdatesCounter: Counter<string>;
   private static subscriptionsConnectionsGauge: Gauge<string>;
+  private static subscriptionsTopicConnectionsGauge: Gauge<string>;
 
   constructor(
     private readonly apiConfigService: ApiConfigService,
@@ -40,6 +41,14 @@ export class ApiMetricsService {
       });
     }
 
+    if (!ApiMetricsService.subscriptionsTopicConnectionsGauge) {
+      ApiMetricsService.subscriptionsTopicConnectionsGauge = new Gauge({
+        name: 'websocket_subscriptions_topic_connections',
+        help: 'Unique websocket clients per topic',
+        labelNames: ['topic'],
+      });
+    }
+
     if (!ApiMetricsService.vmQueriesHistogram) {
       ApiMetricsService.vmQueriesHistogram = new Histogram({
         name: 'vm_query',
@@ -191,10 +200,16 @@ export class ApiMetricsService {
   }
 
   @OnEvent(MetricsEvents.SetWebsocketMetrics)
-  setWebsocketSubscriptionsMetrics(payload: { connectedClients: number }) {
-    const { connectedClients } = payload;
+  setWebsocketSubscriptionsMetrics(payload: { connectedClients: number; topics?: Record<string, number> }) {
+    const { connectedClients, topics } = payload;
 
     ApiMetricsService.subscriptionsConnectionsGauge.set(connectedClients);
+
+    if (topics) {
+      for (const [topic, count] of Object.entries(topics)) {
+        ApiMetricsService.subscriptionsTopicConnectionsGauge.set({ topic }, count);
+      }
+    }
   }
 
 

src/crons/websocket/connection.handler.ts

@@ -15,7 +15,7 @@ export class ConnectionHandler implements OnGatewayDisconnect, OnGatewayConnecti
     handleDisconnect(_client: Socket) { }
 
     handleConnection(client: Socket, ..._args: any[]) {
-        client.setMaxListeners(12);
+        client.setMaxListeners(16);
     }
 
     hasSubscriptionsWithPrefixes(prefixes: string[]): boolean {

src/crons/websocket/websocket.cron.service.ts

@@ -45,21 +45,6 @@ export class WebsocketCronService {
     private readonly apiConfigService: ApiConfigService,
   ) { }
 
-  @Cron('*/1 * * * * *')
-  handleWebsocketMetrics() {
-    const connectedClients = this.server.sockets.sockets.size ?? 0;
-    // TODO: add more metrics in the future
-    // const subscriptions: Record<string, number> = {};
-
-    // this.server.sockets.adapter.rooms.forEach((socketsSet, roomName) => {
-    //   subscriptions[roomName] = socketsSet.size;
-    // });
-
-    this.eventEmitter.emit(MetricsEvents.SetWebsocketMetrics, {
-      connectedClients,
-    });
-  }
-
   @Cron('*/6 * * * * *')
   @Lock({ name: 'Push transactions to subscribers', verbose: true })
   async handleTransactionsUpdate() {
@@ -128,6 +113,59 @@ export class WebsocketCronService {
     );
   }
 
+  @Cron('*/10 * * * * *')
+  @Lock({ name: 'Push websocket subscriptions metrics', verbose: true })
+  handleWebsocketMetrics() {
+    const connectedClients = this.server.sockets.sockets.size ?? 0;
+
+    // Efficient unique-listener counts per topic
+    const adapter = this.server.sockets.adapter as any;
+    const sids: Map<string, Set<string>> = adapter?.sids ?? new Map();
+
+    const topicPrefixes: Array<{ key: string; prefix?: string; room?: string }> = [
+      { key: 'tx', prefix: TransactionsGateway.keyPrefix },
+      { key: 'customTx', prefix: TransactionsCustomGateway.keyPrefix },
+      { key: 'events', prefix: EventsGateway.keyPrefix },
+      { key: 'customEvents', prefix: EventsCustomGateway.keyPrefix },
+      { key: 'blocks', prefix: BlocksGateway.keyPrefix },
+      { key: 'pool', prefix: PoolGateway.keyPrefix },
+      { key: 'stats', room: 'statsRoom' },
+    ];
+
+    const topics: Record<string, number> = {};
+    for (const { key } of topicPrefixes) topics[key] = 0;
+
+    // Count unique sockets per prefix-based topic by scanning socket -> rooms map once
+    if (sids && sids.size > 0) {
+      for (const [, rooms] of sids) {
+        // Track whether this socket has been counted for a given topic key
+        const matched: Record<string, boolean> = {};
+
+        for (const roomName of rooms) {
+          for (const { key, prefix } of topicPrefixes) {
+            if (!prefix || matched[key]) continue;
+            if (roomName.startsWith(prefix)) {
+              topics[key] += 1;
+              matched[key] = true;
+            }
+          }
+        }
+      }
+    }
+
+    // Handle exact-room topics (like statsRoom) directly from rooms map
+    const rooms: Map<string, Set<string>> = adapter?.rooms ?? new Map();
+    const statsRoomSet = rooms.get('statsRoom');
+    if (statsRoomSet) {
+      topics['stats'] = statsRoomSet.size;
+    }
+
+    this.eventEmitter.emit(MetricsEvents.SetWebsocketMetrics, {
+      connectedClients,
+      topics,
+    });
+  }
+
   private async getLatestRoundOnChainData() {
     const rounds = await this.roundService.getRounds(new RoundFilter({ size: 1 }));
     return rounds[0];