feat(s3stream): reduce histogram sample size by dividing into explici…

…t percentiles

Signed-off-by: Shichao Nie <[email protected]>

s3stream/build.gradle

@@ -48,6 +48,7 @@ dependencies {
     api 'org.aspectj:aspectjrt:1.9.20.1'
     api 'org.aspectj:aspectjweaver:1.9.20.1'
     api 'com.github.jnr:jnr-posix:3.1.19'
+    api 'com.yammer.metrics:metrics-core:2.2.0'
     testImplementation 'org.slf4j:slf4j-simple:2.0.9'
     testImplementation 'org.junit.jupiter:junit-jupiter:5.10.0'
     testImplementation 'org.mockito:mockito-core:5.5.0'

s3stream/src/main/java/com/automq/stream/s3/metrics/S3StreamMetricsConstant.java

@@ -70,7 +70,6 @@ public class S3StreamMetricsConstant {
     public static final String OBJECT_COUNT_METRIC_NAME = "object_count";
     public static final String OBJECT_STAGE_COST_METRIC_NAME = "object_stage_cost";
     public static final String OBJECT_UPLOAD_SIZE_METRIC_NAME = "object_upload_size";
-    public static final String OBJECT_DOWNLOAD_SIZE_METRIC_NAME = "object_download_size";
     public static final String NETWORK_INBOUND_USAGE_METRIC_NAME = "network_inbound_usage";
     public static final String NETWORK_OUTBOUND_USAGE_METRIC_NAME = "network_outbound_usage";
     public static final String NETWORK_INBOUND_AVAILABLE_BANDWIDTH_METRIC_NAME = "network_inbound_available_bandwidth";
@@ -80,6 +79,12 @@ public class S3StreamMetricsConstant {
     public static final String NETWORK_INBOUND_LIMITER_QUEUE_TIME_METRIC_NAME = "network_inbound_limiter_queue_time";
     public static final String NETWORK_OUTBOUND_LIMITER_QUEUE_TIME_METRIC_NAME = "network_outbound_limiter_queue_time";
     public static final String READ_AHEAD_SIZE_METRIC_NAME = "read_ahead_size";
+    public static final String SUM_METRIC_NAME_SUFFIX = "_sum";
+    public static final String COUNT_METRIC_NAME_SUFFIX = "_count";
+    public static final String P50_METRIC_NAME_SUFFIX = "_50p";
+    public static final String P99_METRIC_NAME_SUFFIX = "_99p";
+    public static final String MEAN_METRIC_NAME_SUFFIX = "_mean";
+    public static final String MAX_METRIC_NAME_SUFFIX = "_max";
     public static final String WAL_START_OFFSET = "wal_start_offset";
     public static final String WAL_TRIMMED_OFFSET = "wal_trimmed_offset";
     public static final String DELTA_WAL_CACHE_SIZE = "delta_wal_cache_size";

s3stream/src/main/java/com/automq/stream/s3/metrics/operations/S3Operation.java

@@ -67,7 +67,7 @@ public enum S3Operation {
     S3Operation(S3MetricsType type, String name) {
         this.type = type;
         this.name = name;
-        uniqueKey = type + "-" + name;
+        uniqueKey = type.getName() + "-" + name;
     }
 
     public String getName() {

s3stream/src/main/java/com/automq/stream/s3/metrics/stats/S3OperationStats.java

@@ -81,11 +81,13 @@ public YammerHistogramMetric getObjectStats(long size, boolean isSuccess) {
         String label = AttributesUtils.getObjectBucketLabel(size);
         if (isSuccess) {
             return getObjectSuccessStats.computeIfAbsent(label, name -> S3StreamMetricsManager.buildOperationMetric(
-                new MetricName(S3OperationStats.class, S3Operation.GET_OBJECT.getUniqueKey() + S3StreamMetricsConstant.LABEL_STATUS_SUCCESS),
+                new MetricName(S3OperationStats.class, S3Operation.GET_OBJECT.getUniqueKey()
+                    + S3StreamMetricsConstant.LABEL_STATUS_SUCCESS + label),
                 MetricsLevel.INFO, S3Operation.GET_OBJECT, S3StreamMetricsConstant.LABEL_STATUS_SUCCESS, label));
         } else {
             return getObjectFailedStats.computeIfAbsent(label, name -> S3StreamMetricsManager.buildOperationMetric(
-                new MetricName(S3OperationStats.class, S3Operation.GET_OBJECT.getUniqueKey() + S3StreamMetricsConstant.LABEL_STATUS_FAILED),
+                new MetricName(S3OperationStats.class, S3Operation.GET_OBJECT.getUniqueKey()
+                    + S3StreamMetricsConstant.LABEL_STATUS_FAILED + label),
                 MetricsLevel.INFO, S3Operation.GET_OBJECT, S3StreamMetricsConstant.LABEL_STATUS_FAILED, label));
         }
     }
@@ -94,11 +96,13 @@ public YammerHistogramMetric putObjectStats(long size, boolean isSuccess) {
         String label = AttributesUtils.getObjectBucketLabel(size);
         if (isSuccess) {
             return putObjectSuccessStats.computeIfAbsent(label, name -> S3StreamMetricsManager.buildOperationMetric(
-                new MetricName(S3OperationStats.class, S3Operation.PUT_OBJECT.getUniqueKey() + S3StreamMetricsConstant.LABEL_STATUS_SUCCESS),
+                new MetricName(S3OperationStats.class, S3Operation.PUT_OBJECT.getUniqueKey()
+                    + S3StreamMetricsConstant.LABEL_STATUS_SUCCESS + label),
                 MetricsLevel.INFO, S3Operation.PUT_OBJECT, S3StreamMetricsConstant.LABEL_STATUS_SUCCESS, label));
         } else {
             return putObjectFailedStats.computeIfAbsent(label, name -> S3StreamMetricsManager.buildOperationMetric(
-                new MetricName(S3OperationStats.class, S3Operation.PUT_OBJECT.getUniqueKey() + S3StreamMetricsConstant.LABEL_STATUS_FAILED),
+                new MetricName(S3OperationStats.class, S3Operation.PUT_OBJECT.getUniqueKey() +
+                    S3StreamMetricsConstant.LABEL_STATUS_FAILED + label),
                 MetricsLevel.INFO, S3Operation.PUT_OBJECT, S3StreamMetricsConstant.LABEL_STATUS_FAILED, label));
         }
     }
@@ -107,11 +111,13 @@ public YammerHistogramMetric uploadPartStats(long size, boolean isSuccess) {
         String label = AttributesUtils.getObjectBucketLabel(size);
         if (isSuccess) {
             return uploadPartSuccessStats.computeIfAbsent(label, name -> S3StreamMetricsManager.buildOperationMetric(
-                new MetricName(S3OperationStats.class, S3Operation.UPLOAD_PART.getUniqueKey() + S3StreamMetricsConstant.LABEL_STATUS_SUCCESS),
+                new MetricName(S3OperationStats.class, S3Operation.UPLOAD_PART.getUniqueKey() +
+                    S3StreamMetricsConstant.LABEL_STATUS_SUCCESS + label),
                 MetricsLevel.INFO, S3Operation.UPLOAD_PART, S3StreamMetricsConstant.LABEL_STATUS_SUCCESS, label));
         } else {
             return uploadPartFailedStats.computeIfAbsent(label, name -> S3StreamMetricsManager.buildOperationMetric(
-                new MetricName(S3OperationStats.class, S3Operation.UPLOAD_PART.getUniqueKey() + S3StreamMetricsConstant.LABEL_STATUS_FAILED),
+                new MetricName(S3OperationStats.class, S3Operation.UPLOAD_PART.getUniqueKey() +
+                    S3StreamMetricsConstant.LABEL_STATUS_FAILED + label),
                 MetricsLevel.INFO, S3Operation.UPLOAD_PART, S3StreamMetricsConstant.LABEL_STATUS_FAILED, label));
         }
     }

s3stream/src/main/java/com/automq/stream/s3/metrics/wrapper/HistogramInstrument.java

@@ -17,6 +17,7 @@
 
 package com.automq.stream.s3.metrics.wrapper;
 
+import com.automq.stream.s3.metrics.S3StreamMetricsConstant;
 import io.opentelemetry.api.metrics.Meter;
 import io.opentelemetry.api.metrics.ObservableDoubleGauge;
 import io.opentelemetry.api.metrics.ObservableLongGauge;
@@ -34,7 +35,7 @@ public class HistogramInstrument {
 
     public HistogramInstrument(Meter meter, String name, String desc, String unit) {
         this.histograms = new CopyOnWriteArrayList<>();
-        this.count = meter.gaugeBuilder(name)
+        this.count = meter.gaugeBuilder(name + S3StreamMetricsConstant.COUNT_METRIC_NAME_SUFFIX)
             .setDescription(desc + " (count)")
             .ofLongs()
             .buildWithCallback(result -> {
@@ -44,17 +45,18 @@ public HistogramInstrument(Meter meter, String name, String desc, String unit) {
                     }
                 });
             });
-        this.sum = meter.gaugeBuilder(name)
+        this.sum = meter.gaugeBuilder(name + S3StreamMetricsConstant.SUM_METRIC_NAME_SUFFIX)
             .setDescription(desc + " (sum)")
             .ofLongs()
+            .setUnit(unit)
             .buildWithCallback(result -> {
                 histograms.forEach(histogram -> {
                     if (histogram.shouldRecord()) {
                         result.record(histogram.sum(), histogram.attributes);
                     }
                 });
             });
-        this.histP50Value = meter.gaugeBuilder(name)
+        this.histP50Value = meter.gaugeBuilder(name + S3StreamMetricsConstant.P50_METRIC_NAME_SUFFIX)
             .setDescription(desc + " (50th percentile)")
             .setUnit(unit)
             .buildWithCallback(result -> {
@@ -64,7 +66,7 @@ public HistogramInstrument(Meter meter, String name, String desc, String unit) {
                     }
                 });
             });
-        this.histP99Value = meter.gaugeBuilder(name)
+        this.histP99Value = meter.gaugeBuilder(name + S3StreamMetricsConstant.P99_METRIC_NAME_SUFFIX)
             .setDescription(desc + " (99th percentile)")
             .setUnit(unit)
             .buildWithCallback(result -> {
@@ -74,7 +76,7 @@ public HistogramInstrument(Meter meter, String name, String desc, String unit) {
                     }
                 });
             });
-        this.histMeanValue = meter.gaugeBuilder(name)
+        this.histMeanValue = meter.gaugeBuilder(name + S3StreamMetricsConstant.MEAN_METRIC_NAME_SUFFIX)
             .setDescription(desc + " (mean)")
             .setUnit(unit)
             .buildWithCallback(result -> {
@@ -84,7 +86,7 @@ public HistogramInstrument(Meter meter, String name, String desc, String unit) {
                     }
                 });
             });
-        this.histMaxValue = meter.gaugeBuilder(name)
+        this.histMaxValue = meter.gaugeBuilder(name + S3StreamMetricsConstant.MAX_METRIC_NAME_SUFFIX)
             .setDescription(desc + " (max)")
             .setUnit(unit)
             .buildWithCallback(result -> {