Rgw ops log latency metrics fix (#29)

* latency metrics logic fixed

On-behalf-of: SAP <filipp.akinfiev@clyso.com>
Signed-off-by: Filipp Akinfiev <filipp.akinfiev@clyso.com>

* readme updated

On-behalf-of: SAP <filipp.akinfiev@clyso.com>
Signed-off-by: Filipp Akinfiev <filipp.akinfiev@clyso.com>

---------

Signed-off-by: Filipp Akinfiev <filipp.akinfiev@clyso.com>

Author: fa-at-pulsit

pkg/producers/opslog/metrics.go

@@ -17,7 +17,15 @@ type Metrics struct {
 	BytesReceived atomic.Uint64
 	Errors        atomic.Uint64
 
-	LatencyByMethod sync.Map // map["user|bucket|method"]
+	// LatencyByMethod sync.Map // map["user|bucket|method"]
+
+	// LatencyObs records a single request‐latency observation into the
+	// `requestsDurationHistogram`, which is registered once at startup.
+	// Because the histogram lives for the entire process life and is never
+	// re‐initialized or cleared, its buckets continuously accumulate across
+	// scrape intervals—ensuring a true cumulative histogram that Prometheus
+	// can derive accurate rates and quantiles from.
+	LatencyObs func(user string, tenant string, bucket string, method string, seconds float64)
 
 	RequestsByMethod     sync.Map // map[string]*atomic.Uint64
 	RequestsByOperation  sync.Map // map[string]*atomic.Uint64
@@ -39,8 +47,17 @@ type Metrics struct {
 
 }
 
-func NewMetrics() *Metrics {
-	return &Metrics{}
+func NewMetrics(obs ...func(user string, tenant string, bucket string, method string, seconds float64)) *Metrics {
+	var cb func(user, tenant, bucket, method string, seconds float64)
+	if len(obs) > 0 {
+		cb = obs[0]
+	} else {
+		// default no‐op so nobody ever has a nil-pointer
+		cb = func(_, _, _, _ string, _ float64) {}
+	}
+	return &Metrics{
+		LatencyObs: cb,
+	}
 }
 
 // Convert metrics to a JSON-friendly struct
@@ -115,11 +132,6 @@ func (m *Metrics) ToJSON(metricsConfig *MetricsConfig) ([]byte, error) {
 		data["errors_by_ip_and_bucket"] = loadSyncMap(&m.ErrorsByIPAndBucket)
 	}
 
-	// Latency Tracking
-	if metricsConfig.TrackLatencyByMethod {
-		data["latency_by_method"] = loadSyncMap(&m.LatencyByMethod)
-	}
-
 	return json.Marshal(data)
 }
 
@@ -223,16 +235,15 @@ func (m *Metrics) Update(logEntry S3OperationLog, metricsConfig *MetricsConfig)
 
 	// Latency Tracking
 	if logEntry.TotalTime > 0 {
-		latencyMs := uint64(logEntry.TotalTime)
-
 		if metricsConfig.TrackLatencyByMethod ||
 			metricsConfig.TrackLatencyByBucket ||
 			metricsConfig.TrackLatencyByTenant ||
 			metricsConfig.TrackLatencyByUser ||
 			metricsConfig.TrackLatencyByBucketAndMethod {
-			// Key format: "user|bucket|method"
-			latencyKey := logEntry.User + "|" + logEntry.Bucket + "|" + method
-			incrementSyncMapValue(&m.LatencyByMethod, latencyKey, latencyMs)
+
+			latencySec := float64(logEntry.TotalTime) / 1000.0
+			userStr, tenantStr := extractUserAndTenant(logEntry.User)
+			m.LatencyObs(userStr, tenantStr, logEntry.Bucket, method, latencySec)
 		}
 	}
 }
@@ -244,7 +255,6 @@ func (m *Metrics) Reset() {
 	m.BytesReceived.Store(0)
 	m.Errors.Store(0)
 
-	resetSyncMap(&m.LatencyByMethod)
 	resetSyncMap(&m.RequestsByMethod)
 	resetSyncMap(&m.RequestsByOperation)
 	resetSyncMap(&m.RequestsByStatusCode)
@@ -263,10 +273,6 @@ func (m *Metrics) Reset() {
 	resetSyncMap(&m.ErrorsByIPAndBucket)
 }
 
-func (m *Metrics) ResetPerWindowMetrics() {
-	resetSyncMap(&m.LatencyByMethod)
-}
-
 // Helper function: Update max atomic value
 func updateMaxAtomic(target *atomic.Uint64, value uint64) {
 	for {
@@ -398,14 +404,13 @@ func ExtractHTTPMethod(uri string) string {
 
 // Clone creates a deep copy of the Metrics
 func (m *Metrics) Clone() *Metrics {
-	clone := NewMetrics()
+	clone := NewMetrics(m.LatencyObs)
 
 	clone.TotalRequests.Store(m.TotalRequests.Load())
 	clone.BytesSent.Store(m.BytesSent.Load())
 	clone.BytesReceived.Store(m.BytesReceived.Load())
 	clone.Errors.Store(m.Errors.Load())
 
-	copySyncMap(&m.LatencyByMethod, &clone.LatencyByMethod)
 	copySyncMap(&m.RequestsByMethod, &clone.RequestsByMethod)
 	copySyncMap(&m.RequestsByOperation, &clone.RequestsByOperation)
 	copySyncMap(&m.RequestsByStatusCode, &clone.RequestsByStatusCode)
@@ -440,7 +445,7 @@ func copySyncMap(src, dst *sync.Map) {
 
 // SubtractMetrics calculates the delta between two metrics objects: total - previous
 func SubtractMetrics(total, previous *Metrics) *Metrics {
-	delta := NewMetrics()
+	delta := NewMetrics(total.LatencyObs)
 
 	// Handle top-level counters
 	delta.TotalRequests.Store(diff(total.TotalRequests.Load(), previous.TotalRequests.Load()))
@@ -465,7 +470,6 @@ func SubtractMetrics(total, previous *Metrics) *Metrics {
 	subtractSyncMap(&total.BytesReceivedByIP, &previous.BytesReceivedByIP, &delta.BytesReceivedByIP)
 	subtractSyncMap(&total.ErrorsByUserAndBucket, &previous.ErrorsByUserAndBucket, &delta.ErrorsByUserAndBucket)
 	subtractSyncMap(&total.ErrorsByIPAndBucket, &previous.ErrorsByIPAndBucket, &delta.ErrorsByIPAndBucket)
-	subtractSyncMap(&total.LatencyByMethod, &previous.LatencyByMethod, &delta.LatencyByMethod)
 
 	return delta
 }

pkg/producers/opslog/metrics_test.go

@@ -92,6 +92,37 @@ func TestSubtractMetrics_MissingInPrev(t *testing.T) {
 	assert.Equal(t, uint64(7), v.(*atomic.Uint64).Load())
 }
 
+func TestLatencyObsPropagation(t *testing.T) {
+	called := false
+	cb := func(u, tnt, bucket, method string, sec float64) {
+		assert.Equal(t, "u1", u)
+		assert.Equal(t, "t1", tnt)
+		assert.Equal(t, "b1", bucket)
+		assert.Equal(t, "M", method)
+		assert.InDelta(t, 0.123, sec, 1e-6)
+		called = true
+	}
+
+	m := NewMetrics(cb)
+	// direct call
+	m.LatencyObs("u1", "t1", "b1", "M", 0.123)
+	assert.True(t, called)
+
+	// clone carries it forward
+	clone := m.Clone()
+	called = false
+	clone.LatencyObs("u1", "t1", "b1", "M", 0.123)
+	assert.True(t, called)
+
+	// subtract carries it forward
+	total := NewMetrics(cb)
+	prev := NewMetrics(cb)
+	delta := SubtractMetrics(total, prev)
+	called = false
+	delta.LatencyObs("u1", "t1", "b1", "M", 0.123)
+	assert.True(t, called)
+}
+
 func newUint64(val uint64) *atomic.Uint64 {
 	var u atomic.Uint64
 	u.Store(val)

pkg/producers/opslog/opslog.go

@@ -74,10 +74,42 @@ func StartFileOpsLogger(cfg OpsLogConfig) {
 
 	if cfg.Prometheus {
 		StartPrometheusServer(cfg.PrometheusPort, &cfg.MetricsConfig)
+
+		if cfg.MetricsConfig.TrackLatencyByMethod || cfg.MetricsConfig.TrackLatencyByUser ||
+			cfg.MetricsConfig.TrackLatencyByBucket || cfg.MetricsConfig.TrackLatencyByTenant ||
+			cfg.MetricsConfig.TrackLatencyByBucketAndMethod {
+			latencyObs = func(user, tenant, bucket, method string, seconds float64) {
+				// raw per-method
+				if cfg.MetricsConfig.TrackLatencyByBucketAndMethod {
+					requestsDurationHistogram.WithLabelValues(user, tenant, bucket, method).
+						Observe(seconds)
+				}
+				// per-method aggregated across users
+				if cfg.MetricsConfig.TrackLatencyByMethod {
+					requestsDurationHistogram.WithLabelValues("all", "all", bucket, method).
+						Observe(seconds)
+				}
+				// per-bucket
+				if cfg.MetricsConfig.TrackLatencyByBucket {
+					requestsDurationHistogram.WithLabelValues("all", "all", bucket, "all").
+						Observe(seconds)
+				}
+				// per-tenant
+				if cfg.MetricsConfig.TrackLatencyByTenant {
+					requestsDurationHistogram.WithLabelValues("all", tenant, "all", "all").
+						Observe(seconds)
+				}
+				// per-user
+				if cfg.MetricsConfig.TrackLatencyByUser {
+					requestsDurationHistogram.WithLabelValues(user, tenant, "all", "all").
+						Observe(seconds)
+				}
+			}
+		}
 	}
 
 	// Initialize metrics
-	metrics := NewMetrics()
+	metrics := NewMetrics(latencyObs)
 	interval := time.Duration(cfg.PrometheusIntervalSeconds) * time.Second
 	ticker := time.NewTicker(interval)
 	defer ticker.Stop()
@@ -106,8 +138,6 @@ func StartFileOpsLogger(cfg OpsLogConfig) {
 		if cfg.UseNats {
 			publishMetricsToNATS(cfg, nc, metrics)
 		}
-
-		metrics.ResetPerWindowMetrics()
 	}
 
 	// Keep the program running
@@ -297,7 +327,7 @@ func StartSocketOpsLogger(cfg OpsLogConfig) {
 		log.Info().Str("nats_url", cfg.NatsURL).Msg("Connected to NATS server")
 	}
 
-	metrics := NewMetrics()
+	metrics := NewMetrics(latencyObs)
 	ticker := time.NewTicker(1 * time.Minute) // Set up a ticker to trigger every 1 minute
 	defer ticker.Stop()
 
@@ -348,7 +378,7 @@ func StartSocketOpsLogger(cfg OpsLogConfig) {
 		}
 
 		// Reset metrics for the next interval
-		metrics = NewMetrics()
+		metrics = NewMetrics(latencyObs)
 	}
 }
 

pkg/producers/opslog/prometheus.go

@@ -140,6 +140,8 @@ var (
 		[]string{"user", "tenant", "bucket", "method"},
 		// []string{"pod", "user", "tenant", "bucket", "method"},
 	)
+
+	latencyObs func(user, tenant, bucket, method string, seconds float64)
 )
 
 func initPrometheusSettings(metricsConfig *MetricsConfig) {
@@ -696,100 +698,6 @@ func PublishToPrometheus(totalMetrics *Metrics, cfg OpsLogConfig) {
 		})
 	}
 
-	// Update request duration histogram (latency metrics)
-	if metricsConfig.TrackLatencyByMethod || metricsConfig.TrackLatencyByUser ||
-		metricsConfig.TrackLatencyByBucket || metricsConfig.TrackLatencyByTenant ||
-		metricsConfig.TrackLatencyByBucketAndMethod {
-
-		diffMetrics.LatencyByMethod.Range(func(key, totalLatency any) bool {
-			parts := strings.Split(key.(string), "|")
-			if len(parts) != 3 {
-				log.Warn().Msgf("Invalid key format in LatencyByMethod: %v", key)
-				return true
-			}
-			user, bucket, method := parts[0], parts[1], parts[2]
-			userStr, tenantStr := extractUserAndTenant(user)
-
-			// Fetch request count for this method
-			countVal, exists := diffMetrics.RequestsByMethod.Load(key)
-			if !exists {
-				countVal, exists = currentMetrics.RequestsByMethod.Load(key)
-				if !exists {
-					log.Warn().Msgf("Missing request count for latency key: %v", key)
-					return true
-				}
-			}
-			count := float64(countVal.(*atomic.Uint64).Load())
-			if count == 0 {
-				log.Warn().Msgf("Zero request count for latency key: %v", key)
-				return true
-			}
-
-			// Compute avg latency (convert ms → sec)
-			avgLatencySec := float64(totalLatency.(*atomic.Uint64).Load()) / count / 1000.0
-
-			if metricsConfig.TrackLatencyByBucketAndMethod {
-				// Fine-grained latency tracking per bucket & method
-				requestsDurationHistogram.With(prometheus.Labels{
-					// "pod":    cfg.PodName,
-					"user":   userStr,
-					"tenant": tenantStr,
-					"bucket": bucket,
-					"method": method,
-				}).Observe(avgLatencySec)
-			}
-
-			if metricsConfig.TrackLatencyByMethod {
-				// Aggregated latency for all users (reduces cardinality)
-				requestsDurationHistogram.With(prometheus.Labels{
-					// "pod":    cfg.PodName,
-					"user":   "all",
-					"tenant": "all",
-					"bucket": bucket,
-					"method": method,
-				}).Observe(avgLatencySec)
-			}
-
-			if metricsConfig.TrackLatencyByBucket {
-				requestsDurationHistogram.With(prometheus.Labels{
-					"user":   "all",
-					"tenant": "all",
-					"bucket": bucket,
-					"method": "all",
-				}).Observe(avgLatencySec)
-			}
-
-			if metricsConfig.TrackLatencyByTenant {
-				requestsDurationHistogram.With(prometheus.Labels{
-					"user":   "all",
-					"tenant": tenantStr,
-					"bucket": "all",
-					"method": "all",
-				}).Observe(avgLatencySec)
-			}
-
-			if metricsConfig.TrackLatencyByUser {
-				requestsDurationHistogram.With(prometheus.Labels{
-					"user":   userStr,
-					"tenant": tenantStr,
-					"bucket": "all",
-					"method": "all",
-				}).Observe(avgLatencySec)
-			}
-
-			// // Aggregate latency for all methods
-			// requestsDurationHistogram.With(prometheus.Labels{
-			// 	"pod":    cfg.PodName,
-			// 	"user":   userStr,
-			// 	"tenant": tenantStr,
-			// 	"bucket": bucket,
-			// 	"method": "all",
-			// }).Observe(avgLatencySec)
-
-			return true
-		})
-	}
-
 	log.Info().Msg("Updated Prometheus metrics for users and buckets")
 }
 

pkg/producers/radosgwusage/README.md

@@ -60,17 +60,16 @@ Configuration can also be set through environment variables:
 The RadosGW Usage Exporter collects and exposes the following metrics:
 
 
-### Bucket Usage Metrics
+### Bucket / User Usage Metrics
 
-- `radosgw_usage_bucket_bytes`: Bucket used bytes.
-- `radosgw_usage_bucket_utilized_bytes`: Bucket utilized bytes.
-- `radosgw_usage_bucket_objects`: Number of objects in the bucket.
+- `radosgw_user_buckets_total`: Total number of buckets for each user.
+- `radosgw_user_objects_total`: Total number of objects for each user.
+- `radosgw_user_data_size_bytes`: Total size of data for each user in bytes
 
 ### Quota Metrics
 
 - `radosgw_usage_bucket_quota_enabled`: Indicates if quota is enabled for the bucket.
 - `radosgw_usage_bucket_quota_size`: Maximum allowed bucket size.
-- `radosgw_usage_bucket_quota_size_bytes`: Maximum allowed bucket size in bytes.
 - `radosgw_usage_bucket_quota_size_objects`: Maximum allowed number of objects in the bucket.
 - `radosgw_usage_user_quota_enabled`: Indicates if user quota is enabled.
 - `radosgw_usage_user_quota_size`: Maximum allowed size for the user.
@@ -81,9 +80,6 @@ The RadosGW Usage Exporter collects and exposes the following metrics:
 - `radosgw_usage_bucket_shards`: Number of shards in the bucket.
 - `radosgw_user_metadata`: User metadata (e.g., display name, email, storage class).
 
-### Miscellaneous Metrics
-
-- `radosgw_usage_scrape_duration_seconds`: Amount of time each scrape takes.
 
 ## Example Workflow