refactor(agent): channel capacity as histogram (#450)

* refactor(agent): channel capacity as histogram (#429)

Channel capacity is a metric of each agent, that is created in each handle
so the capacity varies depending on the client/agent being used and we need
to differentiate the lows/highs of each in the metric. Adding a client label
to the Gauge is not feasible, since it'd explode cardinality. Thus, adding
a new metric as an histogram so we maintain compat until deprecating the old
gauge one

* fix(metrics): tests and usage in nats rpc (#451)

* fix(metrics): missing parenthesis on histogram
* fix(metrics): report histogram in rpc channel

* refactor(metrics): remove depreacted channel capacity gauge

* chore(go): update to 1.23

Author: hspedro

.github/workflows/tests.yaml

@@ -20,7 +20,7 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v5
       with:
-        go-version: '1.22'
+        go-version: '1.23'
     - name: Download dependencies
       run: go mod download
   unit-test:
@@ -33,7 +33,7 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v5
       with:
-        go-version: '1.22'
+        go-version: '1.23'
     - name: Setup dependencies
       env:
         GO111MODULE: auto
@@ -54,7 +54,7 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v5
       with:
-        go-version: '1.22'
+        go-version: '1.23'
     - name: Run tests
       run: make e2e-test-nats
   e2e-test-grpc:
@@ -67,6 +67,6 @@ jobs:
     - name: Set up Go
       uses: actions/setup-go@v5
       with:
-        go-version: '1.22'
+        go-version: '1.23'
     - name: Run tests
       run: make e2e-test-grpc

.readthedocs.yaml

@@ -13,7 +13,7 @@ build:
     # You can also specify other tool versions:
     # nodejs: "20"
     # rust: "1.70"
-    golang: "1.22"
+    golang: "1.23"
 
 # Build documentation in the "docs/" directory with Sphinx
 sphinx:

go.mod

@@ -1,6 +1,6 @@
 module github.com/topfreegames/pitaya/v3
 
-go 1.22.0
+go 1.23
 
 require (
 	github.com/DataDog/datadog-go v4.8.3+incompatible

go.work

@@ -1,4 +1,4 @@
-go 1.22.0
+go 1.23
 
 toolchain go1.23.4
 

pkg/agent/agent.go

@@ -698,8 +698,8 @@ func (a *agentImpl) reportChannelSize() {
 		logger.Log.Warnf("chSend is at maximum capacity")
 	}
 	for _, mr := range a.metricsReporters {
-		if err := mr.ReportGauge(metrics.ChannelCapacity, map[string]string{"channel": "agent_chsend"}, float64(chSendCapacity)); err != nil {
-			logger.Log.Warnf("failed to report chSend channel capaacity: %s", err.Error())
+		if err := mr.ReportHistogram(metrics.ChannelCapacity, map[string]string{"channel": "agent_chsend"}, float64(chSendCapacity)); err != nil {
+			logger.Log.Warnf("failed to report histogram chSend channel capacity: %s", err.Error())
 		}
 	}
 }

pkg/agent/agent_test.go

@@ -275,7 +275,7 @@ func TestAgentSendSerializeErr(t *testing.T) {
 		wg.Done()
 	})
 	go ag.write()
-	mockMetricsReporter.EXPECT().ReportGauge(gomock.Any(), gomock.Any(), gomock.Any())
+	mockMetricsReporter.EXPECT().ReportHistogram(gomock.Any(), gomock.Any(), gomock.Any())
 	ag.send(expected)
 	wg.Wait()
 
@@ -347,7 +347,7 @@ func TestAgentPushStruct(t *testing.T) {
 				close(ag.chSend)
 			}
 
-			mockMetricsReporter.EXPECT().ReportGauge(metrics.ChannelCapacity, gomock.Any(), float64(10))
+			mockMetricsReporter.EXPECT().ReportHistogram(metrics.ChannelCapacity, gomock.Any(), float64(10))
 			err = ag.Push(msg.Route, table.data)
 			assert.Equal(t, table.err, err)
 
@@ -409,7 +409,7 @@ func TestAgentPush(t *testing.T) {
 				close(ag.chSend)
 			}
 
-			mockMetricsReporter.EXPECT().ReportGauge(metrics.ChannelCapacity, gomock.Any(), float64(10))
+			mockMetricsReporter.EXPECT().ReportHistogram(metrics.ChannelCapacity, gomock.Any(), float64(10))
 			err = ag.Push(msg.Route, table.data)
 			assert.Equal(t, table.err, err)
 
@@ -442,7 +442,7 @@ func TestAgentPushFullChannel(t *testing.T) {
 	ag := newAgent(mockConn, mockDecoder, mockEncoder, mockSerializer, hbTime, writeTimeout, 0, dieChan, messageEncoder, mockMetricsReporters, sessionPool).(*agentImpl)
 	assert.NotNil(t, ag)
 
-	mockMetricsReporter.EXPECT().ReportGauge(metrics.ChannelCapacity, gomock.Any(), float64(0))
+	mockMetricsReporter.EXPECT().ReportHistogram(metrics.ChannelCapacity, gomock.Any(), float64(0))
 
 	msg := &message.Message{
 		Route: "route",
@@ -528,7 +528,7 @@ func TestAgentResponseMID(t *testing.T) {
 			ctx := getCtxWithRequestKeys()
 			if table.mid != 0 {
 				mockEncoder.EXPECT().Encode(gomock.Any(), gomock.Any()).Return([]byte("ok!"), nil)
-				mockMetricsReporter.EXPECT().ReportGauge(metrics.ChannelCapacity, gomock.Any(), float64(10))
+				mockMetricsReporter.EXPECT().ReportHistogram(metrics.ChannelCapacity, gomock.Any(), float64(10))
 			}
 			if table.mid != 0 {
 				if table.err != nil {
@@ -581,7 +581,7 @@ func TestAgentResponseMIDFullChannel(t *testing.T) {
 	sessionPool := session.NewSessionPool()
 	ag := newAgent(mockConn, mockDecoder, mockEncoder, mockSerializer, hbTime, writeTimeout, 0, dieChan, messageEncoder, mockMetricsReporters, sessionPool).(*agentImpl)
 	assert.NotNil(t, ag)
-	mockMetricsReporters[0].(*metricsmocks.MockReporter).EXPECT().ReportGauge(metrics.ChannelCapacity, gomock.Any(), float64(0))
+	mockMetricsReporters[0].(*metricsmocks.MockReporter).EXPECT().ReportHistogram(metrics.ChannelCapacity, gomock.Any(), float64(0))
 	go func() {
 		err := ag.ResponseMID(nil, 1, []byte("data"))
 		assert.NoError(t, err)
@@ -1204,7 +1204,7 @@ func TestNatsRPCServerReportMetrics(t *testing.T) {
 
 	ag.chSend <- pendingWrite{}
 
-	mockMetricsReporter.EXPECT().ReportGauge(metrics.ChannelCapacity, gomock.Any(), float64(-1)) // because buffersize is 0 and chan sz is 1
+	mockMetricsReporter.EXPECT().ReportHistogram(metrics.ChannelCapacity, gomock.Any(), float64(-1))
 	ag.reportChannelSize()
 }
 

pkg/cluster/nats_rpc_server.go

@@ -408,16 +408,15 @@ func (ns *NatsRPCServer) reportMetrics() {
 			if subChanCapacity == 0 {
 				logger.Log.Warn("subChan is at maximum capacity")
 			}
-			if err := mr.ReportGauge(metrics.ChannelCapacity, map[string]string{"channel": "rpc_server_subchan"}, float64(subChanCapacity)); err != nil {
+			if err := mr.ReportHistogram(metrics.ChannelCapacity, map[string]string{"channel": "rpc_server_subchan"}, float64(subChanCapacity)); err != nil {
 				logger.Log.Warnf("failed to report subChan queue capacity: %s", err.Error())
 			}
-
 			// bindingschan
 			bindingsChanCapacity := ns.messagesBufferSize - len(ns.bindingsChan)
 			if bindingsChanCapacity == 0 {
 				logger.Log.Warn("bindingsChan is at maximum capacity")
 			}
-			if err := mr.ReportGauge(metrics.ChannelCapacity, map[string]string{"channel": "rpc_server_bindingschan"}, float64(bindingsChanCapacity)); err != nil {
+			if err := mr.ReportHistogram(metrics.ChannelCapacity, map[string]string{"channel": "rpc_server_bindingschan"}, float64(bindingsChanCapacity)); err != nil {
 				logger.Log.Warnf("failed to report bindingsChan capacity: %s", err.Error())
 			}
 
@@ -426,7 +425,7 @@ func (ns *NatsRPCServer) reportMetrics() {
 			if userPushChanCapacity == 0 {
 				logger.Log.Warn("userPushChan is at maximum capacity")
 			}
-			if err := mr.ReportGauge(metrics.ChannelCapacity, map[string]string{"channel": "rpc_server_userpushchan"}, float64(userPushChanCapacity)); err != nil {
+			if err := mr.ReportHistogram(metrics.ChannelCapacity, map[string]string{"channel": "rpc_server_userpushchan"}, float64(userPushChanCapacity)); err != nil {
 				logger.Log.Warnf("failed to report userPushCh capacity: %s", err.Error())
 			}
 		}

pkg/cluster/nats_rpc_server_test.go

@@ -316,7 +316,7 @@ func TestNatsRPCServerHandleMessages(t *testing.T) {
 			assert.NoError(t, err)
 
 			mockMetricsReporter.EXPECT().ReportGauge(metrics.DroppedMessages, gomock.Any(), float64(0))
-			mockMetricsReporter.EXPECT().ReportGauge(metrics.ChannelCapacity, gomock.Any(), gomock.Any()).Times(3)
+			mockMetricsReporter.EXPECT().ReportHistogram(metrics.ChannelCapacity, gomock.Any(), gomock.Any()).Times(3)
 
 			conn.Publish(table.topic, b)
 			r := helpers.ShouldEventuallyReceive(t, rpcServer.unhandledReqCh).(*protos.Request)
@@ -505,6 +505,6 @@ func TestNatsRPCServerReportMetrics(t *testing.T) {
 	rpcServer.userPushCh <- &protos.Push{}
 
 	mockMetricsReporter.EXPECT().ReportGauge(metrics.DroppedMessages, gomock.Any(), float64(rpcServer.dropped))
-	mockMetricsReporter.EXPECT().ReportGauge(metrics.ChannelCapacity, gomock.Any(), float64(99)).Times(3)
+	mockMetricsReporter.EXPECT().ReportHistogram(metrics.ChannelCapacity, gomock.Any(), float64(99)).Times(3)
 	rpcServer.reportMetrics()
 }

pkg/metrics/constants.go

@@ -7,7 +7,7 @@ var (
 	ConnectedClients = "connected_clients"
 	// CountServers counts the number of servers of different types
 	CountServers = "count_servers"
-	// ChannelCapacity represents the capacity of a channel (available slots)
+	// ChannelCapacity represents the capacity of a channel as a histogram (distribution of available slots)
 	ChannelCapacity = "channel_capacity"
 	// DroppedMessages reports the number of dropped messages in rpc server (messages that will not be handled)
 	DroppedMessages = "dropped_messages"

pkg/metrics/prometheus_reporter.go

@@ -188,12 +188,13 @@ func (p *PrometheusReporter) registerMetrics(
 		append([]string{"type"}, additionalLabelsKeys...),
 	)
 
-	p.gaugeReportersMap[ChannelCapacity] = prometheus.NewGaugeVec(
-		prometheus.GaugeOpts{
+	p.histogramReportersMap[ChannelCapacity] = prometheus.NewHistogramVec(
+		prometheus.HistogramOpts{
 			Namespace:   "pitaya",
 			Subsystem:   "channel",
 			Name:        ChannelCapacity,
 			Help:        "the available capacity of the channel",
+			Buckets:     []float64{0, 1, 10, 50, 100, 250, 500, 750, 1000, 1500, 2000, 3000, 4000, 5000},
 			ConstLabels: constLabels,
 		},
 		append([]string{"channel"}, additionalLabelsKeys...),