add metrics on subscription

Author: stefangutica

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

@@ -22,6 +22,7 @@ export class ApiMetricsService {
   private static transactionsCompletedCounter: Counter<string>;
   private static transactionsPendingResultsCounter: Counter<string>;
   private static batchUpdatesCounter: Counter<string>;
+  private static subscriptionsConnectionsGauge: Gauge<string>;
 
   constructor(
     private readonly apiConfigService: ApiConfigService,
@@ -32,6 +33,13 @@ export class ApiMetricsService {
     private readonly metricsService: MetricsService,
   ) {
 
+    if (!ApiMetricsService.subscriptionsConnectionsGauge) {
+      ApiMetricsService.subscriptionsConnectionsGauge = new Gauge({
+        name: 'websocket_subscriptions_connections',
+        help: 'Number of websocket connections for subscriptions',
+      });
+    }
+
     if (!ApiMetricsService.vmQueriesHistogram) {
       ApiMetricsService.vmQueriesHistogram = new Histogram({
         name: 'vm_query',
@@ -182,6 +190,14 @@ export class ApiMetricsService {
     ApiMetricsService.lastProcessedTransactionCompletedProcessorNonce.set({ shardId }, nonce);
   }
 
+  @OnEvent(MetricsEvents.SetWebsocketMetrics)
+  setWebsocketSubscriptionsMetrics(payload: { connectedClients: number }) {
+    const { connectedClients } = payload;
+
+    ApiMetricsService.subscriptionsConnectionsGauge.set(connectedClients);
+  }
+
+
   @OnEvent(MetricsEvents.SetTransactionsCompleted)
   recordTransactionsCompleted(payload: { transactions: any[] }) {
     ApiMetricsService.transactionsCompletedCounter.inc(payload.transactions.length);

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

@@ -6,16 +6,40 @@ import { NetworkGateway } from 'src/crons/websocket/network.gateway';
 import { Lock } from "@multiversx/sdk-nestjs-common";
 import { PoolGateway } from 'src/crons/websocket/pool.gateway';
 import { EventsGateway } from 'src/crons/websocket/events.gateway';
+import { WebSocketGateway, WebSocketServer } from '@nestjs/websockets';
+import { EventEmitter2 } from '@nestjs/event-emitter';
+import { MetricsEvents } from 'src/utils/metrics-events.constants';
+import { Server } from 'socket.io';
 @Injectable()
+@WebSocketGateway({ cors: { origin: '*' }, path: '/ws/subscription' })
 export class WebsocketCronService {
+  @WebSocketServer()
+  server!: Server;
+
   constructor(
     private readonly transactionsGateway: TransactionsGateway,
     private readonly blocksGateway: BlocksGateway,
     private readonly networkGateway: NetworkGateway,
     private readonly poolGateway: PoolGateway,
     private readonly eventsGateway: EventsGateway,
+    private readonly eventEmitter: EventEmitter2,
   ) { }
 
+  @Cron('*/6 * * * * *')
+  async 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() {

src/utils/metrics-events.constants.ts

@@ -10,4 +10,5 @@ export enum MetricsEvents {
   SetTransactionsCompleted = "setTransactionsCompleted",
   SetTransactionsPendingResults = "setTransactionsPendingResults",
   SetBatchUpdated = "setBatchUpdated",
+  SetWebsocketMetrics = "setWebsocketMetrics",
 }