diff --git a/docs/configuration.md b/docs/configuration.md index 20c51484e9..fc72ba44f3 100644 --- a/docs/configuration.md +++ b/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 | diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaChannelInitializer.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaChannelInitializer.java index b4f6a49797..7e5293f26d 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaChannelInitializer.java +++ b/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,13 +49,16 @@ public class KafkaChannelInitializer extends ChannelInitializer { 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; @@ -62,6 +66,7 @@ public KafkaChannelInitializer(PulsarService pulsarService, 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)); } } diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaProtocolHandler.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaProtocolHandler.java index d0426521a7..fe43cb8514 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaProtocolHandler.java +++ b/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> 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 { diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java index 133124d4af..9c408cd0e4 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandler.java @@ -57,6 +57,7 @@ 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; @@ -64,11 +65,13 @@ 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 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 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)); }); } diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaServiceConfiguration.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaServiceConfiguration.java index 92040c79c9..0b3735cde6 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KafkaServiceConfiguration.java +++ b/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; diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KopServerStats.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/KopServerStats.java new file mode 100644 index 0000000000..c2a5507561 --- /dev/null +++ b/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"; + +} diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/PendingProduce.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/PendingProduce.java index f0a3248027..980995205b 100644 --- a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/PendingProduce.java +++ b/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,6 +46,8 @@ public class PendingProduce { private final TransactionCoordinator transactionCoordinator; private long pid; private boolean isTransactional; + private RequestStats requestStats; + private final long enqueueTimestamp; public PendingProduce(CompletableFuture responseFuture, KafkaTopicManager topicManager, @@ -51,7 +55,9 @@ public PendingProduce(CompletableFuture responseFuture, 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 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(); diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/RequestStats.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/RequestStats.java new file mode 100644 index 0000000000..e9b9ff0e4f --- /dev/null +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/RequestStats.java @@ -0,0 +1,69 @@ +/** + * 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; + +import static io.streamnative.pulsar.handlers.kop.KopServerStats.CATEGORY_SERVER; +import static io.streamnative.pulsar.handlers.kop.KopServerStats.HANDLE_PRODUCE_REQUEST; +import static io.streamnative.pulsar.handlers.kop.KopServerStats.MESSAGE_PUBLISH; +import static io.streamnative.pulsar.handlers.kop.KopServerStats.MESSAGE_QUEUED_LATENCY; +import static io.streamnative.pulsar.handlers.kop.KopServerStats.PRODUCE_ENCODE; +import static io.streamnative.pulsar.handlers.kop.KopServerStats.SERVER_SCOPE; + +import lombok.Getter; +import org.apache.bookkeeper.stats.OpStatsLogger; +import org.apache.bookkeeper.stats.StatsLogger; +import org.apache.bookkeeper.stats.annotations.StatsDoc; + +/** + * Kop request stats metric for prometheus metrics. + */ +@StatsDoc( + name = SERVER_SCOPE, + category = CATEGORY_SERVER, + help = "KOP request stats" +) + +@Getter +public class RequestStats { + @StatsDoc( + name = HANDLE_PRODUCE_REQUEST, + help = "handle produce request stats of Kop" + ) + private final OpStatsLogger handleProduceRequestStats; + + @StatsDoc( + name = PRODUCE_ENCODE, + help = "produce encode stats of Kop" + ) + private final OpStatsLogger produceEncodeStats; + + @StatsDoc( + name = MESSAGE_PUBLISH, + help = "message publish stats from kop to pulsar broker" + ) + private final OpStatsLogger messagePublishStats; + + @StatsDoc( + name = MESSAGE_QUEUED_LATENCY, + help = "message queued stats from kop to pulsar broker" + ) + private final OpStatsLogger messageQueuedLatencyStats; + + public RequestStats(StatsLogger statsLogger) { + this.handleProduceRequestStats = statsLogger.getOpStatsLogger(HANDLE_PRODUCE_REQUEST); + this.produceEncodeStats = statsLogger.getOpStatsLogger(PRODUCE_ENCODE); + this.messagePublishStats = statsLogger.getOpStatsLogger(MESSAGE_PUBLISH); + this.messageQueuedLatencyStats = statsLogger.getOpStatsLogger(MESSAGE_QUEUED_LATENCY); + } +} diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/DataSketchesOpStatsLogger.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/DataSketchesOpStatsLogger.java new file mode 100644 index 0000000000..e78c916f8d --- /dev/null +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/DataSketchesOpStatsLogger.java @@ -0,0 +1,193 @@ +/** + * 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.stats; + +import com.yahoo.sketches.quantiles.DoublesSketch; +import com.yahoo.sketches.quantiles.DoublesSketchBuilder; +import com.yahoo.sketches.quantiles.DoublesUnion; +import com.yahoo.sketches.quantiles.DoublesUnionBuilder; +import io.netty.util.concurrent.FastThreadLocal; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.LongAdder; +import java.util.concurrent.locks.StampedLock; +import org.apache.bookkeeper.stats.OpStatsData; +import org.apache.bookkeeper.stats.OpStatsLogger; + +/** + * OpStatsLogger implementation that uses DataSketches library to calculate the approximated latency quantiles. + */ +public class DataSketchesOpStatsLogger implements OpStatsLogger { + + /* + * Use 2 rotating thread local accessor so that we can safely swap them. + */ + private volatile ThreadLocalAccessor current; + private volatile ThreadLocalAccessor replacement; + + /* + * These are the sketches where all the aggregated results are published. + */ + private volatile DoublesSketch successResult; + private volatile DoublesSketch failResult; + + private final LongAdder successCountAdder = new LongAdder(); + private final LongAdder failCountAdder = new LongAdder(); + + private final LongAdder successSumAdder = new LongAdder(); + private final LongAdder failSumAdder = new LongAdder(); + + DataSketchesOpStatsLogger() { + this.current = new ThreadLocalAccessor(); + this.replacement = new ThreadLocalAccessor(); + } + + @Override + public void registerFailedEvent(long eventLatency, TimeUnit unit) { + double valueMillis = unit.toMicros(eventLatency) / 1000.0; + + failCountAdder.increment(); + failSumAdder.add((long) valueMillis); + + LocalData localData = current.localData.get(); + + long stamp = localData.lock.readLock(); + try { + localData.failSketch.update(valueMillis); + } finally { + localData.lock.unlockRead(stamp); + } + } + + @Override + public void registerSuccessfulEvent(long eventLatency, TimeUnit unit) { + double valueMillis = unit.toMicros(eventLatency) / 1000.0; + + successCountAdder.increment(); + successSumAdder.add((long) valueMillis); + + LocalData localData = current.localData.get(); + + long stamp = localData.lock.readLock(); + try { + localData.successSketch.update(valueMillis); + } finally { + localData.lock.unlockRead(stamp); + } + } + + @Override + public void registerSuccessfulValue(long value) { + successCountAdder.increment(); + successSumAdder.add(value); + + LocalData localData = current.localData.get(); + + long stamp = localData.lock.readLock(); + try { + localData.successSketch.update(value); + } finally { + localData.lock.unlockRead(stamp); + } + } + + @Override + public void registerFailedValue(long value) { + failCountAdder.increment(); + failSumAdder.add(value); + + LocalData localData = current.localData.get(); + + long stamp = localData.lock.readLock(); + try { + localData.failSketch.update(value); + } finally { + localData.lock.unlockRead(stamp); + } + } + + @Override + public OpStatsData toOpStatsData() { + // Not relevant as we don't use JMX here + throw new UnsupportedOperationException(); + } + + @Override + public void clear() { + // Not relevant as we don't use JMX here + throw new UnsupportedOperationException(); + } + + public void rotateLatencyCollection() { + // Swap current with replacement + ThreadLocalAccessor local = current; + current = replacement; + replacement = local; + + final DoublesUnion aggregateSuccesss = new DoublesUnionBuilder().build(); + final DoublesUnion aggregateFail = new DoublesUnionBuilder().build(); + local.map.forEach((localData, b) -> { + long stamp = localData.lock.writeLock(); + try { + aggregateSuccesss.update(localData.successSketch); + localData.successSketch.reset(); + aggregateFail.update(localData.failSketch); + localData.failSketch.reset(); + } finally { + localData.lock.unlockWrite(stamp); + } + }); + + successResult = aggregateSuccesss.getResultAndReset(); + failResult = aggregateFail.getResultAndReset(); + } + + public long getCount(boolean success) { + return success ? successCountAdder.sum() : failCountAdder.sum(); + } + + public long getSum(boolean success) { + return success ? successSumAdder.sum() : failSumAdder.sum(); + } + + public double getQuantileValue(boolean success, double quantile) { + DoublesSketch s = success ? successResult : failResult; + return s != null ? s.getQuantile(quantile) : Double.NaN; + } + + private static class LocalData { + private final DoublesSketch successSketch = new DoublesSketchBuilder().build(); + private final DoublesSketch failSketch = new DoublesSketchBuilder().build(); + private final StampedLock lock = new StampedLock(); + } + + private static class ThreadLocalAccessor { + private final Map map = new ConcurrentHashMap<>(); + private final FastThreadLocal localData = new FastThreadLocal() { + + @Override + protected LocalData initialValue() throws Exception { + LocalData localData = new LocalData(); + map.put(localData, Boolean.TRUE); + return localData; + } + + @Override + protected void onRemoval(LocalData value) throws Exception { + map.remove(value); + } + }; + } +} diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/LongAdderCounter.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/LongAdderCounter.java new file mode 100644 index 0000000000..7d678fd0bb --- /dev/null +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/LongAdderCounter.java @@ -0,0 +1,52 @@ +/** + * 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.stats; + +import java.util.concurrent.atomic.LongAdder; +import org.apache.bookkeeper.stats.Counter; + +/** + * {@link Counter} implementation based on {@link LongAdder}. + * + *

LongAdder keeps a counter per-thread and then aggregates to get the result, in order to avoid contention between + * multiple threads. + */ +public class LongAdderCounter implements Counter { + private final LongAdder counter = new LongAdder(); + + @Override + public void clear() { + counter.reset(); + } + + @Override + public void inc() { + counter.increment(); + } + + @Override + public void dec() { + counter.decrement(); + } + + @Override + public void add(long delta) { + counter.add(delta); + } + + @Override + public Long get() { + return counter.sum(); + } +} diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/PrometheusMetricsProvider.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/PrometheusMetricsProvider.java new file mode 100644 index 0000000000..eb42bbe728 --- /dev/null +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/PrometheusMetricsProvider.java @@ -0,0 +1,121 @@ +/** + * 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.stats; + +import com.google.common.annotations.VisibleForTesting; +import io.netty.util.concurrent.DefaultThreadFactory; +import io.prometheus.client.Collector; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.ConcurrentSkipListMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; +import org.apache.bookkeeper.stats.CachingStatsProvider; +import org.apache.bookkeeper.stats.StatsLogger; +import org.apache.bookkeeper.stats.StatsProvider; +import org.apache.commons.configuration.Configuration; +import org.apache.commons.lang.StringUtils; +import org.apache.pulsar.broker.stats.prometheus.PrometheusRawMetricsProvider; +import org.apache.pulsar.common.util.SimpleTextOutputStream; + +/** + * A Prometheus based {@link StatsProvider} implementation. + */ +public class PrometheusMetricsProvider implements PrometheusRawMetricsProvider { + private ScheduledExecutorService executor; + + public static final String PROMETHEUS_STATS_LATENCY_ROLLOVER_SECONDS = "prometheusStatsLatencyRolloverSeconds"; + public static final int DEFAULT_PROMETHEUS_STATS_LATENCY_ROLLOVER_SECONDS = 60; + + private final CachingStatsProvider cachingStatsProvider; + + /* + * These acts a registry of the metrics defined in this provider + */ + public final ConcurrentMap counters = new ConcurrentSkipListMap<>(); + public final ConcurrentMap> gauges = new ConcurrentSkipListMap<>(); + public final ConcurrentMap opStats = new ConcurrentSkipListMap<>(); + + + public PrometheusMetricsProvider() { + this.cachingStatsProvider = new CachingStatsProvider(new StatsProvider() { + @Override + public void start(Configuration conf) { + // nop + } + + @Override + public void stop() { + // nop + } + + @Override + public StatsLogger getStatsLogger(String scope) { + return new PrometheusStatsLogger(PrometheusMetricsProvider.this, scope); + } + + @Override + public String getStatsName(String... statsComponents) { + String completeName; + if (statsComponents.length == 0) { + return ""; + } else if (statsComponents[0].isEmpty()) { + completeName = StringUtils.join(statsComponents, '_', 1, statsComponents.length); + } else { + completeName = StringUtils.join(statsComponents, '_'); + } + return Collector.sanitizeMetricName(completeName); + } + }); + } + + public void start(Configuration conf) { + + executor = Executors.newSingleThreadScheduledExecutor(new DefaultThreadFactory("metrics")); + + int latencyRolloverSeconds = conf.getInt(PROMETHEUS_STATS_LATENCY_ROLLOVER_SECONDS, + DEFAULT_PROMETHEUS_STATS_LATENCY_ROLLOVER_SECONDS); + + executor.scheduleAtFixedRate(() -> { + rotateLatencyCollection(); + }, 1, latencyRolloverSeconds, TimeUnit.SECONDS); + + } + + public void stop() { + executor.shutdown(); + } + + public StatsLogger getStatsLogger(String scope) { + return this.cachingStatsProvider.getStatsLogger(scope); + } + + @Override + public void generate(SimpleTextOutputStream writer) { + gauges.forEach((name, gauge) -> PrometheusTextFormatUtil.writeGauge(writer, name, gauge)); + counters.forEach((name, counter) -> PrometheusTextFormatUtil.writeCounter(writer, name, counter)); + opStats.forEach((name, opStatLogger) -> PrometheusTextFormatUtil.writeOpStat(writer, name, opStatLogger)); + } + + public String getStatsName(String... statsComponents) { + return cachingStatsProvider.getStatsName(statsComponents); + } + + @VisibleForTesting + void rotateLatencyCollection() { + opStats.forEach((name, metric) -> { + metric.rotateLatencyCollection(); + }); + } +} diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/PrometheusStatsLogger.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/PrometheusStatsLogger.java new file mode 100644 index 0000000000..5727fca2fd --- /dev/null +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/PrometheusStatsLogger.java @@ -0,0 +1,70 @@ +/** + * 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.stats; + +import com.google.common.base.Joiner; +import io.prometheus.client.Collector; +import org.apache.bookkeeper.stats.Counter; +import org.apache.bookkeeper.stats.Gauge; +import org.apache.bookkeeper.stats.OpStatsLogger; +import org.apache.bookkeeper.stats.StatsLogger; + +/** + * A {@code Prometheus} based {@link StatsLogger} implementation. + */ +public class PrometheusStatsLogger implements StatsLogger { + + private final PrometheusMetricsProvider provider; + private final String scope; + + PrometheusStatsLogger(PrometheusMetricsProvider provider, String scope) { + this.provider = provider; + this.scope = scope; + } + + @Override + public OpStatsLogger getOpStatsLogger(String name) { + return provider.opStats.computeIfAbsent(completeName(name), x -> new DataSketchesOpStatsLogger()); + } + + @Override + public Counter getCounter(String name) { + return provider.counters.computeIfAbsent(completeName(name), x -> new LongAdderCounter()); + } + + @Override + public void registerGauge(String name, Gauge gauge) { + provider.gauges.computeIfAbsent(completeName(name), x -> new SimpleGauge(gauge)); + } + + @Override + public void unregisterGauge(String name, Gauge gauge) { + // no-op + } + + @Override + public void removeScope(String name, StatsLogger statsLogger) { + // no-op + } + + @Override + public StatsLogger scope(String name) { + return new PrometheusStatsLogger(provider, completeName(name)); + } + + private String completeName(String name) { + String completeName = scope.isEmpty() ? name : Joiner.on('_').join(scope, name); + return Collector.sanitizeMetricName(completeName); + } +} diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/PrometheusTextFormatUtil.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/PrometheusTextFormatUtil.java new file mode 100644 index 0000000000..2cc21dbd75 --- /dev/null +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/PrometheusTextFormatUtil.java @@ -0,0 +1,133 @@ +/** + * 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.stats; + +import io.prometheus.client.Collector; +import io.prometheus.client.Collector.MetricFamilySamples; +import io.prometheus.client.Collector.MetricFamilySamples.Sample; +import io.prometheus.client.CollectorRegistry; +import java.util.Enumeration; +import org.apache.bookkeeper.stats.Counter; +import org.apache.pulsar.common.util.SimpleTextOutputStream; + +/** + * Logic to write metrics in Prometheus text format. + */ +public class PrometheusTextFormatUtil { + public static void writeGauge(SimpleTextOutputStream w, String name, SimpleGauge gauge) { + // Example: + // # TYPE bookie_storage_entries_count gauge + // bookie_storage_entries_count 519 + w.write("# TYPE ").write(name).write(" gauge\n"); + w.write(name).write(' ').write(gauge.getSample().toString()).write('\n'); + + } + + public static void writeCounter(SimpleTextOutputStream w, String name, Counter counter) { + // Example: + // # TYPE jvm_threads_started_total counter + // jvm_threads_started_total 59 + w.write("# TYPE ").write(name).write(" counter\n"); + w.write(name).write(' ').write(counter.get().toString()).write('\n'); + } + + public static void writeOpStat(SimpleTextOutputStream w, String name, DataSketchesOpStatsLogger opStat) { + // Example: + // # TYPE bookie_journal_JOURNAL_ADD_ENTRY summary + // bookie_journal_JOURNAL_ADD_ENTRY{success="false",quantile="0.5",} NaN + // bookie_journal_JOURNAL_ADD_ENTRY{success="false",quantile="0.75",} NaN + // bookie_journal_JOURNAL_ADD_ENTRY{success="false",quantile="0.95",} NaN + // bookie_journal_JOURNAL_ADD_ENTRY{success="false",quantile="0.99",} NaN + // bookie_journal_JOURNAL_ADD_ENTRY{success="false",quantile="0.999",} NaN + // bookie_journal_JOURNAL_ADD_ENTRY{success="false",quantile="0.9999",} NaN + // bookie_journal_JOURNAL_ADD_ENTRY{success="false",quantile="1.0",} NaN + // bookie_journal_JOURNAL_ADD_ENTRY_count{success="false",} 0.0 + // bookie_journal_JOURNAL_ADD_ENTRY_sum{success="false",} 0.0 + // bookie_journal_JOURNAL_ADD_ENTRY{success="true",quantile="0.5",} 1.706 + // bookie_journal_JOURNAL_ADD_ENTRY{success="true",quantile="0.75",} 1.89 + // bookie_journal_JOURNAL_ADD_ENTRY{success="true",quantile="0.95",} 2.121 + // bookie_journal_JOURNAL_ADD_ENTRY{success="true",quantile="0.99",} 10.708 + // bookie_journal_JOURNAL_ADD_ENTRY{success="true",quantile="0.999",} 10.902 + // bookie_journal_JOURNAL_ADD_ENTRY{success="true",quantile="0.9999",} 10.902 + // bookie_journal_JOURNAL_ADD_ENTRY{success="true",quantile="1.0",} 10.902 + // bookie_journal_JOURNAL_ADD_ENTRY_count{success="true",} 658.0 + // bookie_journal_JOURNAL_ADD_ENTRY_sum{success="true",} 1265.0800000000002 + w.write("# TYPE ").write(name).write(" summary\n"); + writeQuantile(w, opStat, name, false, 0.5); + writeQuantile(w, opStat, name, false, 0.75); + writeQuantile(w, opStat, name, false, 0.95); + writeQuantile(w, opStat, name, false, 0.99); + writeQuantile(w, opStat, name, false, 0.999); + writeQuantile(w, opStat, name, false, 0.9999); + writeQuantile(w, opStat, name, false, 1.0); + writeCount(w, opStat, name, false); + writeSum(w, opStat, name, false); + + writeQuantile(w, opStat, name, true, 0.5); + writeQuantile(w, opStat, name, true, 0.75); + writeQuantile(w, opStat, name, true, 0.95); + writeQuantile(w, opStat, name, true, 0.99); + writeQuantile(w, opStat, name, true, 0.999); + writeQuantile(w, opStat, name, true, 0.9999); + writeQuantile(w, opStat, name, true, 1.0); + writeCount(w, opStat, name, true); + writeSum(w, opStat, name, true); + + } + + private static void writeQuantile(SimpleTextOutputStream w, DataSketchesOpStatsLogger opStat, String name, + Boolean success, double quantile) { + w.write(name).write("{success=\"").write(success.toString()).write("\",quantile=\"") + .write(Double.toString(quantile)).write("\"} ") + .write(Double.toString(opStat.getQuantileValue(success, quantile))).write('\n'); + } + + private static void writeCount(SimpleTextOutputStream w, DataSketchesOpStatsLogger opStat, String name, + Boolean success) { + w.write(name).write("_count{success=\"").write(success.toString()).write("\"} ") + .write(Long.toString(opStat.getCount(success))).write('\n'); + } + + private static void writeSum(SimpleTextOutputStream w, DataSketchesOpStatsLogger opStat, String name, + Boolean success) { + w.write(name).write("_sum{success=\"").write(success.toString()).write("\"} ") + .write(Double.toString(opStat.getSum(success))).write('\n'); + } + + public static void writeMetricsCollectedByPrometheusClient(SimpleTextOutputStream w, CollectorRegistry registry) { + Enumeration metricFamilySamples = registry.metricFamilySamples(); + while (metricFamilySamples.hasMoreElements()) { + MetricFamilySamples metricFamily = metricFamilySamples.nextElement(); + + for (int i = 0; i < metricFamily.samples.size(); i++) { + Sample sample = metricFamily.samples.get(i); + w.write(sample.name); + w.write('{'); + for (int j = 0; j < sample.labelNames.size(); j++) { + if (j != 0) { + w.write(", "); + } + w.write(sample.labelNames.get(j)); + w.write("=\""); + w.write(sample.labelValues.get(j)); + w.write('"'); + } + + w.write("} "); + w.write(Collector.doubleToGoString(sample.value)); + w.write('\n'); + } + } + } +} diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/SimpleGauge.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/SimpleGauge.java new file mode 100644 index 0000000000..3290d00ffd --- /dev/null +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/SimpleGauge.java @@ -0,0 +1,37 @@ +/** + * 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.stats; + +import org.apache.bookkeeper.stats.Gauge; + +/** + * A {@link Gauge} implementation that forwards on the value supplier. + */ +public class SimpleGauge { + + // public SimpleGauge(CollectorRegistry registry, String name) { + // this.gauge = PrometheusUtil.safeRegister(registry, + // Gauge.build().name(Collector.sanitizeMetricName(name)).help("-").create()); + // } + + private final Gauge gauge; + + public SimpleGauge(final Gauge gauge) { + this.gauge = gauge; + } + + public Number getSample() { + return gauge.getSample(); + } +} diff --git a/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/package-info.java b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/package-info.java new file mode 100644 index 0000000000..c2eddc6adb --- /dev/null +++ b/kafka-impl/src/main/java/io/streamnative/pulsar/handlers/kop/stats/package-info.java @@ -0,0 +1,19 @@ +/** + * 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. + */ +/** + * Timer related classes. + * + *

The classes under this package are ported from Kafka. + */ +package io.streamnative.pulsar.handlers.kop.stats; diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/EntryPublishTimeTest.java b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/EntryPublishTimeTest.java index c24a9b822e..054b5198fd 100644 --- a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/EntryPublishTimeTest.java +++ b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/EntryPublishTimeTest.java @@ -23,6 +23,7 @@ import io.streamnative.pulsar.handlers.kop.coordinator.transaction.TransactionCoordinator; import java.net.InetSocketAddress; import java.net.SocketAddress; +import org.apache.bookkeeper.stats.NullStatsLogger; import org.apache.pulsar.broker.protocol.ProtocolHandler; import org.apache.pulsar.common.policies.data.ClusterData; import org.apache.pulsar.common.policies.data.RetentionPolicies; @@ -90,7 +91,8 @@ protected void setup() throws Exception { groupCoordinator, transactionCoordinator, false, - getPlainEndPoint()); + getPlainEndPoint(), + NullStatsLogger.INSTANCE); ChannelHandlerContext mockCtx = mock(ChannelHandlerContext.class); Channel mockChannel = mock(Channel.class); doReturn(mockChannel).when(mockCtx).channel(); diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaApisTest.java b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaApisTest.java index eae817c433..233d797300 100644 --- a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaApisTest.java +++ b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaApisTest.java @@ -44,6 +44,7 @@ import java.util.stream.Collectors; import lombok.Cleanup; import lombok.extern.slf4j.Slf4j; +import org.apache.bookkeeper.stats.NullStatsLogger; import org.apache.commons.lang3.tuple.Pair; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; @@ -141,7 +142,8 @@ protected void setup() throws Exception { groupCoordinator, transactionCoordinator, false, - getPlainEndPoint()); + getPlainEndPoint(), + NullStatsLogger.INSTANCE); ChannelHandlerContext mockCtx = mock(ChannelHandlerContext.class); Channel mockChannel = mock(Channel.class); doReturn(mockChannel).when(mockCtx).channel(); diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandlerTest.java b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandlerTest.java index d82578b5d0..4264e591a3 100644 --- a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandlerTest.java +++ b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaRequestHandlerTest.java @@ -50,6 +50,7 @@ import lombok.Cleanup; import lombok.extern.slf4j.Slf4j; +import org.apache.bookkeeper.stats.NullStatsLogger; import org.apache.kafka.clients.admin.AdminClient; import org.apache.kafka.clients.admin.AdminClientConfig; import org.apache.kafka.clients.admin.NewTopic; @@ -146,7 +147,8 @@ protected void setup() throws Exception { groupCoordinator, transactionCoordinator, false, - getPlainEndPoint()); + getPlainEndPoint(), + NullStatsLogger.INSTANCE); } @AfterMethod diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaTopicConsumerManagerTest.java b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaTopicConsumerManagerTest.java index bb3ddb516a..0eca00c80e 100644 --- a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaTopicConsumerManagerTest.java +++ b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/KafkaTopicConsumerManagerTest.java @@ -31,6 +31,7 @@ import java.util.concurrent.atomic.AtomicLong; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.mledger.ManagedCursor; +import org.apache.bookkeeper.stats.NullStatsLogger; import org.apache.commons.lang3.tuple.Pair; import org.apache.kafka.clients.producer.KafkaProducer; import org.apache.kafka.clients.producer.ProducerConfig; @@ -69,7 +70,8 @@ protected void setup() throws Exception { groupCoordinator, transactionCoordinator, false, - getPlainEndPoint()); + getPlainEndPoint(), + NullStatsLogger.INSTANCE); ChannelHandlerContext mockCtx = mock(ChannelHandlerContext.class); Channel mockChannel = mock(Channel.class); diff --git a/tests/src/test/java/io/streamnative/pulsar/handlers/kop/MetricsProviderTest.java b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/MetricsProviderTest.java new file mode 100644 index 0000000000..332416f360 --- /dev/null +++ b/tests/src/test/java/io/streamnative/pulsar/handlers/kop/MetricsProviderTest.java @@ -0,0 +1,129 @@ +/** + * 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; + +import com.google.common.collect.Sets; +import java.io.BufferedReader; +import java.io.InputStream; +import java.io.InputStreamReader; +import lombok.extern.slf4j.Slf4j; +import org.apache.http.HttpResponse; +import org.apache.http.client.HttpClient; +import org.apache.http.client.methods.HttpGet; +import org.apache.http.impl.client.HttpClientBuilder; +import org.apache.kafka.clients.producer.ProducerRecord; +import org.apache.pulsar.common.policies.data.ClusterData; +import org.apache.pulsar.common.policies.data.RetentionPolicies; +import org.apache.pulsar.common.policies.data.TenantInfo; +import org.testng.Assert; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +/** + * test for kop prometheus metrics. + */ +@Slf4j +public class MetricsProviderTest extends KopProtocolHandlerTestBase{ + + @BeforeMethod + @Override + protected void setup() throws Exception { + super.internalSetup(); + log.info("success internal setup"); + + if (!admin.clusters().getClusters().contains(configClusterName)) { + // so that clients can test short names + admin.clusters().createCluster(configClusterName, + new ClusterData("http://127.0.0.1:" + brokerWebservicePort)); + } else { + admin.clusters().updateCluster(configClusterName, + new ClusterData("http://127.0.0.1:" + brokerWebservicePort)); + } + + if (!admin.tenants().getTenants().contains("public")) { + admin.tenants().createTenant("public", + new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test"))); + } else { + admin.tenants().updateTenant("public", + new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test"))); + } + if (!admin.namespaces().getNamespaces("public").contains("public/default")) { + admin.namespaces().createNamespace("public/default"); + admin.namespaces().setNamespaceReplicationClusters("public/default", Sets.newHashSet("test")); + admin.namespaces().setRetention("public/default", + new RetentionPolicies(60, 1000)); + } + if (!admin.namespaces().getNamespaces("public").contains("public/__kafka")) { + admin.namespaces().createNamespace("public/__kafka"); + admin.namespaces().setNamespaceReplicationClusters("public/__kafka", Sets.newHashSet("test")); + admin.namespaces().setRetention("public/__kafka", + new RetentionPolicies(-1, -1)); + } + } + + @AfterMethod + @Override + protected void cleanup() throws Exception { + super.internalCleanup(); + } + + + @Test(timeOut = 30000) + public void testMetricsProvider() throws Exception { + int partitionNumber = 3; + String kafkaTopicName = "kopKafkaProducePulsarMetrics" + partitionNumber; + + // create partitioned topic. + admin.topics().createPartitionedTopic(kafkaTopicName, partitionNumber); + + // 1. produce message with Kafka producer. + KProducer kProducer = new KProducer(kafkaTopicName, false, getKafkaBrokerPort()); + + int totalMsgs = 10; + + String messageStrPrefix = "Message_Kop_KafkaProducePulsarConsume_" + partitionNumber + "_"; + + for (int i = 0; i < totalMsgs; i++) { + String messageStr = messageStrPrefix + i; + ProducerRecord record = new ProducerRecord<>( + kafkaTopicName, + i, + messageStr); + + kProducer.getProducer().send(record).get(); + + if (log.isDebugEnabled()) { + log.debug("Kafka Producer Sent message: ({}, {})", i, messageStr); + } + } + + + HttpClient httpClient = HttpClientBuilder.create().build(); + final String metricsEndPoint = pulsar.getWebServiceAddress() + "/metrics"; + HttpResponse response = httpClient.execute(new HttpGet(metricsEndPoint)); + InputStream inputStream = response.getEntity().getContent(); + InputStreamReader isReader = new InputStreamReader(inputStream); + BufferedReader reader = new BufferedReader(isReader); + StringBuffer sb = new StringBuffer(); + String str; + while ((str = reader.readLine()) != null) { + sb.append(str); + } + Assert.assertTrue(sb.toString().contains("kop_server_HANDLE_PRODUCE_REQUEST")); + Assert.assertTrue(sb.toString().contains("kop_server_PRODUCE_ENCODE")); + Assert.assertTrue(sb.toString().contains("kop_server_MESSAGE_PUBLISH")); + Assert.assertTrue(sb.toString().contains("kop_server_MESSAGE_QUEUED_LATENCY")); + } +}