add non-persistent topic metrics (apache#28)

Signed-off-by: druidliu <[email protected]>
dragonls · Aug 26, 2022 · 1e502e7 · 1e502e7
1 parent 9419c09
commit 1e502e7
Show file tree

Hide file tree

Showing 4 changed files with 51 additions and 30 deletions.
diff --git a/.../src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedSubscriptionStats.java b/.../src/main/java/org/apache/pulsar/broker/stats/prometheus/AggregatedSubscriptionStats.java
@@ -58,5 +58,7 @@ public class AggregatedSubscriptionStats {
 
     long totalMsgExpired;
 
+    double msgDropRate;
+
     public Map<Consumer, AggregatedConsumerStats> consumerStat = new HashMap<>();
 }
diff --git a/...ker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java b/...ker/src/main/java/org/apache/pulsar/broker/stats/prometheus/NamespaceStatsAggregator.java
@@ -109,6 +109,36 @@ private static Optional<CompactorMXBean> getCompactorMXBean(PulsarService pulsar
         return Optional.ofNullable(compactor).map(c -> c.getStats());
     }
 
+    private static void aggregateTopicStats(TopicStats stats, SubscriptionStats subscriptionStats,
+                                            AggregatedSubscriptionStats subsStats) {
+        stats.subscriptionsCount++;
+        stats.msgBacklog += subscriptionStats.msgBacklog;
+        subsStats.msgBacklog = subscriptionStats.msgBacklog;
+        subsStats.msgDelayed = subscriptionStats.msgDelayed;
+        subsStats.msgRateExpired = subscriptionStats.msgRateExpired;
+        subsStats.totalMsgExpired = subscriptionStats.totalMsgExpired;
+        subsStats.msgBacklogNoDelayed = subsStats.msgBacklog - subsStats.msgDelayed;
+        subsStats.lastExpireTimestamp = subscriptionStats.lastExpireTimestamp;
+        subsStats.lastAckedTimestamp = subscriptionStats.lastAckedTimestamp;
+        subsStats.lastConsumedFlowTimestamp = subscriptionStats.lastConsumedFlowTimestamp;
+        subsStats.lastConsumedTimestamp = subscriptionStats.lastConsumedTimestamp;
+        subsStats.lastMarkDeleteAdvancedTimestamp = subscriptionStats.lastMarkDeleteAdvancedTimestamp;
+        subscriptionStats.consumers.forEach(cStats -> {
+            stats.consumersCount++;
+            subsStats.unackedMessages += cStats.unackedMessages;
+            subsStats.msgRateRedeliver += cStats.msgRateRedeliver;
+            subsStats.msgRateOut += cStats.msgRateOut;
+            subsStats.msgThroughputOut += cStats.msgThroughputOut;
+            subsStats.bytesOutCounter += cStats.bytesOutCounter;
+            subsStats.msgOutCounter += cStats.msgOutCounter;
+            if (!subsStats.blockedSubscriptionOnUnackedMsgs && cStats.blockedConsumerOnUnackedMsgs) {
+                subsStats.blockedSubscriptionOnUnackedMsgs = true;
+            }
+        });
+        stats.rateOut += subsStats.msgRateOut;
+        stats.throughputOut += subsStats.msgThroughputOut;
+    }
+
     private static void getTopicStats(Topic topic, TopicStats stats, boolean includeConsumerMetrics,
             boolean includeProducerMetrics, boolean getPreciseBacklog, boolean subscriptionBacklogSize,
                                       Optional<CompactorMXBean> compactorMXBean) {
@@ -175,37 +205,21 @@ private static void getTopicStats(Topic topic, TopicStats stats, boolean include
             }
         });
 
-        tStatus.subscriptions.forEach((subName, subscriptionStats) -> {
-            stats.subscriptionsCount++;
-            stats.msgBacklog += subscriptionStats.msgBacklog;
-
-            AggregatedSubscriptionStats subsStats = stats.subscriptionStats
-                    .computeIfAbsent(subName, k -> new AggregatedSubscriptionStats());
-            subsStats.msgBacklog = subscriptionStats.msgBacklog;
-            subsStats.msgDelayed = subscriptionStats.msgDelayed;
-            subsStats.msgRateExpired = subscriptionStats.msgRateExpired;
-            subsStats.totalMsgExpired = subscriptionStats.totalMsgExpired;
-            subsStats.msgBacklogNoDelayed = subsStats.msgBacklog - subsStats.msgDelayed;
-            subsStats.lastExpireTimestamp = subscriptionStats.lastExpireTimestamp;
-            subsStats.lastAckedTimestamp = subscriptionStats.lastAckedTimestamp;
-            subsStats.lastConsumedFlowTimestamp = subscriptionStats.lastConsumedFlowTimestamp;
-            subsStats.lastConsumedTimestamp = subscriptionStats.lastConsumedTimestamp;
-            subsStats.lastMarkDeleteAdvancedTimestamp = subscriptionStats.lastMarkDeleteAdvancedTimestamp;
-            subscriptionStats.consumers.forEach(cStats -> {
-                stats.consumersCount++;
-                subsStats.unackedMessages += cStats.unackedMessages;
-                subsStats.msgRateRedeliver += cStats.msgRateRedeliver;
-                subsStats.msgRateOut += cStats.msgRateOut;
-                subsStats.msgThroughputOut += cStats.msgThroughputOut;
-                subsStats.bytesOutCounter += cStats.bytesOutCounter;
-                subsStats.msgOutCounter += cStats.msgOutCounter;
-                if (!subsStats.blockedSubscriptionOnUnackedMsgs && cStats.blockedConsumerOnUnackedMsgs) {
-                    subsStats.blockedSubscriptionOnUnackedMsgs = true;
-                }
+        if (topic instanceof PersistentTopic) {
+            tStatus.subscriptions.forEach((subName, subscriptionStats) -> {
+                AggregatedSubscriptionStats subsStats = stats.subscriptionStats
+                        .computeIfAbsent(subName, k -> new AggregatedSubscriptionStats());
+                aggregateTopicStats(stats, subscriptionStats, subsStats);
             });
-            stats.rateOut += subsStats.msgRateOut;
-            stats.throughputOut += subsStats.msgThroughputOut;
-        });
+        } else {
+            ((NonPersistentTopicStats) tStatus).getSubscriptions()
+                    .forEach((subName, nonPersistentSubscriptionStats) -> {
+                        AggregatedSubscriptionStats subsStats = stats.subscriptionStats
+                                .computeIfAbsent(subName, k -> new AggregatedSubscriptionStats());
+                        aggregateTopicStats(stats, nonPersistentSubscriptionStats, subsStats);
+                        subsStats.msgDropRate += nonPersistentSubscriptionStats.getMsgDropRate();
+                    });
+        }
 
         // Consumer stats can be a lot if a subscription has many consumers
         if (includeConsumerMetrics) {

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
@@ -267,6 +267,8 @@ static void printTopicStats(SimpleTextOutputStream stream, String cluster, Strin
                     subsStats.msgRateExpired, splitTopicAndPartitionIndexLabel);
             metric(stream, cluster, namespace, topic, n, "pulsar_subscription_total_msg_expired",
                     subsStats.totalMsgExpired, splitTopicAndPartitionIndexLabel);
+            metric(stream, cluster, namespace, topic, n, "pulsar_subscription_msg_drop_rate",
+                    subsStats.msgDropRate, splitTopicAndPartitionIndexLabel);
             subsStats.consumerStat.forEach((c, consumerStats) -> {
                 metric(stream, cluster, namespace, topic, n, c.consumerName(), c.consumerId(),
                         "pulsar_consumer_msg_rate_redeliver", consumerStats.msgRateRedeliver,

diff --git a/...java/org/apache/pulsar/common/policies/data/stats/NonPersistentSubscriptionStatsImpl.java b/...java/org/apache/pulsar/common/policies/data/stats/NonPersistentSubscriptionStatsImpl.java
@@ -22,6 +22,9 @@
 import java.util.Objects;
 import lombok.Getter;
 import org.apache.pulsar.common.policies.data.NonPersistentSubscriptionStats;
+import lombok.Getter;
+
+import static com.google.common.base.Preconditions.checkNotNull;
 
 /**
  * Statistics for subscription to non-persistent topics.