[feat][broker] PIP-264: Add topic messaging metrics

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerMXBean.java

@@ -85,6 +85,11 @@ public interface ManagedLedgerMXBean {
      */
     long getAddEntrySucceed();
 
+    /**
+     * @return the total number of addEntry requests that succeeded
+     */
+    long getAddEntrySucceedTotal();
+
     /**
      * @return the number of addEntry requests that failed
      */
@@ -100,6 +105,11 @@ public interface ManagedLedgerMXBean {
      */
     long getReadEntriesSucceeded();
 
+    /**
+     * @return the total number of readEntries requests that succeeded
+     */
+    long getReadEntriesSucceededTotal();
+
     /**
      * @return the number of readEntries requests that failed
      */

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerMBeanImpl.java

@@ -230,6 +230,11 @@ public long getAddEntrySucceed() {
         return addEntryOps.getCount();
     }
 
+    @Override
+    public long getAddEntrySucceedTotal() {
+        return addEntryOps.getTotalCount();
+    }
+
     @Override
     public long getAddEntryErrors() {
         return addEntryOpsFailed.getCount();
@@ -240,6 +245,11 @@ public long getReadEntriesSucceeded() {
         return readEntriesOps.getCount();
     }
 
+    @Override
+    public long getReadEntriesSucceededTotal() {
+        return readEntriesOps.getTotalCount();
+    }
+
     @Override
     public long getReadEntriesErrors() {
         return readEntriesOpsFailed.getCount();

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerMBeanTest.java

@@ -77,10 +77,12 @@ public void simple() throws Exception {
         assertEquals(mbean.getAddEntryWithReplicasBytesRate(), 0.0);
         assertEquals(mbean.getAddEntryMessagesRate(), 0.0);
         assertEquals(mbean.getAddEntrySucceed(), 0);
+        assertEquals(mbean.getAddEntrySucceedTotal(), 0);
         assertEquals(mbean.getAddEntryErrors(), 0);
         assertEquals(mbean.getReadEntriesBytesRate(), 0.0);
         assertEquals(mbean.getReadEntriesRate(), 0.0);
         assertEquals(mbean.getReadEntriesSucceeded(), 0);
+        assertEquals(mbean.getReadEntriesSucceededTotal(), 0);
         assertEquals(mbean.getReadEntriesErrors(), 0);
         assertEquals(mbean.getMarkDeleteRate(), 0.0);
 
@@ -105,10 +107,12 @@ public void simple() throws Exception {
         assertEquals(mbean.getAddEntryWithReplicasBytesRate(), 1600.0);
         assertEquals(mbean.getAddEntryMessagesRate(), 2.0);
         assertEquals(mbean.getAddEntrySucceed(), 2);
+        assertEquals(mbean.getAddEntrySucceedTotal(), 2);
         assertEquals(mbean.getAddEntryErrors(), 0);
         assertEquals(mbean.getReadEntriesBytesRate(), 0.0);
         assertEquals(mbean.getReadEntriesRate(), 0.0);
         assertEquals(mbean.getReadEntriesSucceeded(), 0);
+        assertEquals(mbean.getReadEntriesSucceededTotal(), 0);
         assertEquals(mbean.getReadEntriesErrors(), 0);
         assertTrue(mbean.getMarkDeleteRate() > 0.0);
 
@@ -134,10 +138,14 @@ public void simple() throws Exception {
         assertEquals(mbean.getReadEntriesBytesRate(), 600.0);
         assertEquals(mbean.getReadEntriesRate(), 1.0);
         assertEquals(mbean.getReadEntriesSucceeded(), 1);
+        assertEquals(mbean.getReadEntriesSucceededTotal(), 1);
         assertEquals(mbean.getReadEntriesErrors(), 0);
         assertEquals(mbean.getNumberOfMessagesInBacklog(), 1);
         assertEquals(mbean.getMarkDeleteRate(), 0.0);
 
+        assertEquals(mbean.getAddEntrySucceed(), 0);
+        assertEquals(mbean.getAddEntrySucceedTotal(), 2);
+
         factory.shutdown();
     }
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java

@@ -109,6 +109,7 @@
 import org.apache.pulsar.broker.service.schema.SchemaRegistryService;
 import org.apache.pulsar.broker.service.schema.SchemaStorageFactory;
 import org.apache.pulsar.broker.stats.MetricsGenerator;
+import org.apache.pulsar.broker.stats.OpenTelemetryTopicStats;
 import org.apache.pulsar.broker.stats.PulsarBrokerOpenTelemetry;
 import org.apache.pulsar.broker.stats.prometheus.PrometheusRawMetricsProvider;
 import org.apache.pulsar.broker.stats.prometheus.PulsarPrometheusMetricsServlet;
@@ -251,6 +252,7 @@ public class PulsarService implements AutoCloseable, ShutdownService {
 
     private MetricsGenerator metricsGenerator;
     private final PulsarBrokerOpenTelemetry openTelemetry;
+    private OpenTelemetryTopicStats openTelemetryTopicStats;
 
     private TransactionMetadataStoreService transactionMetadataStoreService;
     private TransactionBufferProvider transactionBufferProvider;
@@ -622,6 +624,10 @@ public CompletableFuture<Void> closeAsync() {
             brokerClientSharedTimer.stop();
             monotonicSnapshotClock.close();
 
+            if (openTelemetryTopicStats != null) {
+                openTelemetryTopicStats.close();
+            }
+
             asyncCloseFutures.add(EventLoopUtil.shutdownGracefully(ioEventLoopGroup));
 
 
@@ -762,6 +768,8 @@ public void start() throws PulsarServerException {
                         config.getDefaultRetentionTimeInMinutes() * 60));
             }
 
+            openTelemetryTopicStats = new OpenTelemetryTopicStats(this);
+
             localMetadataSynchronizer = StringUtils.isNotBlank(config.getMetadataSyncEventTopic())
                     ? new PulsarMetadataEventSynchronizer(this, config.getMetadataSyncEventTopic())
                     : null;

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java

@@ -133,6 +133,9 @@ public abstract class AbstractTopic implements Topic, TopicPolicyListener<TopicP
     private static final AtomicLongFieldUpdater<AbstractTopic> RATE_LIMITED_UPDATER =
             AtomicLongFieldUpdater.newUpdater(AbstractTopic.class, "publishRateLimitedTimes");
     protected volatile long publishRateLimitedTimes = 0L;
+    private static final AtomicLongFieldUpdater<AbstractTopic> TOTAL_RATE_LIMITED_UPDATER =
+            AtomicLongFieldUpdater.newUpdater(AbstractTopic.class, "totalPublishRateLimitedCounter");
+    protected volatile long totalPublishRateLimitedCounter = 0L;
 
     private static final AtomicIntegerFieldUpdater<AbstractTopic> USER_CREATED_PRODUCER_COUNTER_UPDATER =
             AtomicIntegerFieldUpdater.newUpdater(AbstractTopic.class, "userCreatedProducerCount");
@@ -887,6 +890,7 @@ public void recordAddLatency(long latency, TimeUnit unit) {
 
     @Override
     public long increasePublishLimitedTimes() {
+        TOTAL_RATE_LIMITED_UPDATER.incrementAndGet(this);
         return RATE_LIMITED_UPDATER.incrementAndGet(this);
     }
 
@@ -1174,6 +1178,10 @@ public long getBytesOutCounter() {
                 + sumSubscriptions(AbstractSubscription::getBytesOutCounter);
     }
 
+    public long getTotalPublishRateLimitCounter() {
+        return TOTAL_RATE_LIMITED_UPDATER.get(this);
+    }
+
     private long sumSubscriptions(ToLongFunction<AbstractSubscription> toCounter) {
         return getSubscriptions().values().stream()
                 .map(AbstractSubscription.class::cast)