diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java index e2bcf03353b59..970f28bd6fa18 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java @@ -366,7 +366,7 @@ The delayed message index time step(in seconds) in per bucket snapshot segment, private int delayedDeliveryMaxTimeStepPerBucketSnapshotSegmentSeconds = 300; @FieldContext(category = CATEGORY_SERVER, doc = """ - The max number of delayed message index in per bucket snapshot segment, -1 means no limitation\ + The max number of delayed message index in per bucket snapshot segment, -1 means no limitation, \ after reaching the max number limitation, the snapshot segment will be cut off.""") private int delayedDeliveryMaxIndexesPerBucketSnapshotSegment = 5000; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTracker.java index 3cc2da8db1e4d..78229fef25a5a 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTracker.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/DelayedDeliveryTracker.java @@ -67,12 +67,6 @@ public interface DelayedDeliveryTracker extends AutoCloseable { */ boolean shouldPauseAllDeliveries(); - /** - * Tells whether this DelayedDeliveryTracker contains this message index, - * if the tracker is not supported it or disabled this feature also will return false. - */ - boolean containsMessage(long ledgerId, long entryId); - /** * Reset tick time use zk policies cache. * @param tickTime diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java index 8de6ee58e2ce5..58358b06a46bb 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java @@ -178,11 +178,6 @@ && getNumberOfDelayedMessages() >= fixedDelayDetectionLookahead && !hasMessageAvailable(); } - @Override - public boolean containsMessage(long ledgerId, long entryId) { - return false; - } - protected long nextDeliveryTime() { return priorityQueue.peekN1(); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java index 08202bb19155d..e7d4f9301dd36 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BookkeeperBucketSnapshotStorage.java @@ -67,7 +67,7 @@ public CompletableFuture createBucketSnapshot(SnapshotMetadata snapshotMet @Override public CompletableFuture getBucketSnapshotMetadata(long bucketId) { return openLedger(bucketId).thenCompose( - ledgerHandle -> getLedgerEntryThenCloseLedger(ledgerHandle, 0, 0). + ledgerHandle -> getLedgerEntry(ledgerHandle, 0, 0). thenApply(entryEnumeration -> parseSnapshotMetadataEntry(entryEnumeration.nextElement()))); } @@ -75,17 +75,13 @@ public CompletableFuture getBucketSnapshotMetadata(long bucket public CompletableFuture> getBucketSnapshotSegment(long bucketId, long firstSegmentEntryId, long lastSegmentEntryId) { return openLedger(bucketId).thenCompose( - ledgerHandle -> getLedgerEntryThenCloseLedger(ledgerHandle, firstSegmentEntryId, + ledgerHandle -> getLedgerEntry(ledgerHandle, firstSegmentEntryId, lastSegmentEntryId).thenApply(this::parseSnapshotSegmentEntries)); } @Override public CompletableFuture getBucketSnapshotLength(long bucketId) { - return openLedger(bucketId).thenApply(ledgerHandle -> { - long length = ledgerHandle.getLength(); - closeLedger(ledgerHandle); - return length; - }); + return openLedger(bucketId).thenApply(LedgerHandle::getLength); } @Override @@ -212,8 +208,8 @@ private CompletableFuture addEntry(LedgerHandle ledgerHandle, byte[] data) }); } - CompletableFuture> getLedgerEntryThenCloseLedger(LedgerHandle ledger, - long firstEntryId, long lastEntryId) { + CompletableFuture> getLedgerEntry(LedgerHandle ledger, + long firstEntryId, long lastEntryId) { final CompletableFuture> future = new CompletableFuture<>(); ledger.asyncReadEntries(firstEntryId, lastEntryId, (rc, handle, entries, ctx) -> { @@ -222,7 +218,6 @@ CompletableFuture> getLedgerEntryThenCloseLedger(Ledger } else { future.complete(entries); } - closeLedger(handle); }, null ); return future; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java index ef7be187cec3d..a34bd51af98e4 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTracker.java @@ -57,6 +57,7 @@ import org.apache.pulsar.broker.delayed.proto.DelayedMessageIndexBucketSnapshotFormat; import org.apache.pulsar.broker.delayed.proto.DelayedMessageIndexBucketSnapshotFormat.DelayedIndex; import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers; +import org.apache.pulsar.common.policies.data.stats.TopicMetricBean; import org.apache.pulsar.common.util.FutureUtil; import org.apache.pulsar.common.util.collections.TripleLongPriorityQueue; import org.roaringbitmap.RoaringBitmap; @@ -69,6 +70,10 @@ public class BucketDelayedDeliveryTracker extends AbstractDelayedDeliveryTracker static final int AsyncOperationTimeoutSeconds = 60; + private static final Long INVALID_BUCKET_ID = -1L; + + private static final int MAX_MERGE_NUM = 4; + private final long minIndexCountPerBucket; private final long timeStepPerBucketSnapshotSegmentInMillis; @@ -93,9 +98,7 @@ public class BucketDelayedDeliveryTracker extends AbstractDelayedDeliveryTracker private final Table snapshotSegmentLastIndexTable; - private static final Long INVALID_BUCKET_ID = -1L; - - private static final int MAX_MERGE_NUM = 4; + private final BucketDelayedMessageIndexStats stats; public BucketDelayedDeliveryTracker(PersistentDispatcherMultipleConsumers dispatcher, Timer timer, long tickTimeMillis, @@ -125,6 +128,7 @@ public BucketDelayedDeliveryTracker(PersistentDispatcherMultipleConsumers dispat this.lastMutableBucket = new MutableBucket(dispatcher.getName(), dispatcher.getCursor(), FutureUtil.Sequencer.create(), bucketSnapshotStorage); + this.stats = new BucketDelayedMessageIndexStats(); this.numberDelayedMessages = recoverBucketSnapshot(); } @@ -161,8 +165,9 @@ private synchronized long recoverBucketSnapshot() throws RuntimeException { } try { - FutureUtil.waitForAll(futures.values()).get(AsyncOperationTimeoutSeconds, TimeUnit.SECONDS); + FutureUtil.waitForAll(futures.values()).get(AsyncOperationTimeoutSeconds * 2, TimeUnit.SECONDS); } catch (InterruptedException | ExecutionException | TimeoutException e) { + log.error("[{}] Failed to recover delayed message index bucket snapshot.", dispatcher.getName(), e); if (e instanceof InterruptedException) { Thread.currentThread().interrupt(); } @@ -193,7 +198,7 @@ private synchronized long recoverBucketSnapshot() throws RuntimeException { ImmutableBucket immutableBucket = mapEntry.getValue(); immutableBucketMap.remove(key); // delete asynchronously without waiting for completion - immutableBucket.asyncDeleteBucketSnapshot(); + immutableBucket.asyncDeleteBucketSnapshot(stats); } MutableLong numberDelayedMessages = new MutableLong(0); @@ -246,7 +251,8 @@ private Optional findImmutableBucket(long ledgerId) { return Optional.ofNullable(immutableBuckets.get(ledgerId)); } - private void afterCreateImmutableBucket(Pair immutableBucketDelayedIndexPair) { + private void afterCreateImmutableBucket(Pair immutableBucketDelayedIndexPair, + long startTime) { if (immutableBucketDelayedIndexPair != null) { ImmutableBucket immutableBucket = immutableBucketDelayedIndexPair.getLeft(); immutableBuckets.put(Range.closed(immutableBucket.startLedgerId, immutableBucket.endLedgerId), @@ -260,14 +266,19 @@ private void afterCreateImmutableBucket(Pair immu CompletableFuture future = createFuture.handle((bucketId, ex) -> { if (ex == null) { immutableBucket.setSnapshotSegments(null); + immutableBucket.asyncUpdateSnapshotLength(); log.info("[{}] Creat bucket snapshot finish, bucketKey: {}", dispatcher.getName(), immutableBucket.bucketKey()); + + stats.recordSuccessEvent(BucketDelayedMessageIndexStats.Type.create, + System.currentTimeMillis() - startTime); + return bucketId; } - //TODO Record create snapshot failed - log.error("[{}] Failed to create bucket snapshot, bucketKey: {}", - dispatcher.getName(), immutableBucket.bucketKey(), ex); + log.error("[{}] Failed to create bucket snapshot, bucketKey: {}", dispatcher.getName(), + immutableBucket.bucketKey(), ex); + stats.recordFailEvent(BucketDelayedMessageIndexStats.Type.create); // Put indexes back into the shared queue and downgrade to memory mode synchronized (BucketDelayedDeliveryTracker.this) { @@ -311,12 +322,14 @@ public synchronized boolean addMessage(long ledgerId, long entryId, long deliver if (!existBucket && ledgerId > lastMutableBucket.endLedgerId && lastMutableBucket.size() >= minIndexCountPerBucket && !lastMutableBucket.isEmpty()) { + long createStartTime = System.currentTimeMillis(); + stats.recordTriggerEvent(BucketDelayedMessageIndexStats.Type.create); Pair immutableBucketDelayedIndexPair = lastMutableBucket.sealBucketAndAsyncPersistent( this.timeStepPerBucketSnapshotSegmentInMillis, this.maxIndexesPerBucketSnapshotSegment, this.sharedBucketPriorityQueue); - afterCreateImmutableBucket(immutableBucketDelayedIndexPair); + afterCreateImmutableBucket(immutableBucketDelayedIndexPair, createStartTime); lastMutableBucket.resetLastMutableBucketRange(); if (immutableBuckets.asMapOfRanges().size() > maxNumBuckets) { @@ -374,7 +387,7 @@ private synchronized List selectMergedBuckets(final List= 0) { - return values.subList(minIndex, minIndex + MAX_MERGE_NUM); + return values.subList(minIndex, minIndex + mergeNum); } else if (mergeNum > 2){ return selectMergedBuckets(values, mergeNum - 1); } else { @@ -400,6 +413,9 @@ private synchronized CompletableFuture asyncMergeBucketSnapshot() { for (ImmutableBucket immutableBucket : toBeMergeImmutableBuckets) { immutableBucket.merging = true; } + + long mergeStartTime = System.currentTimeMillis(); + stats.recordTriggerEvent(BucketDelayedMessageIndexStats.Type.merge); return asyncMergeBucketSnapshot(toBeMergeImmutableBuckets).whenComplete((__, ex) -> { synchronized (this) { for (ImmutableBucket immutableBucket : toBeMergeImmutableBuckets) { @@ -409,9 +425,14 @@ private synchronized CompletableFuture asyncMergeBucketSnapshot() { if (ex != null) { log.error("[{}] Failed to merge bucket snapshot, bucketKeys: {}", dispatcher.getName(), bucketsStr, ex); + + stats.recordFailEvent(BucketDelayedMessageIndexStats.Type.merge); } else { log.info("[{}] Merge bucket snapshot finish, bucketKeys: {}, bucketNum: {}", dispatcher.getName(), bucketsStr, immutableBuckets.asMapOfRanges().size()); + + stats.recordSuccessEvent(BucketDelayedMessageIndexStats.Type.merge, + System.currentTimeMillis() - mergeStartTime); } }); } @@ -436,6 +457,8 @@ private synchronized CompletableFuture asyncMergeBucketSnapshot(List { synchronized (BucketDelayedDeliveryTracker.this) { + long createStartTime = System.currentTimeMillis(); + stats.recordTriggerEvent(BucketDelayedMessageIndexStats.Type.create); Pair immutableBucketDelayedIndexPair = lastMutableBucket.createImmutableBucketAndAsyncPersistent( timeStepPerBucketSnapshotSegmentInMillis, @@ -461,12 +484,12 @@ private synchronized CompletableFuture asyncMergeBucketSnapshot(List { List> removeFutures = - buckets.stream().map(ImmutableBucket::asyncDeleteBucketSnapshot) + buckets.stream().map(bucket -> bucket.asyncDeleteBucketSnapshot(stats)) .toList(); return FutureUtil.waitForAll(removeFutures); }); @@ -557,15 +580,17 @@ public synchronized NavigableSet getScheduledMessages(int maxMessa if (bucket.currentSegmentEntryId == bucket.lastSegmentEntryId) { immutableBuckets.asMapOfRanges().remove(Range.closed(bucket.startLedgerId, bucket.endLedgerId)); - bucket.asyncDeleteBucketSnapshot(); + bucket.asyncDeleteBucketSnapshot(stats); continue; } + long loadStartTime = System.currentTimeMillis(); + stats.recordTriggerEvent(BucketDelayedMessageIndexStats.Type.load); bucket.asyncLoadNextBucketSnapshotEntry().thenAccept(indexList -> { if (CollectionUtils.isEmpty(indexList)) { immutableBuckets.asMapOfRanges() .remove(Range.closed(bucket.startLedgerId, bucket.endLedgerId)); - bucket.asyncDeleteBucketSnapshot(); + bucket.asyncDeleteBucketSnapshot(stats); return; } DelayedMessageIndexBucketSnapshotFormat.DelayedIndex @@ -583,9 +608,14 @@ public synchronized NavigableSet getScheduledMessages(int maxMessa log.error("[{}] Failed to load bucket snapshot segment, bucketKey: {}, segmentEntryId: {}", dispatcher.getName(), bucket.bucketKey(), preSegmentEntryId + 1, ex); + + stats.recordFailEvent(BucketDelayedMessageIndexStats.Type.load); } else { log.info("[{}] Load next bucket snapshot segment finish, bucketKey: {}, segmentEntryId: {}", dispatcher.getName(), bucket.bucketKey(), preSegmentEntryId + 1); + + stats.recordSuccessEvent(BucketDelayedMessageIndexStats.Type.load, + System.currentTimeMillis() - loadStartTime); } }).get(AsyncOperationTimeoutSeconds * (MaxRetryTimes + 1), TimeUnit.SECONDS); } catch (Exception e) { @@ -645,7 +675,7 @@ private CompletableFuture cleanImmutableBuckets() { Iterator iterator = immutableBuckets.asMapOfRanges().values().iterator(); while (iterator.hasNext()) { ImmutableBucket bucket = iterator.next(); - futures.add(bucket.clear()); + futures.add(bucket.clear(stats)); numberDelayedMessages -= bucket.getNumberBucketDelayedMessages(); iterator.remove(); } @@ -661,7 +691,6 @@ private boolean removeIndexBit(long ledgerId, long entryId) { .orElse(false); } - @Override public boolean containsMessage(long ledgerId, long entryId) { if (lastMutableBucket.containsMessage(ledgerId, entryId)) { return true; @@ -670,4 +699,15 @@ public boolean containsMessage(long ledgerId, long entryId) { return findImmutableBucket(ledgerId).map(bucket -> bucket.containsMessage(ledgerId, entryId)) .orElse(false); } + + public Map genTopicMetricMap() { + stats.recordNumOfBuckets(immutableBuckets.asMapOfRanges().size() + 1); + stats.recordDelayedMessageIndexLoaded(this.sharedBucketPriorityQueue.size() + this.lastMutableBucket.size()); + MutableLong totalSnapshotLength = new MutableLong(); + immutableBuckets.asMapOfRanges().values().forEach(immutableBucket -> { + totalSnapshotLength.add(immutableBucket.getSnapshotLength()); + }); + stats.recordBucketSnapshotSizeBytes(totalSnapshotLength.longValue()); + return stats.genTopicMetricMap(); + } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedMessageIndexStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedMessageIndexStats.java new file mode 100644 index 0000000000000..68788c359d560 --- /dev/null +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedMessageIndexStats.java @@ -0,0 +1,146 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.pulsar.broker.delayed.bucket; + +import java.util.HashMap; +import java.util.Map; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; +import java.util.concurrent.atomic.LongAdder; +import org.apache.bookkeeper.mledger.util.StatsBuckets; +import org.apache.pulsar.common.policies.data.stats.TopicMetricBean; + +public class BucketDelayedMessageIndexStats { + + private static final long[] BUCKETS = new long[]{50, 100, 500, 1000, 5000, 30000, 60000}; + + enum State { + succeed, + failed, + all + } + + enum Type { + create, + load, + delete, + merge + } + + private static final String BUCKET_TOTAL_NAME = "pulsar_delayed_message_index_bucket_total"; + private static final String INDEX_LOADED_NAME = "pulsar_delayed_message_index_loaded"; + private static final String SNAPSHOT_SIZE_BYTES_NAME = "pulsar_delayed_message_index_bucket_snapshot_size_bytes"; + private static final String OP_COUNT_NAME = "pulsar_delayed_message_index_bucket_op_count"; + private static final String OP_LATENCY_NAME = "pulsar_delayed_message_index_bucket_op_latency_ms"; + + private final AtomicInteger delayedMessageIndexBucketTotal = new AtomicInteger(); + private final AtomicLong delayedMessageIndexLoaded = new AtomicLong(); + private final AtomicLong delayedMessageIndexBucketSnapshotSizeBytes = new AtomicLong(); + private final Map delayedMessageIndexBucketOpLatencyMs = new ConcurrentHashMap<>(); + private final Map delayedMessageIndexBucketOpCount = new ConcurrentHashMap<>(); + + public BucketDelayedMessageIndexStats() { + } + + public Map genTopicMetricMap() { + Map metrics = new HashMap<>(); + + metrics.put(BUCKET_TOTAL_NAME, + new TopicMetricBean(BUCKET_TOTAL_NAME, delayedMessageIndexBucketTotal.get(), null)); + + metrics.put(INDEX_LOADED_NAME, + new TopicMetricBean(INDEX_LOADED_NAME, delayedMessageIndexLoaded.get(), null)); + + metrics.put(SNAPSHOT_SIZE_BYTES_NAME, + new TopicMetricBean(SNAPSHOT_SIZE_BYTES_NAME, delayedMessageIndexBucketSnapshotSizeBytes.get(), null)); + + delayedMessageIndexBucketOpCount.forEach((k, count) -> { + String[] labels = splitKey(k); + String[] labelsAndValues = new String[] {"state", labels[0], "type", labels[1]}; + String key = OP_COUNT_NAME + joinKey(labelsAndValues); + metrics.put(key, new TopicMetricBean(OP_COUNT_NAME, count.sumThenReset(), labelsAndValues)); + }); + + delayedMessageIndexBucketOpLatencyMs.forEach((typeName, statsBuckets) -> { + statsBuckets.refresh(); + long[] buckets = statsBuckets.getBuckets(); + for (int i = 0; i < buckets.length; i++) { + long count = buckets[i]; + if (count == 0L) { + continue; + } + String quantile; + if (i == BUCKETS.length) { + quantile = "overflow"; + } else { + quantile = String.valueOf(BUCKETS[i]); + } + String[] labelsAndValues = new String[] {"type", typeName, "quantile", quantile}; + String key = OP_LATENCY_NAME + joinKey(labelsAndValues); + + metrics.put(key, new TopicMetricBean(OP_LATENCY_NAME, count, labelsAndValues)); + } + String[] labelsAndValues = new String[] {"type", typeName}; + metrics.put(OP_LATENCY_NAME + "_count" + joinKey(labelsAndValues), + new TopicMetricBean(OP_LATENCY_NAME + "_count", statsBuckets.getCount(), labelsAndValues)); + metrics.put(OP_LATENCY_NAME + "_sum" + joinKey(labelsAndValues), + new TopicMetricBean(OP_LATENCY_NAME + "_sum", statsBuckets.getSum(), labelsAndValues)); + }); + + return metrics; + } + + public void recordNumOfBuckets(int numOfBuckets) { + delayedMessageIndexBucketTotal.set(numOfBuckets); + } + + public void recordDelayedMessageIndexLoaded(long num) { + delayedMessageIndexLoaded.set(num); + } + + public void recordBucketSnapshotSizeBytes(long sizeBytes) { + delayedMessageIndexBucketSnapshotSizeBytes.set(sizeBytes); + } + + public void recordTriggerEvent(Type eventType) { + delayedMessageIndexBucketOpCount.computeIfAbsent(joinKey(State.all.name(), eventType.name()), + k -> new LongAdder()).increment(); + } + + public void recordSuccessEvent(Type eventType, long cost) { + delayedMessageIndexBucketOpCount.computeIfAbsent(joinKey(State.succeed.name(), eventType.name()), + k -> new LongAdder()).increment(); + delayedMessageIndexBucketOpLatencyMs.computeIfAbsent(eventType.name(), + k -> new StatsBuckets(BUCKETS)).addValue(cost); + } + + public void recordFailEvent(Type eventType) { + delayedMessageIndexBucketOpCount.computeIfAbsent(joinKey(State.failed.name(), eventType.name()), + k -> new LongAdder()).increment(); + } + + public static String joinKey(String... values) { + return String.join("_", values); + } + + public static String[] splitKey(String key) { + return key.split("_"); + } +} diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/ImmutableBucket.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/ImmutableBucket.java index 969d326e28187..82e98cefa5d98 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/ImmutableBucket.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/ImmutableBucket.java @@ -131,6 +131,9 @@ private CompletableFuture> asyncLoadNextBucketSnapshotEntry(b List indexList = snapshotSegment.getIndexesList(); this.setCurrentSegmentEntryId(nextSegmentEntryId); + if (isRecover) { + this.asyncUpdateSnapshotLength(); + } return indexList; }); }); @@ -175,7 +178,9 @@ CompletableFuture> }, BucketSnapshotPersistenceException.class, MaxRetryTimes); } - CompletableFuture asyncDeleteBucketSnapshot() { + CompletableFuture asyncDeleteBucketSnapshot(BucketDelayedMessageIndexStats stats) { + long deleteStartTime = System.currentTimeMillis(); + stats.recordTriggerEvent(BucketDelayedMessageIndexStats.Type.delete); String bucketKey = bucketKey(); long bucketId = getAndUpdateBucketId(); return removeBucketCursorProperty(bucketKey).thenCompose(__ -> @@ -184,16 +189,33 @@ CompletableFuture asyncDeleteBucketSnapshot() { if (ex != null) { log.error("[{}] Failed to delete bucket snapshot, bucketId: {}, bucketKey: {}", dispatcherName, bucketId, bucketKey, ex); + + stats.recordFailEvent(BucketDelayedMessageIndexStats.Type.delete); } else { log.info("[{}] Delete bucket snapshot finish, bucketId: {}, bucketKey: {}", dispatcherName, bucketId, bucketKey); + + stats.recordSuccessEvent(BucketDelayedMessageIndexStats.Type.delete, + System.currentTimeMillis() - deleteStartTime); } }); } - CompletableFuture clear() { + CompletableFuture clear(BucketDelayedMessageIndexStats stats) { delayedIndexBitMap.clear(); return getSnapshotCreateFuture().orElse(NULL_LONG_PROMISE).exceptionally(e -> null) - .thenCompose(__ -> asyncDeleteBucketSnapshot()); + .thenCompose(__ -> asyncDeleteBucketSnapshot(stats)); + } + + protected CompletableFuture asyncUpdateSnapshotLength() { + long bucketId = getAndUpdateBucketId(); + return bucketSnapshotStorage.getBucketSnapshotLength(bucketId).whenComplete((length, ex) -> { + if (ex != null) { + log.error("[{}] Failed to get snapshot length, bucketId: {}, bucketKey: {}", + dispatcherName, bucketId, bucketKey(), ex); + } else { + setSnapshotLength(length); + } + }); } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/MutableBucket.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/MutableBucket.java index f8a4ecc7a4ddb..e49ebe9606e01 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/MutableBucket.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/delayed/bucket/MutableBucket.java @@ -62,8 +62,10 @@ Pair createImmutableBucketAndAsyncPersistent( final long timeStepPerBucketSnapshotSegment, final int maxIndexesPerBucketSnapshotSegment, TripleLongPriorityQueue sharedQueue, DelayedIndexQueue delayedIndexQueue, final long startLedgerId, final long endLedgerId) { - log.info("[{}] Creating bucket snapshot, startLedgerId: {}, endLedgerId: {}", dispatcherName, - startLedgerId, endLedgerId); + if (log.isDebugEnabled()) { + log.debug("[{}] Creating bucket snapshot, startLedgerId: {}, endLedgerId: {}", dispatcherName, + startLedgerId, endLedgerId); + } if (delayedIndexQueue.isEmpty()) { return null; diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java index e5c9e85bac3f2..7ff6e72d02aed 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java @@ -72,6 +72,7 @@ import org.apache.pulsar.client.impl.Backoff; import org.apache.pulsar.common.api.proto.CommandSubscribe.SubType; import org.apache.pulsar.common.api.proto.MessageMetadata; +import org.apache.pulsar.common.policies.data.stats.TopicMetricBean; import org.apache.pulsar.common.protocol.Commands; import org.apache.pulsar.common.util.Codec; import org.slf4j.Logger; @@ -332,7 +333,7 @@ public synchronized void readMoreEntries() { Predicate skipCondition = null; final DelayedDeliveryTracker deliveryTracker = delayedDeliveryTracker.get(); if (deliveryTracker instanceof BucketDelayedDeliveryTracker) { - skipCondition = position -> deliveryTracker + skipCondition = position -> ((BucketDelayedDeliveryTracker) deliveryTracker) .containsMessage(position.getLedgerId(), position.getEntryId()); } cursor.asyncReadEntriesWithSkipOrWait(messagesToRead, bytesToRead, this, ReadType.Normal, @@ -1180,6 +1181,18 @@ public long getDelayedTrackerMemoryUsage() { return 0; } + public Map getBucketDelayedIndexStats() { + if (delayedDeliveryTracker.isEmpty()) { + return Collections.emptyMap(); + } + + if (delayedDeliveryTracker.get() instanceof BucketDelayedDeliveryTracker) { + return ((BucketDelayedDeliveryTracker) delayedDeliveryTracker.get()).genTopicMetricMap(); + } + + return Collections.emptyMap(); + } + public ManagedCursor getCursor() { return cursor; } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java index bdb3c9fc391ce..4ed191a9b4f61 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java @@ -1147,6 +1147,9 @@ public SubscriptionStatsImpl getStats(Boolean getPreciseBacklog, boolean subscri if (dispatcher instanceof PersistentDispatcherMultipleConsumers) { subStats.delayedMessageIndexSizeInBytes = ((PersistentDispatcherMultipleConsumers) dispatcher).getDelayedTrackerMemoryUsage(); + + subStats.bucketDelayedIndexStats = + ((PersistentDispatcherMultipleConsumers) dispatcher).getBucketDelayedIndexStats(); } if (Subscription.isIndividualAckMode(subType)) { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index 82a4f5312357a..fa08330ff3c35 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -153,6 +153,7 @@ import org.apache.pulsar.common.policies.data.stats.PublisherStatsImpl; import org.apache.pulsar.common.policies.data.stats.ReplicatorStatsImpl; import org.apache.pulsar.common.policies.data.stats.SubscriptionStatsImpl; +import org.apache.pulsar.common.policies.data.stats.TopicMetricBean; import org.apache.pulsar.common.policies.data.stats.TopicStatsImpl; import org.apache.pulsar.common.protocol.Commands; import org.apache.pulsar.common.protocol.schema.SchemaData; @@ -2183,6 +2184,14 @@ public CompletableFuture asyncGetStats(boolean getPreciseBacklog stats.nonContiguousDeletedMessagesRangesSerializedSize += subStats.nonContiguousDeletedMessagesRangesSerializedSize; stats.delayedMessageIndexSizeInBytes += subStats.delayedMessageIndexSizeInBytes; + + subStats.bucketDelayedIndexStats.forEach((k, v) -> { + TopicMetricBean topicMetricBean = + stats.bucketDelayedIndexStats.computeIfAbsent(k, __ -> new TopicMetricBean()); + topicMetricBean.name = v.name; + topicMetricBean.labelsAndValues = v.labelsAndValues; + topicMetricBean.value += v.value; + }); }); replicators.forEach((cluster, replicator) -> { diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedNamespaceStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedNamespaceStats.java index 0a905daa341f3..ea77bd69302a0 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedNamespaceStats.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedNamespaceStats.java @@ -21,6 +21,7 @@ import java.util.HashMap; import java.util.Map; import org.apache.bookkeeper.mledger.util.StatsBuckets; +import org.apache.pulsar.common.policies.data.stats.TopicMetricBean; import org.apache.pulsar.compaction.CompactionRecord; public class AggregatedNamespaceStats { @@ -65,6 +66,8 @@ public class AggregatedNamespaceStats { StatsBuckets compactionLatencyBuckets = new StatsBuckets(CompactionRecord.WRITE_LATENCY_BUCKETS_USEC); int delayedMessageIndexSizeInBytes; + Map bucketDelayedIndexStats = new HashMap<>(); + void updateStats(TopicStats stats) { topicsCount++; @@ -83,6 +86,14 @@ void updateStats(TopicStats stats) { msgOutCounter += stats.msgOutCounter; delayedMessageIndexSizeInBytes += stats.delayedMessageIndexSizeInBytes; + stats.bucketDelayedIndexStats.forEach((k, v) -> { + TopicMetricBean topicMetricBean = + bucketDelayedIndexStats.computeIfAbsent(k, __ -> new TopicMetricBean()); + topicMetricBean.name = v.name; + topicMetricBean.labelsAndValues = v.labelsAndValues; + topicMetricBean.value += v.value; + }); + this.ongoingTxnCount += stats.ongoingTxnCount; this.abortedTxnCount += stats.abortedTxnCount; this.committedTxnCount += stats.committedTxnCount; @@ -132,6 +143,13 @@ void updateStats(TopicStats stats) { subsStats.filterRejectedMsgCount += as.filterRejectedMsgCount; subsStats.filterRescheduledMsgCount += as.filterRescheduledMsgCount; subsStats.delayedMessageIndexSizeInBytes += as.delayedMessageIndexSizeInBytes; + as.bucketDelayedIndexStats.forEach((k, v) -> { + TopicMetricBean topicMetricBean = + subsStats.bucketDelayedIndexStats.computeIfAbsent(k, __ -> new TopicMetricBean()); + topicMetricBean.name = v.name; + topicMetricBean.labelsAndValues = v.labelsAndValues; + topicMetricBean.value += v.value; + }); as.consumerStat.forEach((c, v) -> { AggregatedConsumerStats consumerStats = subsStats.consumerStat.computeIfAbsent(c, k -> new AggregatedConsumerStats()); @@ -172,5 +190,6 @@ public void reset() { replicationStats.clear(); subscriptionStats.clear(); delayedMessageIndexSizeInBytes = 0; + bucketDelayedIndexStats.clear(); } } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedSubscriptionStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedSubscriptionStats.java index 383c671754dc1..da0324c55655c 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedSubscriptionStats.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedSubscriptionStats.java @@ -21,6 +21,7 @@ import java.util.HashMap; import java.util.Map; import org.apache.pulsar.broker.service.Consumer; +import org.apache.pulsar.common.policies.data.stats.TopicMetricBean; public class AggregatedSubscriptionStats { @@ -75,4 +76,6 @@ public class AggregatedSubscriptionStats { public Map consumerStat = new HashMap<>(); long delayedMessageIndexSizeInBytes; + + public Map bucketDelayedIndexStats = new HashMap<>(); } diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java index 918aef539cff4..32fb06ea3ce8c 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java @@ -28,6 +28,7 @@ import org.apache.bookkeeper.client.LedgerHandle; import org.apache.bookkeeper.mledger.ManagedLedger; import org.apache.bookkeeper.mledger.impl.ManagedLedgerMBeanImpl; +import org.apache.commons.lang3.ArrayUtils; import org.apache.pulsar.broker.PulsarService; import org.apache.pulsar.broker.service.Topic; import org.apache.pulsar.broker.service.persistent.PersistentTopic; @@ -155,6 +156,7 @@ private static void aggregateTopicStats(TopicStats stats, SubscriptionStatsImpl subsStats.filterRejectedMsgCount = subscriptionStats.filterRejectedMsgCount; subsStats.filterRescheduledMsgCount = subscriptionStats.filterRescheduledMsgCount; subsStats.delayedMessageIndexSizeInBytes = subscriptionStats.delayedMessageIndexSizeInBytes; + subsStats.bucketDelayedIndexStats = subscriptionStats.bucketDelayedIndexStats; } private static void getTopicStats(Topic topic, TopicStats stats, boolean includeConsumerMetrics, @@ -197,6 +199,7 @@ private static void getTopicStats(Topic topic, TopicStats stats, boolean include stats.averageMsgSize = tStatus.averageMsgSize; stats.publishRateLimitedTimes = tStatus.publishRateLimitedTimes; stats.delayedMessageIndexSizeInBytes = tStatus.delayedMessageIndexSizeInBytes; + stats.bucketDelayedIndexStats = tStatus.bucketDelayedIndexStats; stats.abortedTxnCount = tStatus.abortedTxnCount; stats.ongoingTxnCount = tStatus.ongoingTxnCount; stats.committedTxnCount = tStatus.committedTxnCount; @@ -379,6 +382,10 @@ private static void printNamespaceStats(PrometheusMetricStreams stream, Aggregat writeMetric(stream, "pulsar_delayed_message_index_size_bytes", stats.delayedMessageIndexSizeInBytes, cluster, namespace); + stats.bucketDelayedIndexStats.forEach((k, metric) -> { + writeMetric(stream, metric.name, metric.value, cluster, namespace, metric.labelsAndValues); + }); + writePulsarMsgBacklog(stream, stats.msgBacklog, cluster, namespace); stats.managedLedgerStats.storageWriteLatencyBuckets.refresh(); @@ -472,8 +479,10 @@ private static void writeMetric(PrometheusMetricStreams stream, String metricNam } private static void writeMetric(PrometheusMetricStreams stream, String metricName, Number value, String cluster, - String namespace) { - stream.writeSample(metricName, value, "cluster", cluster, "namespace", namespace); + String namespace, String... extraLabelsAndValues) { + String[] labelsAndValues = new String[]{"cluster", cluster, "namespace", namespace}; + String[] labels = ArrayUtils.addAll(labelsAndValues, extraLabelsAndValues); + stream.writeSample(metricName, value, labels); } private static void writeReplicationStat(PrometheusMetricStreams stream, String metricName, diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/TopicStats.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/TopicStats.java index abc6979484e58..3a2563a87587b 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/TopicStats.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/prometheus/TopicStats.java @@ -25,6 +25,7 @@ import org.apache.bookkeeper.mledger.util.StatsBuckets; import org.apache.commons.lang3.ArrayUtils; import org.apache.pulsar.broker.service.Consumer; +import org.apache.pulsar.common.policies.data.stats.TopicMetricBean; import org.apache.pulsar.compaction.CompactionRecord; import org.apache.pulsar.compaction.CompactorMXBean; @@ -70,6 +71,8 @@ class TopicStats { StatsBuckets compactionLatencyBuckets = new StatsBuckets(CompactionRecord.WRITE_LATENCY_BUCKETS_USEC); public long delayedMessageIndexSizeInBytes; + Map bucketDelayedIndexStats = new HashMap<>(); + public void reset() { subscriptionsCount = 0; producersCount = 0; @@ -107,6 +110,7 @@ public void reset() { compactionCompactedEntriesSize = 0; compactionLatencyBuckets.reset(); delayedMessageIndexSizeInBytes = 0; + bucketDelayedIndexStats.clear(); } public static void printTopicStats(PrometheusMetricStreams stream, TopicStats stats, @@ -162,6 +166,11 @@ public static void printTopicStats(PrometheusMetricStreams stream, TopicStats st writeMetric(stream, "pulsar_delayed_message_index_size_bytes", stats.delayedMessageIndexSizeInBytes, cluster, namespace, topic, splitTopicAndPartitionIndexLabel); + for (TopicMetricBean topicMetricBean : stats.bucketDelayedIndexStats.values()) { + writeTopicMetric(stream, topicMetricBean.name, topicMetricBean.value, cluster, namespace, + topic, splitTopicAndPartitionIndexLabel, topicMetricBean.labelsAndValues); + } + long[] latencyBuckets = stats.managedLedgerStats.storageWriteLatencyBuckets.getBuckets(); writeMetric(stream, "pulsar_storage_write_latency_le_0_5", latencyBuckets[0], cluster, namespace, topic, splitTopicAndPartitionIndexLabel); @@ -310,6 +319,13 @@ public static void printTopicStats(PrometheusMetricStreams stream, TopicStats st subsStats.delayedMessageIndexSizeInBytes, cluster, namespace, topic, sub, splitTopicAndPartitionIndexLabel); + final String[] subscriptionLabel = {"subscription", sub}; + for (TopicMetricBean topicMetricBean : subsStats.bucketDelayedIndexStats.values()) { + String[] labelsAndValues = ArrayUtils.addAll(subscriptionLabel, topicMetricBean.labelsAndValues); + writeTopicMetric(stream, topicMetricBean.name, topicMetricBean.value, cluster, namespace, + topic, splitTopicAndPartitionIndexLabel, labelsAndValues); + } + subsStats.consumerStat.forEach((c, consumerStats) -> { writeConsumerMetric(stream, "pulsar_consumer_msg_rate_redeliver", consumerStats.msgRateRedeliver, cluster, namespace, topic, sub, c, splitTopicAndPartitionIndexLabel); @@ -409,6 +425,12 @@ public static void printTopicStats(PrometheusMetricStreams stream, TopicStats st writeMetric(stream, "pulsar_compaction_latency_count", stats.compactionLatencyBuckets.getCount(), cluster, namespace, topic, splitTopicAndPartitionIndexLabel); + + for (TopicMetricBean topicMetricBean : stats.bucketDelayedIndexStats.values()) { + String[] labelsAndValues = topicMetricBean.labelsAndValues; + writeTopicMetric(stream, topicMetricBean.name, topicMetricBean.value, cluster, namespace, + topic, splitTopicAndPartitionIndexLabel, labelsAndValues); + } } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/MockBucketSnapshotStorage.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/MockBucketSnapshotStorage.java index cf7310c7067c3..9e924bdeda341 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/MockBucketSnapshotStorage.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/MockBucketSnapshotStorage.java @@ -174,7 +174,7 @@ public CompletableFuture getBucketSnapshotLength(long bucketId) { long length = 0; List bufList = this.bucketSnapshots.get(bucketId); for (ByteBuf byteBuf : bufList) { - length += byteBuf.array().length; + length += byteBuf.readableBytes(); } return length; }, executorService); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java index 717bada7705dc..95234d688f6a0 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/delayed/bucket/BucketDelayedDeliveryTrackerTest.java @@ -46,7 +46,6 @@ import org.apache.bookkeeper.mledger.ManagedCursor; import org.apache.bookkeeper.mledger.impl.PositionImpl; import org.apache.pulsar.broker.delayed.AbstractDeliveryTrackerTest; -import org.apache.pulsar.broker.delayed.DelayedDeliveryTracker; import org.apache.pulsar.broker.delayed.MockBucketSnapshotStorage; import org.apache.pulsar.broker.delayed.MockManagedCursor; import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers; @@ -158,7 +157,7 @@ public Object[][] provider(Method method) throws Exception { } @Test(dataProvider = "delayedTracker") - public void testContainsMessage(DelayedDeliveryTracker tracker) { + public void testContainsMessage(BucketDelayedDeliveryTracker tracker) { tracker.addMessage(1, 1, 10); tracker.addMessage(2, 2, 20); @@ -191,6 +190,12 @@ public void testRecoverSnapshot(BucketDelayedDeliveryTracker tracker) { clockTime.set(1 * 10); + Awaitility.await().untilAsserted(() -> { + Assert.assertTrue( + tracker.getImmutableBuckets().asMapOfRanges().values().stream().noneMatch(x -> x.merging || + !x.getSnapshotCreateFuture().get().isDone())); + }); + assertTrue(tracker.hasMessageAvailable()); Set scheduledMessages = tracker.getScheduledMessages(100); @@ -202,16 +207,16 @@ public void testRecoverSnapshot(BucketDelayedDeliveryTracker tracker) { clockTime.set(30 * 10); - tracker = new BucketDelayedDeliveryTracker(dispatcher, timer, 1000, clock, - true, bucketSnapshotStorage, 5, TimeUnit.MILLISECONDS.toMillis(10), -1,50); + BucketDelayedDeliveryTracker tracker2 = new BucketDelayedDeliveryTracker(dispatcher, timer, 1000, clock, + true, bucketSnapshotStorage, 5, TimeUnit.MILLISECONDS.toMillis(10), -1, 50); - assertFalse(tracker.containsMessage(101, 101)); - assertEquals(tracker.getNumberOfDelayedMessages(), 70); + assertFalse(tracker2.containsMessage(101, 101)); + assertEquals(tracker2.getNumberOfDelayedMessages(), 70); clockTime.set(100 * 10); - assertTrue(tracker.hasMessageAvailable()); - scheduledMessages = tracker.getScheduledMessages(70); + assertTrue(tracker2.hasMessageAvailable()); + scheduledMessages = tracker2.getScheduledMessages(70); assertEquals(scheduledMessages.size(), 70); @@ -221,7 +226,7 @@ public void testRecoverSnapshot(BucketDelayedDeliveryTracker tracker) { i++; } - tracker.close(); + tracker2.close(); } @Test diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BucketDelayedDeliveryTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BucketDelayedDeliveryTest.java index 09b7cbbf1b99d..5480a2e7a704a 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BucketDelayedDeliveryTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/BucketDelayedDeliveryTest.java @@ -19,18 +19,26 @@ package org.apache.pulsar.broker.service.persistent; import static org.apache.bookkeeper.mledger.impl.ManagedCursorImpl.CURSOR_INTERNAL_PROPERTY_PREFIX; +import static org.testng.Assert.assertEquals; import static org.testng.Assert.assertTrue; +import com.google.common.collect.Multimap; +import java.io.ByteArrayOutputStream; +import java.nio.charset.StandardCharsets; import java.util.List; import java.util.Map; +import java.util.Optional; import java.util.concurrent.TimeUnit; import lombok.Cleanup; import org.apache.bookkeeper.client.BKException; import org.apache.bookkeeper.client.BookKeeper; import org.apache.bookkeeper.client.LedgerHandle; import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl; +import org.apache.commons.lang3.mutable.MutableInt; import org.apache.pulsar.broker.BrokerTestUtil; import org.apache.pulsar.broker.delayed.BucketDelayedDeliveryTrackerFactory; import org.apache.pulsar.broker.service.Dispatcher; +import org.apache.pulsar.broker.stats.PrometheusMetricsTest; +import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsGenerator; import org.apache.pulsar.client.api.Consumer; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.Schema; @@ -162,4 +170,139 @@ public void testUnsubscribe() throws Exception { } } } + + + @Test + public void testBucketDelayedIndexMetrics() throws Exception { + cleanup(); + setup(); + + String topic = BrokerTestUtil.newUniqueName("persistent://public/default/testBucketDelayedIndexMetrics"); + + @Cleanup + Consumer consumer = pulsarClient.newConsumer(Schema.STRING) + .topic(topic) + .subscriptionName("test_sub") + .subscriptionType(SubscriptionType.Shared) + .subscribe(); + + @Cleanup + Consumer consumer2 = pulsarClient.newConsumer(Schema.STRING) + .topic(topic) + .subscriptionName("test_sub2") + .subscriptionType(SubscriptionType.Shared) + .subscribe(); + + @Cleanup + Producer producer = pulsarClient.newProducer(Schema.STRING) + .topic(topic) + .create(); + + final int N = 101; + + for (int i = 0; i < N; i++) { + producer.newMessage() + .value("msg-" + i) + .deliverAfter(3600 + i, TimeUnit.SECONDS) + .sendAsync(); + } + producer.flush(); + + Thread.sleep(2000); + + ByteArrayOutputStream output = new ByteArrayOutputStream(); + PrometheusMetricsGenerator.generate(pulsar, true, true, true, output); + String metricsStr = output.toString(StandardCharsets.UTF_8); + Multimap metricsMap = PrometheusMetricsTest.parseMetrics(metricsStr); + + List bucketsMetrics = + metricsMap.get("pulsar_delayed_message_index_bucket_total").stream() + .filter(metric -> metric.tags.get("topic").equals(topic)).toList(); + MutableInt bucketsSum = new MutableInt(); + bucketsMetrics.stream().filter(metric -> metric.tags.containsKey("subscription")).forEach(metric -> { + assertEquals(3, metric.value); + bucketsSum.add(metric.value); + }); + assertEquals(6, bucketsSum.intValue()); + Optional bucketsTopicMetric = + bucketsMetrics.stream().filter(metric -> !metric.tags.containsKey("subscription")).findFirst(); + assertTrue(bucketsTopicMetric.isPresent()); + assertEquals(bucketsSum.intValue(), bucketsTopicMetric.get().value); + + List loadedIndexMetrics = + metricsMap.get("pulsar_delayed_message_index_loaded").stream() + .filter(metric -> metric.tags.get("topic").equals(topic)).toList(); + MutableInt loadedIndexSum = new MutableInt(); + long count = loadedIndexMetrics.stream().filter(metric -> metric.tags.containsKey("subscription")).peek(metric -> { + assertTrue(metric.value > 0 && metric.value <= N); + loadedIndexSum.add(metric.value); + }).count(); + assertEquals(2, count); + Optional loadedIndexTopicMetrics = + bucketsMetrics.stream().filter(metric -> !metric.tags.containsKey("subscription")).findFirst(); + assertTrue(loadedIndexTopicMetrics.isPresent()); + assertEquals(loadedIndexSum.intValue(), loadedIndexTopicMetrics.get().value); + + List snapshotSizeBytesMetrics = + metricsMap.get("pulsar_delayed_message_index_bucket_snapshot_size_bytes").stream() + .filter(metric -> metric.tags.get("topic").equals(topic)).toList(); + MutableInt snapshotSizeBytesSum = new MutableInt(); + count = snapshotSizeBytesMetrics.stream().filter(metric -> metric.tags.containsKey("subscription")) + .peek(metric -> { + assertTrue(metric.value > 0); + snapshotSizeBytesSum.add(metric.value); + }).count(); + assertEquals(2, count); + Optional snapshotSizeBytesTopicMetrics = + snapshotSizeBytesMetrics.stream().filter(metric -> !metric.tags.containsKey("subscription")).findFirst(); + assertTrue(snapshotSizeBytesTopicMetrics.isPresent()); + assertEquals(snapshotSizeBytesSum.intValue(), snapshotSizeBytesTopicMetrics.get().value); + + List opCountMetrics = + metricsMap.get("pulsar_delayed_message_index_bucket_op_count").stream() + .filter(metric -> metric.tags.get("topic").equals(topic)).toList(); + MutableInt opCountMetricsSum = new MutableInt(); + count = opCountMetrics.stream() + .filter(metric -> metric.tags.get("state").equals("succeed") && metric.tags.get("type").equals("create") + && metric.tags.containsKey("subscription")) + .peek(metric -> { + assertTrue(metric.value >= 2); + opCountMetricsSum.add(metric.value); + }).count(); + assertEquals(2, count); + Optional opCountTopicMetrics = + opCountMetrics.stream() + .filter(metric -> metric.tags.get("state").equals("succeed") && metric.tags.get("type") + .equals("create") && !metric.tags.containsKey("subscription")).findFirst(); + assertTrue(opCountTopicMetrics.isPresent()); + assertEquals(opCountMetricsSum.intValue(), opCountTopicMetrics.get().value); + + List opLatencyMetrics = + metricsMap.get("pulsar_delayed_message_index_bucket_op_latency_ms").stream() + .filter(metric -> metric.tags.get("topic").equals(topic)).toList(); + MutableInt opLatencyMetricsSum = new MutableInt(); + count = opLatencyMetrics.stream() + .filter(metric -> metric.tags.get("type").equals("create") + && metric.tags.containsKey("subscription")) + .peek(metric -> { + assertTrue(metric.tags.containsKey("quantile")); + opLatencyMetricsSum.add(metric.value); + }).count(); + assertTrue(count >= 2); + Optional opLatencyTopicMetrics = + opCountMetrics.stream() + .filter(metric -> metric.tags.get("type").equals("create") + && !metric.tags.containsKey("subscription")).findFirst(); + assertTrue(opLatencyTopicMetrics.isPresent()); + assertEquals(opLatencyMetricsSum.intValue(), opLatencyTopicMetrics.get().value); + + ByteArrayOutputStream namespaceOutput = new ByteArrayOutputStream(); + PrometheusMetricsGenerator.generate(pulsar, false, true, true, namespaceOutput); + Multimap namespaceMetricsMap = PrometheusMetricsTest.parseMetrics(namespaceOutput.toString(StandardCharsets.UTF_8)); + + Optional namespaceMetric = + namespaceMetricsMap.get("pulsar_delayed_message_index_bucket_total").stream().findFirst(); + assertTrue(namespaceMetric.isPresent()); + assertEquals(6, namespaceMetric.get().value); + } } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java index 9c7e24ba02118..25fa666523f3c 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/SubscriptionStatsImpl.java @@ -134,6 +134,8 @@ public class SubscriptionStatsImpl implements SubscriptionStats { /** The size of InMemoryDelayedDeliveryTracer memory usage. */ public long delayedMessageIndexSizeInBytes; + public Map bucketDelayedIndexStats; + /** SubscriptionProperties (key/value strings) associated with this subscribe. */ public Map subscriptionProperties; @@ -149,6 +151,7 @@ public SubscriptionStatsImpl() { this.consumers = new ArrayList<>(); this.consumersAfterMarkDeletePosition = new LinkedHashMap<>(); this.subscriptionProperties = new HashMap<>(); + this.bucketDelayedIndexStats = new HashMap<>(); } public void reset() { @@ -175,6 +178,7 @@ public void reset() { filterAcceptedMsgCount = 0; filterRejectedMsgCount = 0; filterRescheduledMsgCount = 0; + bucketDelayedIndexStats.clear(); } // if the stats are added for the 1st time, we will need to make a copy of these stats and add it to the current @@ -215,6 +219,14 @@ public SubscriptionStatsImpl add(SubscriptionStatsImpl stats) { this.filterAcceptedMsgCount += stats.filterAcceptedMsgCount; this.filterRejectedMsgCount += stats.filterRejectedMsgCount; this.filterRescheduledMsgCount += stats.filterRescheduledMsgCount; + stats.bucketDelayedIndexStats.forEach((k, v) -> { + TopicMetricBean topicMetricBean = + this.bucketDelayedIndexStats.computeIfAbsent(k, __ -> new TopicMetricBean()); + topicMetricBean.name = v.name; + topicMetricBean.labelsAndValues = v.labelsAndValues; + topicMetricBean.value += v.value; + }); + return this; } } diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/TopicMetricBean.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/TopicMetricBean.java new file mode 100644 index 0000000000000..e01a9d7aa71f3 --- /dev/null +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/TopicMetricBean.java @@ -0,0 +1,30 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.pulsar.common.policies.data.stats; + +import lombok.AllArgsConstructor; +import lombok.NoArgsConstructor; + +@NoArgsConstructor +@AllArgsConstructor +public class TopicMetricBean { + public String name; + public double value; + public String[] labelsAndValues; +} diff --git a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/TopicStatsImpl.java b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/TopicStatsImpl.java index 12d30124f7dcd..c9c4739b904f6 100644 --- a/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/TopicStatsImpl.java +++ b/pulsar-common/src/main/java/org/apache/pulsar/common/policies/data/stats/TopicStatsImpl.java @@ -139,6 +139,9 @@ public class TopicStatsImpl implements TopicStats { /** The size of InMemoryDelayedDeliveryTracer memory usage. */ public long delayedMessageIndexSizeInBytes; + /** Map of bucket delayed index statistics. */ + public Map bucketDelayedIndexStats; + /** The compaction stats. */ public CompactionStatsImpl compaction; @@ -182,6 +185,7 @@ public TopicStatsImpl() { this.subscriptions = new HashMap<>(); this.replication = new TreeMap<>(); this.compaction = new CompactionStatsImpl(); + this.bucketDelayedIndexStats = new HashMap<>(); } public void reset() { @@ -214,6 +218,7 @@ public void reset() { this.delayedMessageIndexSizeInBytes = 0; this.compaction.reset(); this.ownerBroker = null; + this.bucketDelayedIndexStats.clear(); } // if the stats are added for the 1st time, we will need to make a copy of these stats and add it to the current @@ -244,6 +249,14 @@ public TopicStatsImpl add(TopicStats ts) { this.abortedTxnCount = stats.abortedTxnCount; this.committedTxnCount = stats.committedTxnCount; + stats.bucketDelayedIndexStats.forEach((k, v) -> { + TopicMetricBean topicMetricBean = + this.bucketDelayedIndexStats.computeIfAbsent(k, __ -> new TopicMetricBean()); + topicMetricBean.name = v.name; + topicMetricBean.labelsAndValues = v.labelsAndValues; + topicMetricBean.value += v.value; + }); + for (int index = 0; index < stats.getPublishers().size(); index++) { PublisherStats s = stats.getPublishers().get(index); if (s.isSupportsPartialProducer() && s.getProducerName() != null) {