Skip to content

Commit e1543c8

Browse files
fa-at-pulsitfa-at-pulsit
authored
bug fixes and improvements (#25)
On-behalf-of: SAP <filipp.akinfiev@clyso.com> Signed-off-by: Filipp Akinfiev <filipp.akinfiev@clyso.com>
1 parent dfbaf00 commit e1543c8

5 files changed

Lines changed: 79 additions & 16 deletions

File tree

‎pkg/commands/producer_ops_log.go‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@ var (
4444
opsTrackErrorsByStatus bool
4545
opsTrackLatencyByUser bool
4646
opsTrackLatencyByBucket bool
47+
opsTrackLatencyByMethod bool
4748
opsTrackLatencyByTenant bool
4849
)
4950

@@ -97,6 +98,7 @@ Following this configuration change, the RadosGW will log operations to the file
9798
TrackErrorsByStatus: opsTrackErrorsByStatus,
9899
TrackLatencyByUser: opsTrackLatencyByUser,
99100
TrackLatencyByBucket: opsTrackLatencyByBucket,
101+
TrackLatencyByMethod: opsTrackLatencyByMethod,
100102
TrackLatencyByTenant: opsTrackLatencyByTenant,
101103
},
102104
}
@@ -150,6 +152,7 @@ Following this configuration change, the RadosGW will log operations to the file
150152
event.Bool("track_errors_by_status", config.MetricsConfig.TrackErrorsByStatus)
151153
event.Bool("track_latency_by_user", config.MetricsConfig.TrackLatencyByUser)
152154
event.Bool("track_latency_by_bucket", config.MetricsConfig.TrackLatencyByBucket)
155+
event.Bool("track_latency_by_method", config.MetricsConfig.TrackLatencyByMethod)
153156
event.Bool("track_latency_by_tenant", config.MetricsConfig.TrackLatencyByTenant)
154157

155158
event.Msg("OpsLog configuration initialized")
@@ -196,6 +199,7 @@ func mergeOpsLogConfigWithEnv(cfg opslog.OpsLogConfig) opslog.OpsLogConfig {
196199
cfg.MetricsConfig.TrackLatencyByUser = getEnvBool("TRACK_LATENCY_BY_USER", cfg.MetricsConfig.TrackLatencyByUser)
197200
cfg.MetricsConfig.TrackLatencyByBucket = getEnvBool("TRACK_LATENCY_BY_BUCKET", cfg.MetricsConfig.TrackLatencyByBucket)
198201
cfg.MetricsConfig.TrackLatencyByTenant = getEnvBool("TRACK_LATENCY_BY_TENANT", cfg.MetricsConfig.TrackLatencyByTenant)
202+
cfg.MetricsConfig.TrackLatencyByMethod = getEnvBool("TRACK_LATENCY_BY_METHOD", cfg.MetricsConfig.TrackLatencyByMethod)
199203

200204
return cfg
201205
}
@@ -229,6 +233,7 @@ func init() {
229233
opsLogCmd.Flags().BoolVar(&opsTrackRequestsByTenant, "track-requests-by-tenant", false, "Track requests by tenant")
230234
opsLogCmd.Flags().BoolVar(&opsTrackErrorsByBucket, "track-errors-by-bucket", false, "Track errors per bucket")
231235
opsLogCmd.Flags().BoolVar(&opsTrackErrorsByStatus, "track-errors-by-status", false, "Track errors per HTTP status")
236+
opsLogCmd.Flags().BoolVar(&opsTrackLatencyByMethod, "track-latency-by-method", false, "Track latency per method")
232237
opsLogCmd.Flags().BoolVar(&opsTrackLatencyByUser, "track-latency-by-user", false, "Track latency per user")
233238
opsLogCmd.Flags().BoolVar(&opsTrackLatencyByBucket, "track-latency-by-bucket", false, "Track latency per bucket")
234239
opsLogCmd.Flags().BoolVar(&opsTrackLatencyByTenant, "track-latency-by-tenant", false, "Track latency per tenant")

‎pkg/producers/opslog/README.md‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -72,6 +72,7 @@ prysm local-producer ops-log [flags]
7272
| `TRACK_REQUESTS_BY_BUCKET` | Track requests per bucket. |
7373
| `TRACK_REQUESTS_BY_USER` | Track requests per user. |
7474
| `TRACK_REQUESTS_BY_TENANT` | Track requests per tenant. |
75+
| `TRACK_REQUESTS_BY_IP_BUCKET_METHOD_TENANT` | Track requests per IP, bucket, HTTP method and tenant. |
7576
| `TRACK_LATENCY_BY_USER` | Track latency per user. |
7677
| `TRACK_LATENCY_BY_BUCKET` | Track latency per bucket. |
7778
| `TRACK_LATENCY_BY_TENANT` | Track latency per tenant. |
@@ -90,7 +91,7 @@ prysm local-producer ops-log [flags]
9091
| `radosgw_http_errors_by_user` | Counter | `pod`, `user`, `tenant`, `bucket`, `http_status` | HTTP errors grouped by user, tenant, bucket, and status code. |
9192
| `radosgw_http_errors_by_ip` | Counter | `pod`, `bucket`, `ip`, `http_status` | HTTP errors grouped by IP, bucket, and status code. |
9293
| `radosgw_requests_by_ip` | Gauge | `pod`, `user`, `tenant`, `ip` | Total number of requests grouped by IP and user. |
93-
| `radosgw_requests_by_ip_bucket_method_tenant` | Gauge | `pod`, `bucket`, `tenant`, `ip`, `method` | Total number of requests grouped by IP, bucket and method. |
94+
| `radosgw_requests_by_ip_bucket_method_tenant` | Gauge | `pod`, `ip`, `bucket`, `method`, `tenant` | Total number of requests grouped by IP, bucket and method. |
9495
| `radosgw_bytes_sent_by_ip` | Gauge | `pod`, `user`, `tenant`, `ip` | Total bytes sent grouped by IP and user. |
9596
| `radosgw_bytes_received_by_ip` | Gauge | `pod`, `user`, `tenant`, `ip` | Total bytes received grouped by IP and user. |
9697
| `radosgw_requests_duration` | Histogram | `user`, `tenant`, `bucket`, `method` | Histogram of request latencies (in seconds). |

‎pkg/producers/opslog/metrics.go‎

Lines changed: 7 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -225,7 +225,11 @@ func (m *Metrics) Update(logEntry S3OperationLog, metricsConfig *MetricsConfig)
225225
if logEntry.TotalTime > 0 {
226226
latencyMs := uint64(logEntry.TotalTime)
227227

228-
if metricsConfig.TrackLatencyByMethod {
228+
if metricsConfig.TrackLatencyByMethod ||
229+
metricsConfig.TrackLatencyByBucket ||
230+
metricsConfig.TrackLatencyByTenant ||
231+
metricsConfig.TrackLatencyByUser ||
232+
metricsConfig.TrackLatencyByBucketAndMethod {
229233
// Key format: "user|bucket|method"
230234
latencyKey := logEntry.User + "|" + logEntry.Bucket + "|" + method
231235
incrementSyncMapValue(&m.LatencyByMethod, latencyKey, latencyMs)
@@ -456,11 +460,12 @@ func SubtractMetrics(total, previous *Metrics) *Metrics {
456460
subtractSyncMap(&total.BytesSentByBucket, &previous.BytesSentByBucket, &delta.BytesSentByBucket)
457461
subtractSyncMap(&total.BytesReceivedByBucket, &previous.BytesReceivedByBucket, &delta.BytesReceivedByBucket)
458462
subtractSyncMap(&total.RequestsByIP, &previous.RequestsByIP, &delta.RequestsByIP)
459-
subtractSyncMap(&delta.RequestsByIPBucketMethodTenant, &total.RequestsByIPBucketMethodTenant, &previous.RequestsByIPBucketMethodTenant)
463+
subtractSyncMap(&total.RequestsByIPBucketMethodTenant, &previous.RequestsByIPBucketMethodTenant, &delta.RequestsByIPBucketMethodTenant)
460464
subtractSyncMap(&total.BytesSentByIP, &previous.BytesSentByIP, &delta.BytesSentByIP)
461465
subtractSyncMap(&total.BytesReceivedByIP, &previous.BytesReceivedByIP, &delta.BytesReceivedByIP)
462466
subtractSyncMap(&total.ErrorsByUserAndBucket, &previous.ErrorsByUserAndBucket, &delta.ErrorsByUserAndBucket)
463467
subtractSyncMap(&total.ErrorsByIPAndBucket, &previous.ErrorsByIPAndBucket, &delta.ErrorsByIPAndBucket)
468+
subtractSyncMap(&total.LatencyByMethod, &previous.LatencyByMethod, &delta.LatencyByMethod)
464469

465470
return delta
466471
}

‎pkg/producers/opslog/metrics_test.go‎

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -67,6 +67,31 @@ func TestCloneMetrics(t *testing.T) {
6767
assert.Equal(t, uint64(5), v1After.(*atomic.Uint64).Load(), "Clone should remain unchanged")
6868
}
6969

70+
func TestSubtractMetrics_ZeroDelta(t *testing.T) {
71+
total := NewMetrics()
72+
prev := NewMetrics()
73+
74+
total.RequestsByUser.Store("x", newUint64(5))
75+
prev.RequestsByUser.Store("x", newUint64(5))
76+
77+
delta := SubtractMetrics(total, prev)
78+
_, ok := delta.RequestsByUser.Load("x")
79+
assert.False(t, ok)
80+
}
81+
82+
func TestSubtractMetrics_MissingInPrev(t *testing.T) {
83+
total := NewMetrics()
84+
prev := NewMetrics()
85+
86+
total.RequestsByUser.Store("new|key", newUint64(7))
87+
88+
delta := SubtractMetrics(total, prev)
89+
90+
v, ok := delta.RequestsByUser.Load("new|key")
91+
assert.True(t, ok)
92+
assert.Equal(t, uint64(7), v.(*atomic.Uint64).Load())
93+
}
94+
7095
func newUint64(val uint64) *atomic.Uint64 {
7196
var u atomic.Uint64
7297
u.Store(val)

‎pkg/producers/opslog/prometheus.go‎

Lines changed: 40 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -194,7 +194,13 @@ func initPrometheusSettings(metricsConfig *MetricsConfig) {
194194
prometheus.MustRegister(httpErrorsByIPCounter)
195195
}
196196

197-
prometheus.MustRegister(requestsDurationHistogram)
197+
if metricsConfig.TrackLatencyByMethod ||
198+
metricsConfig.TrackLatencyByBucket ||
199+
metricsConfig.TrackLatencyByTenant ||
200+
metricsConfig.TrackLatencyByUser ||
201+
metricsConfig.TrackLatencyByBucketAndMethod {
202+
prometheus.MustRegister(requestsDurationHistogram)
203+
}
198204
}
199205

200206
// PublishToPrometheus updates Prometheus metrics from aggregated data
@@ -207,11 +213,14 @@ func PublishToPrometheus(totalMetrics *Metrics, cfg OpsLogConfig) {
207213
return
208214
}
209215

210-
// Compute diff (delta) between total and previous
211-
diffMetrics := SubtractMetrics(totalMetrics, previousMetrics)
216+
// Snapshot current total
217+
currentMetrics := totalMetrics.Clone()
218+
219+
// Compute diff (delta) against last snapshot
220+
diffMetrics := SubtractMetrics(currentMetrics, previousMetrics)
212221

213-
// Store the current total as the new previous
214-
previousMetrics = totalMetrics.Clone()
222+
// Update snapshot for next interval
223+
previousMetrics = currentMetrics
215224

216225
if metricsConfig.TrackRequestsByUser {
217226
// Total requests grouped by user, bucket, method, and status
@@ -483,7 +492,7 @@ func PublishToPrometheus(totalMetrics *Metrics, cfg OpsLogConfig) {
483492

484493
if metricsConfig.TrackRequestsByIP {
485494
// Publish requests per IP & User
486-
totalMetrics.RequestsByIP.Range(func(key, count any) bool {
495+
currentMetrics.RequestsByIP.Range(func(key, count any) bool {
487496
keyStr := key.(string)
488497
parts := strings.Split(keyStr, "|")
489498
user, ip := parts[0], parts[1]
@@ -503,22 +512,23 @@ func PublishToPrometheus(totalMetrics *Metrics, cfg OpsLogConfig) {
503512
}
504513

505514
if metricsConfig.TrackRequestsByIPBucketMethodTenant {
506-
totalMetrics.RequestsByIPBucketMethodTenant.Range(func(key, count any) bool {
515+
currentMetrics.RequestsByIPBucketMethodTenant.Range(func(key, count any) bool {
507516
parts := strings.Split(key.(string), "|")
508517
if len(parts) != 4 {
509518
log.Warn().Msgf("Invalid key format in RequestsByIPBucketMethodTenant: %v", key)
510519
return true
511520
}
512521

513-
ip, bucket, method, tenant := parts[0], parts[1], parts[2], parts[3]
522+
ip, bucket, method, user := parts[0], parts[1], parts[2], parts[3]
523+
_, tenantStr := extractUserAndTenant(user)
514524

515525
if atomicPtr, ok := count.(*atomic.Uint64); ok {
516526
requestsByIPBucketMethodTenantGauge.With(prometheus.Labels{
517527
"pod": cfg.PodName,
518528
"ip": ip,
519529
"bucket": bucket,
520530
"method": method,
521-
"tenant": tenant,
531+
"tenant": tenantStr,
522532
}).Set(float64(atomicPtr.Load()))
523533
}
524534
return true
@@ -527,7 +537,7 @@ func PublishToPrometheus(totalMetrics *Metrics, cfg OpsLogConfig) {
527537

528538
if metricsConfig.TrackBytesSentByIP {
529539
// Publish bytes sent per IP & User
530-
totalMetrics.BytesSentByIP.Range(func(key, bytesSent any) bool {
540+
currentMetrics.BytesSentByIP.Range(func(key, bytesSent any) bool {
531541
keyStr := key.(string)
532542
parts := strings.Split(keyStr, "|")
533543
user, ip := parts[0], parts[1]
@@ -548,7 +558,7 @@ func PublishToPrometheus(totalMetrics *Metrics, cfg OpsLogConfig) {
548558

549559
if metricsConfig.TrackBytesReceivedByIP {
550560
// Publish bytes received per IP & User
551-
totalMetrics.BytesReceivedByIP.Range(func(key, bytesReceived any) bool {
561+
currentMetrics.BytesReceivedByIP.Range(func(key, bytesReceived any) bool {
552562
keyStr := key.(string)
553563
parts := strings.Split(keyStr, "|")
554564
user, ip := parts[0], parts[1]
@@ -602,6 +612,12 @@ func PublishToPrometheus(totalMetrics *Metrics, cfg OpsLogConfig) {
602612
return true
603613
}
604614
_, bucket, status := parts[0], parts[1], parts[2]
615+
616+
// Exclude HTTP status codes in the 2xx range
617+
if strings.HasPrefix(status, "2") {
618+
return true
619+
}
620+
605621
errorCount := float64(count.(*atomic.Uint64).Load())
606622

607623
errorsCounter.With(prometheus.Labels{
@@ -618,8 +634,14 @@ func PublishToPrometheus(totalMetrics *Metrics, cfg OpsLogConfig) {
618634

619635
if metricsConfig.TrackErrorsByStatus {
620636
diffMetrics.RequestsByStatusCode.Range(func(status, count any) bool {
621-
requestCount := float64(count.(*atomic.Uint64).Load())
637+
statusStr := status.(string)
622638

639+
// Exclude HTTP status codes in the 2xx range
640+
if strings.HasPrefix(statusStr, "2") {
641+
return true
642+
}
643+
644+
requestCount := float64(count.(*atomic.Uint64).Load())
623645
errorsCounter.With(prometheus.Labels{
624646
"pod": cfg.PodName,
625647
"user": "all",
@@ -670,10 +692,15 @@ func PublishToPrometheus(totalMetrics *Metrics, cfg OpsLogConfig) {
670692
// Fetch request count for this method
671693
countVal, exists := diffMetrics.RequestsByMethod.Load(key)
672694
if !exists {
673-
return true // Skip if no requests exist for this method
695+
countVal, exists = currentMetrics.RequestsByMethod.Load(key)
696+
if !exists {
697+
log.Warn().Msgf("Missing request count for latency key: %v", key)
698+
return true
699+
}
674700
}
675701
count := float64(countVal.(*atomic.Uint64).Load())
676702
if count == 0 {
703+
log.Warn().Msgf("Zero request count for latency key: %v", key)
677704
return true
678705
}
679706

0 commit comments

Comments
 (0)