This is an automated email from the ASF dual-hosted git repository. penghui pushed a commit to branch branch-2.8 in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 629a470587a3ff63e06fd1c295260df1e8a20d0a Author: GuoJiwei <[email protected]> AuthorDate: Wed Aug 4 23:42:54 2021 +0800 Add metrics [AddEntryWithReplicasBytesRate] for namespace (#11472) (cherry picked from commit 92e7825fce9b41b801a7762dea49b592d364b6b3) --- .../bookkeeper/mledger/ManagedLedgerMXBean.java | 5 +++++ .../bookkeeper/mledger/impl/EntryCacheManager.java | 6 +++--- .../mledger/impl/ManagedLedgerMBeanImpl.java | 10 +++++++++- .../mledger/impl/ManagedLedgerMBeanTest.java | 2 ++ .../broker/stats/metrics/ManagedLedgerMetrics.java | 2 ++ .../stats/prometheus/NamespaceStatsAggregator.java | 1 + .../pulsar/broker/stats/prometheus/TopicStats.java | 22 +++++++++++----------- .../broker/stats/TransactionMetricsTest.java | 2 +- site2/docs/reference-metrics.md | 4 +++- 9 files changed, 37 insertions(+), 17 deletions(-) diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerMXBean.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerMXBean.java index b4444c0..6eccd02 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerMXBean.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerMXBean.java @@ -61,6 +61,11 @@ public interface ManagedLedgerMXBean { double getAddEntryBytesRate(); /** + * @return the bytes/s rate of messages added with replicas + */ + double getAddEntryWithReplicasBytesRate(); + + /** * @return the msg/s rate of messages read */ double getReadEntriesRate(); diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheManager.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheManager.java index 81008cc..c87bcb8 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheManager.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/EntryCacheManager.java @@ -46,7 +46,7 @@ public class EntryCacheManager { private final long maxSize; private final long evictionTriggerThreshold; - private final double cacheEvictionWatermak; + private final double cacheEvictionWatermark; private final AtomicLong currentSize = new AtomicLong(0); private final ConcurrentMap<String, EntryCache> caches = Maps.newConcurrentMap(); private final EntryCacheEvictionPolicy evictionPolicy; @@ -64,7 +64,7 @@ public class EntryCacheManager { public EntryCacheManager(ManagedLedgerFactoryImpl factory) { this.maxSize = factory.getConfig().getMaxCacheSize(); this.evictionTriggerThreshold = (long) (maxSize * evictionTriggerThresholdPercent); - this.cacheEvictionWatermak = factory.getConfig().getCacheEvictionWatermark(); + this.cacheEvictionWatermark = factory.getConfig().getCacheEvictionWatermark(); this.evictionPolicy = new EntryCacheDefaultEvictionPolicy(); this.mlFactory = factory; this.mlFactoryMBean = factory.mbean; @@ -109,7 +109,7 @@ public class EntryCacheManager { mlFactory.scheduledExecutor.execute(safeRun(() -> { // Trigger a new cache eviction cycle to bring the used memory below the cacheEvictionWatermark // percentage limit - long sizeToEvict = currentSize - (long) (maxSize * cacheEvictionWatermak); + long sizeToEvict = currentSize - (long) (maxSize * cacheEvictionWatermark); long startTime = System.nanoTime(); log.info("Triggering cache eviction. total size: {} Mb -- Need to discard: {} Mb", currentSize / MB, sizeToEvict / MB); diff --git a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerMBeanImpl.java b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerMBeanImpl.java index 9e22b51..01a88c1 100644 --- a/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerMBeanImpl.java +++ b/managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerMBeanImpl.java @@ -35,6 +35,7 @@ public class ManagedLedgerMBeanImpl implements ManagedLedgerMXBean { private final ManagedLedgerImpl managedLedger; private final Rate addEntryOps = new Rate(); + private final Rate addEntryWithReplicasOps = new Rate(); private final Rate addEntryOpsFailed = new Rate(); private final Rate readEntriesOps = new Rate(); private final Rate readEntriesOpsFailed = new Rate(); @@ -49,7 +50,7 @@ public class ManagedLedgerMBeanImpl implements ManagedLedgerMXBean { private final LongAdder cursorLedgerCreateOp = new LongAdder(); private final LongAdder cursorLedgerDeleteOp = new LongAdder(); - // addEntryLatencyStatsUsec measure total latency including time entry spent while waiting in queue + // addEntryLatencyStatsUsec measure total latency including time entry spent while waiting in queue private final StatsBuckets addEntryLatencyStatsUsec = new StatsBuckets(ENTRY_LATENCY_BUCKETS_USEC); // ledgerAddEntryLatencyStatsUsec measure latency to persist entry into ledger private final StatsBuckets ledgerAddEntryLatencyStatsUsec = new StatsBuckets(ENTRY_LATENCY_BUCKETS_USEC); @@ -63,6 +64,7 @@ public class ManagedLedgerMBeanImpl implements ManagedLedgerMXBean { public void refreshStats(long period, TimeUnit unit) { double seconds = unit.toMillis(period) / 1000.0; addEntryOps.calculateRate(seconds); + addEntryWithReplicasOps.calculateRate(seconds); addEntryOpsFailed.calculateRate(seconds); readEntriesOps.calculateRate(seconds); readEntriesOpsFailed.calculateRate(seconds); @@ -77,6 +79,7 @@ public class ManagedLedgerMBeanImpl implements ManagedLedgerMXBean { public void addAddEntrySample(long size) { addEntryOps.recordEvent(size); entryStats.addValue(size); + addEntryWithReplicasOps.recordEvent(size * managedLedger.getConfig().getWriteQuorumSize()); } public void addMarkDeleteOp() { @@ -187,6 +190,11 @@ public class ManagedLedgerMBeanImpl implements ManagedLedgerMXBean { } @Override + public double getAddEntryWithReplicasBytesRate() { + return addEntryWithReplicasOps.getValueRate(); + } + + @Override public double getReadEntriesRate() { return readEntriesOps.getRate(); } diff --git a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerMBeanTest.java b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerMBeanTest.java index 66accf3..89c69a0 100644 --- a/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerMBeanTest.java +++ b/managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerMBeanTest.java @@ -74,6 +74,7 @@ public class ManagedLedgerMBeanTest extends MockedBookKeeperTestCase { }).get(); assertEquals(mbean.getAddEntryBytesRate(), 0.0); + assertEquals(mbean.getAddEntryWithReplicasBytesRate(), 0.0); assertEquals(mbean.getAddEntryMessagesRate(), 0.0); assertEquals(mbean.getAddEntrySucceed(), 0); assertEquals(mbean.getAddEntryErrors(), 0); @@ -101,6 +102,7 @@ public class ManagedLedgerMBeanTest extends MockedBookKeeperTestCase { }).get(); assertEquals(mbean.getAddEntryBytesRate(), 800.0); + assertEquals(mbean.getAddEntryWithReplicasBytesRate(), 1600.0); assertEquals(mbean.getAddEntryMessagesRate(), 2.0); assertEquals(mbean.getAddEntrySucceed(), 2); assertEquals(mbean.getAddEntryErrors(), 0); diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/metrics/ManagedLedgerMetrics.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/metrics/ManagedLedgerMetrics.java index 8e15aff..889eb8e 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/metrics/ManagedLedgerMetrics.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/metrics/ManagedLedgerMetrics.java @@ -85,6 +85,8 @@ public class ManagedLedgerMetrics extends AbstractMetrics { populateAggregationMapWithSum(tempAggregatedMetricsMap, "brk_ml_AddEntryBytesRate", lStats.getAddEntryBytesRate()); + populateAggregationMapWithSum(tempAggregatedMetricsMap, "brk_ml_AddEntryWithReplicasBytesRate", + lStats.getAddEntryWithReplicasBytesRate()); populateAggregationMapWithSum(tempAggregatedMetricsMap, "brk_ml_AddEntryErrors", (double) lStats.getAddEntryErrors()); 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 43119d7..c08641d 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 @@ -231,6 +231,7 @@ public class NamespaceStatsAggregator { metric(stream, cluster, "pulsar_throughput_in", 0); metric(stream, cluster, "pulsar_throughput_out", 0); metric(stream, cluster, "pulsar_storage_size", 0); + metric(stream, cluster, "pulsar_storage_logical_size", 0); metric(stream, cluster, "pulsar_storage_write_rate", 0); metric(stream, cluster, "pulsar_storage_read_rate", 0); metric(stream, cluster, "pulsar_msg_backlog", 0); 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 1ac514a..9474ecb 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 @@ -119,27 +119,27 @@ class TopicStats { metric(stream, cluster, namespace, topic, "pulsar_storage_write_latency_sum", stats.managedLedgerStats.storageWriteLatencyBuckets.getSum()); - long[] ledgerWritelatencyBuckets = stats.managedLedgerStats.storageLedgerWriteLatencyBuckets.getBuckets(); + long[] ledgerWriteLatencyBuckets = stats.managedLedgerStats.storageLedgerWriteLatencyBuckets.getBuckets(); metric(stream, cluster, namespace, topic, "pulsar_storage_ledger_write_latency_le_0_5", - ledgerWritelatencyBuckets[0]); + ledgerWriteLatencyBuckets[0]); metric(stream, cluster, namespace, topic, "pulsar_storage_ledger_write_latency_le_1", - ledgerWritelatencyBuckets[1]); + ledgerWriteLatencyBuckets[1]); metric(stream, cluster, namespace, topic, "pulsar_storage_ledger_write_latency_le_5", - ledgerWritelatencyBuckets[2]); + ledgerWriteLatencyBuckets[2]); metric(stream, cluster, namespace, topic, "pulsar_storage_ledger_write_latency_le_10", - ledgerWritelatencyBuckets[3]); + ledgerWriteLatencyBuckets[3]); metric(stream, cluster, namespace, topic, "pulsar_storage_ledger_write_latency_le_20", - ledgerWritelatencyBuckets[4]); + ledgerWriteLatencyBuckets[4]); metric(stream, cluster, namespace, topic, "pulsar_storage_ledger_write_latency_le_50", - ledgerWritelatencyBuckets[5]); + ledgerWriteLatencyBuckets[5]); metric(stream, cluster, namespace, topic, "pulsar_storage_ledger_write_latency_le_100", - ledgerWritelatencyBuckets[6]); + ledgerWriteLatencyBuckets[6]); metric(stream, cluster, namespace, topic, "pulsar_storage_ledger_write_latency_le_200", - ledgerWritelatencyBuckets[7]); + ledgerWriteLatencyBuckets[7]); metric(stream, cluster, namespace, topic, "pulsar_storage_ledger_write_latency_le_1000", - ledgerWritelatencyBuckets[8]); + ledgerWriteLatencyBuckets[8]); metric(stream, cluster, namespace, topic, "pulsar_storage_ledger_write_latency_overflow", - ledgerWritelatencyBuckets[9]); + ledgerWriteLatencyBuckets[9]); metric(stream, cluster, namespace, topic, "pulsar_storage_ledger_write_latency_count", stats.managedLedgerStats.storageLedgerWriteLatencyBuckets.getCount()); metric(stream, cluster, namespace, topic, "pulsar_storage_ledger_write_latency_sum", diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/TransactionMetricsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/TransactionMetricsTest.java index f0266de..b97f223 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/TransactionMetricsTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/TransactionMetricsTest.java @@ -250,7 +250,7 @@ public class TransactionMetricsTest extends BrokerTestBase { metric = metrics.get("pulsar_storage_size"); assertEquals(metric.size(), 3); metric = metrics.get("pulsar_storage_logical_size"); - assertEquals(metric.size(), 2); + assertEquals(metric.size(), 3); metric = metrics.get("pulsar_storage_backlog_size"); assertEquals(metric.size(), 2); } diff --git a/site2/docs/reference-metrics.md b/site2/docs/reference-metrics.md index c89ca5c..aa69b79 100644 --- a/site2/docs/reference-metrics.md +++ b/site2/docs/reference-metrics.md @@ -144,7 +144,7 @@ All the namespace metrics are labelled with the following labels: | pulsar_throughput_in | Gauge | The total throughput of the namespace coming into this broker (bytes/second). | | pulsar_throughput_out | Gauge | The total throughput of the namespace going out from this broker (bytes/second). | | pulsar_storage_size | Gauge | The total storage size of the topics in this namespace owned by this broker (bytes). | -| pulsar_storage_logical_size | Gauge | The storage size (without accounting for replicas) of the topics in this namespace owned by this broker (bytes). | +| pulsar_storage_logical_size | Gauge | The storage size of topics in the namespace owned by the broker without replicas (in bytes). | | pulsar_storage_backlog_size | Gauge | The total backlog size of the topics of this namespace owned by this broker (messages). | | pulsar_storage_offloaded_size | Gauge | The total amount of the data in this namespace offloaded to the tiered storage (bytes). | | pulsar_storage_write_rate | Gauge | The total message batches (entries) written to the storage for this namespace (message batches / second). | @@ -191,6 +191,7 @@ All the topic metrics are labelled with the following labels: | pulsar_throughput_in | Gauge | The total throughput of the topic coming into this broker (bytes/second). | | pulsar_throughput_out | Gauge | The total throughput of the topic going out from this broker (bytes/second). | | pulsar_storage_size | Gauge | The total storage size of the topics in this topic owned by this broker (bytes). | +| pulsar_storage_logical_size | Gauge | The storage size of topics in the namespace owned by the broker without replicas (in bytes). | | pulsar_storage_backlog_size | Gauge | The total backlog size of the topics of this topic owned by this broker (messages). | | pulsar_storage_offloaded_size | Gauge | The total amount of the data in this topic offloaded to the tiered storage (bytes). | | pulsar_storage_backlog_quota_limit | Gauge | The total amount of the data in this topic that limit the backlog quota (bytes). | @@ -248,6 +249,7 @@ All the managedLedger metrics are labelled with the following labels: | Name | Type | Description | | --- | --- | --- | | pulsar_ml_AddEntryBytesRate | Gauge | The bytes/s rate of messages added | +| pulsar_ml_AddEntryWithReplicasBytesRate | Gauge | The bytes/s rate of messages added with replicas | | pulsar_ml_AddEntryErrors | Gauge | The number of addEntry requests that failed | | pulsar_ml_AddEntryLatencyBuckets | Histogram | The add entry latency of a ledger with a given quantile (threshold).<br> Available quantile: <br><ul><li> quantile="0.0_0.5" is AddEntryLatency between (0.0ms, 0.5ms]</li> <li>quantile="0.5_1.0" is AddEntryLatency between (0.5ms, 1.0ms]</li><li>quantile="1.0_5.0" is AddEntryLatency between (1ms, 5ms]</li><li>quantile="5.0_10.0" is AddEntryLatency between (5ms, 10ms]</li><li>quantile="10.0_20.0" is AddEntryLatency between (10ms, 20ms]</li>< [...] | pulsar_ml_AddEntryLatencyBuckets_OVERFLOW | Gauge | The add entry latency > 1s |
