Add rps debug metric to xdr server (#214)

io/aerospike/xdr/metrics.go

@@ -0,0 +1,69 @@
+// Copyright 2024 Aerospike, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package xdr
+
+import (
+	"context"
+	"log/slog"
+	"sync/atomic"
+	"time"
+)
+
+type metricsCollector struct {
+	ctx context.Context
+
+	requestCount atomic.Uint64
+	increment    func()
+	lastTime     time.Time
+
+	logger *slog.Logger
+}
+
+func mewMetricsCollector(ctx context.Context, logger *slog.Logger) *metricsCollector {
+	mc := &metricsCollector{
+		ctx:       ctx,
+		increment: func() {},
+		lastTime:  time.Now(),
+		logger:    logger,
+	}
+
+	// enable only for logger debug level
+	if mc.logger.Enabled(mc.ctx, slog.LevelDebug) {
+		mc.increment = func() { mc.requestCount.Add(1) }
+		go mc.reportMetrics()
+	}
+
+	return mc
+}
+
+func (mc *metricsCollector) reportMetrics() {
+	ticker := time.NewTicker(time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case t := <-ticker.C:
+			count := mc.requestCount.Swap(0)
+			elapsed := t.Sub(mc.lastTime).Seconds()
+			rps := float64(count) / elapsed
+
+			mc.logger.Debug("metrics", slog.Float64("rps", rps))
+
+			mc.lastTime = t
+		case <-mc.ctx.Done():
+			return
+		}
+	}
+}

io/aerospike/xdr/tcp_server.go

@@ -176,6 +176,8 @@ func (s *TCPServer) GetActiveConnections() int32 {
 // acceptConnections serves connections, not more than maxConnections.
 // All connections over pool will be rejected.
 func (s *TCPServer) acceptConnections(ctx context.Context) {
+	metrics := mewMetricsCollector(ctx, s.logger)
+
 	for {
 		select {
 		case <-ctx.Done():
@@ -203,6 +205,7 @@ func (s *TCPServer) acceptConnections(ctx context.Context) {
 						s.config.ReadTimeoutMilliseconds,
 						s.config.WriteTimeoutMilliseconds,
 						s.logger,
+						metrics,
 					)
 					// Handlers wait when all goroutines are finished.
 					handler.Start(ctx)
@@ -247,7 +250,8 @@ type ConnectionHandler struct {
 	ackMsgRetry   []byte
 	timeNow       int64
 
-	logger *slog.Logger
+	logger  *slog.Logger
+	metrics *metricsCollector
 }
 
 // NewConnectionHandler returns new connection handler.
@@ -259,6 +263,7 @@ func NewConnectionHandler(
 	readTimeout int64,
 	writeTimeout int64,
 	logger *slog.Logger,
+	metrics *metricsCollector,
 ) *ConnectionHandler {
 	return &ConnectionHandler{
 		conn:             conn,
@@ -271,6 +276,7 @@ func NewConnectionHandler(
 		ackMsgSuccess:    NewAckMessage(AckOK),
 		ackMsgRetry:      NewAckMessage(AckRetry),
 		logger:           logger,
+		metrics:          metrics,
 	}
 }
 
@@ -362,6 +368,8 @@ func (h *ConnectionHandler) handleMessages(ctx context.Context) {
 				return
 			}
 
+			h.metrics.increment()
+
 			// Process message asynchronously
 			h.bodyQueue <- message
 		}