rps-metric

io/aerospike/xdr/metrics.go

@@ -0,0 +1,57 @@
+// 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 (
+	"log/slog"
+	"sync/atomic"
+	"time"
+)
+
+type MetricsCollector struct {
+	requestCount uint64
+	lastTime     time.Time
+	logger       *slog.Logger
+}
+
+func NewMetricsCollector(logger *slog.Logger) *MetricsCollector {
+	mc := &MetricsCollector{
+		lastTime: time.Now(),
+		logger:   logger,
+	}
+	go mc.reportMetrics()
+
+	return mc
+}
+
+func (mc *MetricsCollector) IncrementRequests() {
+	atomic.AddUint64(&mc.requestCount, 1)
+}
+
+func (mc *MetricsCollector) reportMetrics() {
+	ticker := time.NewTicker(time.Second)
+	defer ticker.Stop()
+
+	for range ticker.C {
+		now := time.Now()
+		count := atomic.SwapUint64(&mc.requestCount, 0)
+		elapsed := now.Sub(mc.lastTime).Seconds()
+		rps := float64(count) / elapsed
+
+		mc.logger.Debug("metrics", slog.Float64("rps", rps))
+
+		mc.lastTime = now
+	}
+}

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 := NewMetricsCollector(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.IncrementRequests()
+
 			// Process message asynchronously
 			h.bodyQueue <- message
 		}