add prometheus metric exporter framework

docs/configuration.md

@@ -47,4 +47,5 @@ The following table lists all KoP configurations.
 |kopSslTrustmanagerAlgorithm| Kafka SSL configuration map with: SSL_TRUSTMANAGER_ALGORITHM_CONFIG = ssl.trustmanager.algorithm |SunX509|
 |kopSslSecureRandomImplementation| Kafka SSL configuration map with: SSL_SECURE_RANDOM_IMPLEMENTATION_CONFIG = ssl.secure.random.implementation  | N/A|
 |saslAllowedMechanisms| Supported SASL mechanisms exposed by the broker |N/A|
+|kopPrometheusStatsLatencyRolloverSeconds|Kop metrics expose to prometheus rollover latency | 60 |
 

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaChannelInitializer.java

@@ -24,6 +24,7 @@
 import io.streamnative.pulsar.handlers.kop.coordinator.transaction.TransactionCoordinator;
 import io.streamnative.pulsar.handlers.kop.utils.ssl.SSLUtils;
 import lombok.Getter;
+import org.apache.bookkeeper.stats.StatsLogger;
 import org.apache.pulsar.broker.PulsarService;
 import org.eclipse.jetty.util.ssl.SslContextFactory;
 
@@ -48,20 +49,24 @@ public class KafkaChannelInitializer extends ChannelInitializer<SocketChannel> {
     private final EndPoint advertisedEndPoint;
     @Getter
     private final SslContextFactory sslContextFactory;
+    @Getter
+    private final StatsLogger statsLogger;
 
     public KafkaChannelInitializer(PulsarService pulsarService,
                                    KafkaServiceConfiguration kafkaConfig,
                                    GroupCoordinator groupCoordinator,
                                    TransactionCoordinator transactionCoordinator,
                                    boolean enableTLS,
-                                   EndPoint advertisedEndPoint) {
+                                   EndPoint advertisedEndPoint,
+                                   StatsLogger statsLogger) {
         super();
         this.pulsarService = pulsarService;
         this.kafkaConfig = kafkaConfig;
         this.groupCoordinator = groupCoordinator;
         this.transactionCoordinator = transactionCoordinator;
         this.enableTls = enableTLS;
         this.advertisedEndPoint = advertisedEndPoint;
+        this.statsLogger = statsLogger;
 
         if (enableTls) {
             sslContextFactory = SSLUtils.createSslContextFactory(kafkaConfig);
@@ -80,7 +85,7 @@ protected void initChannel(SocketChannel ch) throws Exception {
             new LengthFieldBasedFrameDecoder(MAX_FRAME_LENGTH, 0, 4, 0, 4));
         ch.pipeline().addLast("handler",
             new KafkaRequestHandler(pulsarService, kafkaConfig,
-                    groupCoordinator, transactionCoordinator, enableTls, advertisedEndPoint));
+                    groupCoordinator, transactionCoordinator, enableTls, advertisedEndPoint, statsLogger));
     }
 
 }

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaProtocolHandler.java

@@ -14,6 +14,7 @@
 package io.streamnative.pulsar.handlers.kop;
 
 import static com.google.common.base.Preconditions.checkState;
+import static io.streamnative.pulsar.handlers.kop.KopServerStats.SERVER_SCOPE;
 import static io.streamnative.pulsar.handlers.kop.utils.TopicNameUtils.getKafkaTopicNameFromPulsarTopicname;
 import static org.apache.pulsar.common.naming.TopicName.PARTITIONED_TOPIC_SUFFIX;
 
@@ -25,6 +26,7 @@
 import io.streamnative.pulsar.handlers.kop.coordinator.group.OffsetConfig;
 import io.streamnative.pulsar.handlers.kop.coordinator.transaction.TransactionConfig;
 import io.streamnative.pulsar.handlers.kop.coordinator.transaction.TransactionCoordinator;
+import io.streamnative.pulsar.handlers.kop.stats.PrometheusMetricsProvider;
 import io.streamnative.pulsar.handlers.kop.utils.ConfigurationUtils;
 import io.streamnative.pulsar.handlers.kop.utils.KopTopic;
 import io.streamnative.pulsar.handlers.kop.utils.MetadataUtils;
@@ -39,6 +41,9 @@
 import java.util.stream.Collectors;
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.stats.StatsLogger;
+import org.apache.commons.configuration.Configuration;
+import org.apache.commons.configuration.PropertiesConfiguration;
 import org.apache.kafka.common.internals.Topic;
 import org.apache.kafka.common.record.CompressionType;
 import org.apache.kafka.common.security.auth.SecurityProtocol;
@@ -66,6 +71,8 @@ public class KafkaProtocolHandler implements ProtocolHandler {
     public static final String PROTOCOL_NAME = "kafka";
     public static final String TLS_HANDLER = "tls";
 
+    private StatsLogger rootStatsLogger;
+    private PrometheusMetricsProvider statsProvider;
     /**
      * Listener for the changing of topic that stores offsets of consumer group.
      */
@@ -223,6 +230,9 @@ public void initialize(ServiceConfiguration conf) throws Exception {
         }
         this.bindAddress = ServiceConfigurationUtils.getDefaultOrConfiguredAddress(kafkaConfig.getBindAddress());
         KopTopic.initialize(kafkaConfig.getKafkaTenant() + "/" + kafkaConfig.getKafkaNamespace());
+
+        statsProvider = new PrometheusMetricsProvider();
+        rootStatsLogger = statsProvider.getStatsLogger("");
     }
 
     // This method is called after initialize
@@ -264,6 +274,12 @@ public void start(BrokerService service) {
                 log.error("Initialized transaction coordinator failed.", e);
             }
         }
+
+        Configuration conf = new PropertiesConfiguration();
+        conf.addProperty("prometheusStatsLatencyRolloverSeconds",
+            kafkaConfig.getKopPrometheusStatsLatencyRolloverSeconds());
+        statsProvider.start(conf);
+        brokerService.pulsar().addPrometheusRawMetricsProvider(statsProvider);
     }
 
     // this is called after initialize, and with kafkaConfig, brokerService all set.
@@ -291,12 +307,14 @@ public Map<InetSocketAddress, ChannelInitializer<SocketChannel>> newChannelIniti
                     case PLAINTEXT:
                     case SASL_PLAINTEXT:
                         builder.put(endPoint.getInetAddress(), new KafkaChannelInitializer(brokerService.getPulsar(),
-                                kafkaConfig, groupCoordinator, transactionCoordinator, false, advertisedEndPoint));
+                                kafkaConfig, groupCoordinator, transactionCoordinator, false,
+                            advertisedEndPoint, rootStatsLogger.scope(SERVER_SCOPE)));
                         break;
                     case SSL:
                     case SASL_SSL:
                         builder.put(endPoint.getInetAddress(), new KafkaChannelInitializer(brokerService.getPulsar(),
-                                kafkaConfig, groupCoordinator, transactionCoordinator, true, advertisedEndPoint));
+                                kafkaConfig, groupCoordinator, transactionCoordinator, true,
+                            advertisedEndPoint, rootStatsLogger.scope(SERVER_SCOPE)));
                         break;
                 }
             });
@@ -314,6 +332,7 @@ public void close() {
             groupCoordinator.shutdown();
         }
         KafkaTopicManager.LOOKUP_CACHE.clear();
+        statsProvider.stop();
     }
 
     public void initGroupCoordinator(BrokerService service) throws Exception {

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java

@@ -57,18 +57,21 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 import javax.naming.AuthenticationException;
 
 import lombok.Getter;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.common.util.MathUtils;
 import org.apache.bookkeeper.mledger.AsyncCallbacks;
 import org.apache.bookkeeper.mledger.ManagedLedgerException;
 import org.apache.bookkeeper.mledger.Position;
 import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl;
 import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.bookkeeper.stats.StatsLogger;
 import org.apache.commons.lang3.NotImplementedException;
 import org.apache.kafka.common.Node;
 import org.apache.kafka.common.TopicPartition;
@@ -190,13 +193,16 @@ public class KafkaRequestHandler extends KafkaCommandDecoder {
     private final EntryFormatter entryFormatter;
 
     private final Map<TopicPartition, PendingProduceQueue> pendingProduceQueueMap = new ConcurrentHashMap<>();
+    private final StatsLogger statsLogger;
+    private final RequestStats requestStats;
 
     public KafkaRequestHandler(PulsarService pulsarService,
                                KafkaServiceConfiguration kafkaConfig,
                                GroupCoordinator groupCoordinator,
                                TransactionCoordinator transactionCoordinator,
                                Boolean tlsEnabled,
-                               EndPoint advertisedEndPoint) throws Exception {
+                               EndPoint advertisedEndPoint,
+                               StatsLogger statsLogger) throws Exception {
         super();
         this.pulsarService = pulsarService;
         this.kafkaConfig = kafkaConfig;
@@ -220,6 +226,8 @@ public KafkaRequestHandler(PulsarService pulsarService,
         this.entryFormatter = EntryFormatterFactory.create(kafkaConfig.getEntryFormat());
         this.currentConnectedGroup = new ConcurrentHashMap<>();
         this.groupIdStoredPath = kafkaConfig.getGroupIdZooKeeperPath();
+        this.statsLogger = statsLogger;
+        this.requestStats = new RequestStats(statsLogger);
     }
 
     @Override
@@ -618,6 +626,8 @@ protected void handleTopicMetadataRequest(KafkaHeaderAndRequest metadataHar,
 
     protected void handleProduceRequest(KafkaHeaderAndRequest produceHar,
                                         CompletableFuture<AbstractResponse> resultFuture) {
+        final long startProduceNanos = MathUtils.nowInNano();
+
         checkArgument(produceHar.getRequest() instanceof ProduceRequest);
         ProduceRequest produceRequest = (ProduceRequest) produceHar.getRequest();
         if (produceRequest.transactionalId() != null) {
@@ -650,7 +660,7 @@ protected void handleProduceRequest(KafkaHeaderAndRequest produceHar,
 
                 String fullPartitionName = KopTopic.toString(topicPartition);
                 PendingProduce pendingProduce = new PendingProduce(partitionResponse, topicManager, fullPartitionName,
-                    entryFormatter, validRecords, executor, transactionCoordinator);
+                    entryFormatter, validRecords, executor, transactionCoordinator, requestStats);
                 PendingProduceQueue queue =
                     pendingProduceQueueMap.computeIfAbsent(topicPartition, ignored -> new PendingProduceQueue());
                 queue.add(pendingProduce);
@@ -675,6 +685,15 @@ protected void handleProduceRequest(KafkaHeaderAndRequest produceHar,
                         log.debug("[{}] Request {}: Complete handle produce.",
                                 ctx.channel(), produceHar.toString());
                     }
+
+                    if (ex != null) {
+                        requestStats.getHandleProduceRequestStats()
+                            .registerFailedEvent(MathUtils.elapsedNanos(startProduceNanos), TimeUnit.NANOSECONDS);
+                    } else {
+                        requestStats.getHandleProduceRequestStats()
+                            .registerSuccessfulEvent(MathUtils.elapsedNanos(startProduceNanos), TimeUnit.NANOSECONDS);
+                    }
+
                     resultFuture.complete(new ProduceResponse(responses));
                 });
     }

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaServiceConfiguration.java

@@ -321,6 +321,12 @@ public class KafkaServiceConfiguration extends ServiceConfiguration {
     )
     private String kopOauth2ConfigFile;
 
+    @FieldContext(
+        category = CATEGORY_KOP,
+        doc = "KOP Prometheus stats rollover latency"
+    )
+    private int kopPrometheusStatsLatencyRolloverSeconds = 60;
+
     public @NonNull String getKafkaAdvertisedListeners() {
         if (kafkaAdvertisedListeners != null) {
             return kafkaAdvertisedListeners;

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KopServerStats.java

@@ -0,0 +1,32 @@
+/**
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.streamnative.pulsar.handlers.kop;
+
+/**
+ * Kop server stats for prometheus metrics.
+ */
+public interface KopServerStats {
+    String CATEGORY_SERVER = "server";
+
+    String SERVER_SCOPE = "kop_server";
+
+    /**
+     * PRODUCE STATS.
+     */
+    String HANDLE_PRODUCE_REQUEST = "HANDLE_PRODUCE_REQUEST";
+    String PRODUCE_ENCODE = "PRODUCE_ENCODE";
+    String MESSAGE_PUBLISH = "MESSAGE_PUBLISH";
+    String MESSAGE_QUEUED_LATENCY = "MESSAGE_QUEUED_LATENCY";
+
+}

kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/PendingProduce.java

@@ -19,7 +19,9 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.bookkeeper.common.util.MathUtils;
 import org.apache.kafka.common.protocol.Errors;
 import org.apache.kafka.common.record.MemoryRecords;
 import org.apache.kafka.common.record.RecordBatch;
@@ -44,14 +46,18 @@ public class PendingProduce {
     private final TransactionCoordinator transactionCoordinator;
     private long pid;
     private boolean isTransactional;
+    private RequestStats requestStats;
+    private final long enqueueTimestamp;
 
     public PendingProduce(CompletableFuture<PartitionResponse> responseFuture,
                           KafkaTopicManager topicManager,
                           String partitionName,
                           EntryFormatter entryFormatter,
                           MemoryRecords memoryRecords,
                           ExecutorService executor,
-                          TransactionCoordinator transactionCoordinator) {
+                          TransactionCoordinator transactionCoordinator,
+                          RequestStats requestStats) {
+        this.enqueueTimestamp = MathUtils.nowInNano();
         this.responseFuture = responseFuture;
         this.topicManager = topicManager;
         this.partitionName = partitionName;
@@ -67,12 +73,15 @@ public PendingProduce(CompletableFuture<PartitionResponse> responseFuture,
             return null;
         });
         executor.execute(() -> byteBufFuture.complete(entryFormatter.encode(memoryRecords, numMessages)));
+        requestStats.getProduceEncodeStats()
+            .registerSuccessfulEvent(MathUtils.elapsedNanos(enqueueTimestamp), TimeUnit.NANOSECONDS);
         this.offsetFuture = new CompletableFuture<>();
 
         RecordBatch batch = memoryRecords.batchIterator().next();
         this.transactionCoordinator = transactionCoordinator;
         this.pid = batch.producerId();
         this.isTransactional = batch.isTransactional();
+        this.requestStats = requestStats;
     }
 
     public boolean ready() {
@@ -103,6 +112,10 @@ public void publishMessages() {
         if (!ready()) {
             throw new RuntimeException("Try to send while PendingProduce is not ready");
         }
+
+        requestStats.getMessageQueuedLatencyStats()
+            .registerSuccessfulEvent(MathUtils.elapsedNanos(enqueueTimestamp), TimeUnit.NANOSECONDS);
+
         PersistentTopic persistentTopic;
         ByteBuf byteBuf;
         try {
@@ -129,9 +142,15 @@ public void publishMessages() {
                 if (this.isTransactional) {
                     transactionCoordinator.addActivePidOffset(TopicName.get(partitionName), pid, offset);
                 }
+                requestStats.getMessagePublishStats().registerSuccessfulEvent(
+                    MathUtils.elapsedNanos(enqueueTimestamp), TimeUnit.NANOSECONDS);
+
                 responseFuture.complete(new PartitionResponse(Errors.NONE, offset, -1L, -1L));
             } else {
                 log.error("publishMessages for topic partition: {} failed when write.", partitionName, e);
+                requestStats.getMessagePublishStats()
+                    .registerFailedEvent(MathUtils.elapsedNanos(enqueueTimestamp), TimeUnit.NANOSECONDS);
+
                 responseFuture.complete(new PartitionResponse(Errors.KAFKA_STORAGE_ERROR));
             }
             byteBuf.release();