feat: PRT - adding metric to check number of websocket connections ac…

…tive at any given time (#1735)

protocol/chainlib/consumer_websocket_manager.go

@@ -96,6 +96,10 @@ func (cwm *ConsumerWebsocketManager) handleRateLimitReached(inpData []byte) ([]b
 }
 
 func (cwm *ConsumerWebsocketManager) ListenToMessages() {
+	// adding metrics for how many active connections we have.
+	cwm.rpcConsumerLogs.SetWebSocketConnectionActive(cwm.chainId, cwm.apiInterface, true)
+	defer cwm.rpcConsumerLogs.SetWebSocketConnectionActive(cwm.chainId, cwm.apiInterface, false)
+
 	var (
 		messageType int
 		msg         []byte

protocol/metrics/metrics_consumer_manager.go

@@ -44,6 +44,7 @@ type ConsumerMetricsManager struct {
 	totalFailedWsSubscriptionRequestsMetric     *prometheus.CounterVec
 	totalWsSubscriptionDissconnectMetric        *prometheus.CounterVec
 	totalDuplicatedWsSubscriptionRequestsMetric *prometheus.CounterVec
+	totalWebSocketConnectionsActive             *prometheus.GaugeVec
 	blockMetric                                 *prometheus.GaugeVec
 	latencyMetric                               *prometheus.GaugeVec
 	qosMetric                                   *prometheus.GaugeVec
@@ -113,6 +114,11 @@ func NewConsumerMetricsManager(options ConsumerMetricsManagerOptions) *ConsumerM
 		Help: "The total number of duplicated webscket subscription requests over time per chain id per api interface.",
 	}, []string{"spec", "apiInterface"})
 
+	totalWebSocketConnectionsActive := prometheus.NewGaugeVec(prometheus.GaugeOpts{
+		Name: "lava_consumer_total_websocket_connections_active",
+		Help: "The total number of currently active websocket connections with users",
+	}, []string{"spec", "apiInterface"})
+
 	totalWsSubscriptionDissconnectMetric := prometheus.NewCounterVec(prometheus.CounterOpts{
 		Name: "lava_consumer_total_ws_subscription_disconnect",
 		Help: "The total number of websocket subscription disconnects over time per chain id per api interface per dissconnect reason.",
@@ -218,6 +224,7 @@ func NewConsumerMetricsManager(options ConsumerMetricsManagerOptions) *ConsumerM
 	prometheus.MustRegister(endpointsHealthChecksOkMetric)
 	prometheus.MustRegister(protocolVersionMetric)
 	prometheus.MustRegister(totalRelaysSentByNewBatchTickerMetric)
+	prometheus.MustRegister(totalWebSocketConnectionsActive)
 	prometheus.MustRegister(apiSpecificsMetric)
 	prometheus.MustRegister(averageLatencyMetric)
 	prometheus.MustRegister(totalRelaysSentToProvidersMetric)
@@ -238,6 +245,7 @@ func NewConsumerMetricsManager(options ConsumerMetricsManagerOptions) *ConsumerM
 		totalFailedWsSubscriptionRequestsMetric:     totalFailedWsSubscriptionRequestsMetric,
 		totalDuplicatedWsSubscriptionRequestsMetric: totalDuplicatedWsSubscriptionRequestsMetric,
 		totalWsSubscriptionDissconnectMetric:        totalWsSubscriptionDissconnectMetric,
+		totalWebSocketConnectionsActive:             totalWebSocketConnectionsActive,
 		totalErroredMetric:                          totalErroredMetric,
 		blockMetric:                                 blockMetric,
 		latencyMetric:                               latencyMetric,
@@ -297,6 +305,17 @@ func (pme *ConsumerMetricsManager) SetRelaySentToProviderMetric(chainId string,
 	pme.totalRelaysSentToProvidersMetric.WithLabelValues(chainId, apiInterface).Inc()
 }
 
+func (pme *ConsumerMetricsManager) SetWebSocketConnectionActive(chainId string, apiInterface string, add bool) {
+	if pme == nil {
+		return
+	}
+	if add {
+		pme.totalWebSocketConnectionsActive.WithLabelValues(chainId, apiInterface).Add(1)
+	} else {
+		pme.totalWebSocketConnectionsActive.WithLabelValues(chainId, apiInterface).Sub(1)
+	}
+}
+
 func (pme *ConsumerMetricsManager) SetRelayNodeErrorMetric(chainId string, apiInterface string) {
 	if pme == nil {
 		return

protocol/metrics/rpcconsumerlogs.go

@@ -87,6 +87,10 @@ func NewRPCConsumerLogs(consumerMetricsManager *ConsumerMetricsManager, consumer
 	return rpcConsumerLogs, err
 }
 
+func (rpccl *RPCConsumerLogs) SetWebSocketConnectionActive(chainId string, apiInterface string, add bool) {
+	rpccl.consumerMetricsManager.SetWebSocketConnectionActive(chainId, apiInterface, add)
+}
+
 func (rpccl *RPCConsumerLogs) SetRelaySentToProviderMetric(chainId string, apiInterface string) {
 	rpccl.consumerMetricsManager.SetRelaySentToProviderMetric(chainId, apiInterface)
 }