feat(metrics): add metrics on the log append permits (#2081)

Signed-off-by: Ning Yu <[email protected]>

Author: Chillax-0v0

core/src/main/scala/kafka/log/streamaspect/ElasticLog.scala

@@ -27,6 +27,7 @@ import org.apache.kafka.common.utils.{ThreadUtils, Time}
 import org.apache.kafka.common.{KafkaException, TopicPartition, Uuid}
 import org.apache.kafka.metadata.stream.StreamTags
 import org.apache.kafka.server.metrics.KafkaMetricsGroup
+import org.apache.kafka.server.metrics.s3stream.S3StreamKafkaMetricsManager
 import org.apache.kafka.server.util.Scheduler
 import org.apache.kafka.storage.internals.checkpoint.LeaderEpochCheckpointFile
 import org.apache.kafka.storage.internals.log._
@@ -583,6 +584,7 @@ class ElasticLog(val metaStream: MetaStream,
 object ElasticLog extends Logging {
     private val APPEND_PERMIT = 100 * 1024 * 1024
     private val APPEND_PERMIT_SEMAPHORE = new Semaphore(APPEND_PERMIT)
+    S3StreamKafkaMetricsManager.setLogAppendPermitNumSupplier(() => APPEND_PERMIT_SEMAPHORE.availablePermits())
 
     private val LAST_RECORD_TIMESTAMP = new AtomicLong()
     private val KafkaMetricsGroup = new KafkaMetricsGroup(ElasticLog.getClass)

server-common/src/main/java/org/apache/kafka/server/metrics/s3stream/S3StreamKafkaMetricsConstants.java

@@ -21,6 +21,7 @@ public class S3StreamKafkaMetricsConstants {
     public static final String STREAM_OBJECT_NUM = "stream_object_num";
     public static final String FETCH_LIMITER_PERMIT_NUM = "fetch_limiter_permit_num";
     public static final String FETCH_PENDING_TASK_NUM = "fetch_pending_task_num";
+    public static final String LOG_APPEND_PERMIT_NUM = "log_append_permit_num";
     public static final String SLOW_BROKER_METRIC_NAME = "slow_broker_count";
     public static final String TOPIC_PARTITION_COUNT_METRIC_NAME = "topic_partition_count";
 

server-common/src/main/java/org/apache/kafka/server/metrics/s3stream/S3StreamKafkaMetricsManager.java

@@ -63,6 +63,8 @@ public class S3StreamKafkaMetricsManager {
     private static Supplier<Map<String, Integer>> fetchLimiterPermitNumSupplier = Collections::emptyMap;
     private static ObservableLongGauge fetchPendingTaskNumMetrics = new NoopObservableLongGauge();
     private static Supplier<Map<String, Integer>> fetchPendingTaskNumSupplier = Collections::emptyMap;
+    private static ObservableLongGauge logAppendPermitNumMetrics = new NoopObservableLongGauge();
+    private static Supplier<Integer> logAppendPermitNumSupplier = () -> 0;
     private static MetricsConfig metricsConfig = new MetricsConfig(MetricsLevel.INFO, Attributes.empty());
     private static ObservableLongGauge slowBrokerMetrics = new NoopObservableLongGauge();
     private static Supplier<Map<Integer, Boolean>> slowBrokerSupplier = Collections::emptyMap;
@@ -86,6 +88,7 @@ public static void initMetrics(Meter meter, String prefix) {
         initAutoBalancerMetrics(meter, prefix);
         initObjectMetrics(meter, prefix);
         initFetchMetrics(meter, prefix);
+        initLogAppendMetrics(meter, prefix);
         initPartitionStatusStatisticsMetrics(meter, prefix);
     }
 
@@ -203,6 +206,17 @@ private static void initFetchMetrics(Meter meter, String prefix) {
                 });
     }
 
+    private static void initLogAppendMetrics(Meter meter, String prefix) {
+        logAppendPermitNumMetrics = meter.gaugeBuilder(prefix + S3StreamKafkaMetricsConstants.LOG_APPEND_PERMIT_NUM)
+                .setDescription("The number of permits in elastic log append limiter")
+                .ofLongs()
+                .buildWithCallback(result -> {
+                    if (MetricsLevel.INFO.isWithin(metricsConfig.getMetricsLevel())) {
+                        result.record(logAppendPermitNumSupplier.get(), metricsConfig.getBaseAttributes());
+                    }
+                });
+    }
+
     private static void initPartitionStatusStatisticsMetrics(Meter meter, String prefix) {
         partitionStatusStatisticsMetrics = meter.gaugeBuilder(prefix + S3StreamKafkaMetricsConstants.PARTITION_STATUS_STATISTICS_METRIC_NAME)
                 .setDescription("The statistics of partition status")
@@ -248,6 +262,10 @@ public static void setFetchPendingTaskNumSupplier(Supplier<Map<String, Integer>>
         S3StreamKafkaMetricsManager.fetchPendingTaskNumSupplier = fetchPendingTaskNumSupplier;
     }
 
+    public static void setLogAppendPermitNumSupplier(Supplier<Integer> logAppendPermitNumSupplier) {
+        S3StreamKafkaMetricsManager.logAppendPermitNumSupplier = logAppendPermitNumSupplier;
+    }
+
     public static void setSlowBrokerSupplier(Supplier<Map<Integer, Boolean>> slowBrokerSupplier) {
         S3StreamKafkaMetricsManager.slowBrokerSupplier = slowBrokerSupplier;
     }