chore(metrics): move queue time record to AsyncNetworkBandwidthLimiter (#3011)

Signed-off-by: Robin Han <[email protected]>

Author: superhx

core/src/main/scala/kafka/server/streamaspect/ElasticReplicaManager.scala

@@ -1,7 +1,6 @@
 package kafka.server.streamaspect
 
 import com.automq.stream.api.exceptions.FastReadFailFastException
-import com.automq.stream.s3.metrics.stats.NetworkStats
 import com.automq.stream.s3.metrics.{MetricsLevel, TimerUtil}
 import com.automq.stream.s3.network.{AsyncNetworkBandwidthLimiter, GlobalNetworkBandwidthLimiters, ThrottleStrategy}
 import com.automq.stream.utils.{FutureUtil, Systems}
@@ -980,11 +979,8 @@ class ElasticReplicaManager(
   }
 
   private def acquireNetworkOutPermit(size: Int, throttleStrategy: ThrottleStrategy): Unit = {
-    val start = time.nanoseconds()
     GlobalNetworkBandwidthLimiters.instance().get(AsyncNetworkBandwidthLimiter.Type.OUTBOUND)
       .consume(throttleStrategy, size).join()
-    val networkStats = NetworkStats.getInstance()
-    networkStats.networkLimiterQueueTimeStats(AsyncNetworkBandwidthLimiter.Type.OUTBOUND, throttleStrategy).record(time.nanoseconds() - start)
   }
 
   def handlePartitionFailure(partitionDir: String): Unit = {

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

@@ -96,8 +96,18 @@ public Map<Long, Pair<Counter, Counter>> allStreamReadBytesStats() {
     }
 
     public HistogramMetric networkLimiterQueueTimeStats(AsyncNetworkBandwidthLimiter.Type type, ThrottleStrategy strategy) {
-        return type == AsyncNetworkBandwidthLimiter.Type.INBOUND
-                ? networkInboundLimiterQueueTimeStatsMap.computeIfAbsent(strategy, k -> S3StreamMetricsManager.buildNetworkInboundLimiterQueueTimeMetric(MetricsLevel.INFO, strategy))
-                : networkOutboundLimiterQueueTimeStatsMap.computeIfAbsent(strategy, k -> S3StreamMetricsManager.buildNetworkOutboundLimiterQueueTimeMetric(MetricsLevel.INFO, strategy));
+        HistogramMetric metric;
+        if (type == AsyncNetworkBandwidthLimiter.Type.INBOUND) {
+            metric = networkInboundLimiterQueueTimeStatsMap.get(strategy);
+            if (metric == null) {
+                metric = networkInboundLimiterQueueTimeStatsMap.computeIfAbsent(strategy, k -> S3StreamMetricsManager.buildNetworkInboundLimiterQueueTimeMetric(MetricsLevel.INFO, strategy));
+            }
+        } else {
+            metric = networkOutboundLimiterQueueTimeStatsMap.get(strategy);
+            if (metric == null) {
+                metric = networkOutboundLimiterQueueTimeStatsMap.computeIfAbsent(strategy, k -> S3StreamMetricsManager.buildNetworkOutboundLimiterQueueTimeMetric(MetricsLevel.INFO, strategy));
+            }
+        }
+        return metric;
     }
 }

s3stream/src/main/java/com/automq/stream/s3/network/AsyncNetworkBandwidthLimiter.java

@@ -187,7 +187,7 @@ public String getName() {
         }
     }
 
-    private static class BucketItem implements Comparable<BucketItem> {
+    private class BucketItem implements Comparable<BucketItem> {
         private final ThrottleStrategy strategy;
         private final CompletableFuture<Void> cf;
         private final long timestamp;
@@ -212,6 +212,7 @@ public boolean complete(long completeSize, ExecutorService executor) {
             size -= completeSize;
             if (size <= 0) {
                 executor.submit(() -> cf.complete(null));
+                NetworkStats.getInstance().networkLimiterQueueTimeStats(type, strategy).record(System.nanoTime() - timestamp);
                 return true;
             }
             return false;

s3stream/src/main/java/com/automq/stream/s3/operator/AbstractObjectStorage.java

@@ -24,10 +24,8 @@
 import com.automq.stream.s3.metrics.S3StreamMetricsManager;
 import com.automq.stream.s3.metrics.TimerUtil;
 import com.automq.stream.s3.metrics.operations.S3Operation;
-import com.automq.stream.s3.metrics.stats.NetworkStats;
 import com.automq.stream.s3.metrics.stats.S3OperationStats;
 import com.automq.stream.s3.metrics.stats.StorageOperationStats;
-import com.automq.stream.s3.network.AsyncNetworkBandwidthLimiter;
 import com.automq.stream.s3.network.NetworkBandwidthLimiter;
 import com.automq.stream.s3.network.ThrottleStrategy;
 import com.automq.stream.s3.objects.ObjectAttributes;
@@ -227,15 +225,7 @@ public CompletableFuture<ByteBuf> rangeRead(ReadOptions options, String objectPa
         }
 
         BiFunction<ThrottleStrategy, Long, CompletableFuture<Void>> networkInboundBandwidthLimiterFunction =
-            (throttleStrategy, size) -> {
-                long startTime = System.nanoTime();
-                return networkInboundBandwidthLimiter.consume(throttleStrategy, size)
-                    .whenComplete((v, ex) ->
-                        NetworkStats.getInstance()
-                            .networkLimiterQueueTimeStats(AsyncNetworkBandwidthLimiter.Type.INBOUND, throttleStrategy)
-                            .record(TimerUtil.timeElapsedSince(startTime, TimeUnit.NANOSECONDS)));
-
-            };
+            networkInboundBandwidthLimiter::consume;
 
         long acquiredSize = end - start;
 
@@ -277,12 +267,9 @@ public CompletableFuture<WriteResult> write(WriteOptions options, String objectP
             data.release();
             return retCf;
         }
-        TimerUtil timerUtil = new TimerUtil();
         networkOutboundBandwidthLimiter
             .consume(options.throttleStrategy(), data.readableBytes())
             .whenCompleteAsync((v, ex) -> {
-                NetworkStats.getInstance().networkLimiterQueueTimeStats(AsyncNetworkBandwidthLimiter.Type.OUTBOUND, options.throttleStrategy())
-                    .record(timerUtil.elapsedAs(TimeUnit.NANOSECONDS));
                 if (ex != null) {
                     cf.completeExceptionally(ex);
                     data.release();