[monitoring][broker][metadata] add metadata store metrics

pulsar-broker/src/test/java/org/apache/pulsar/broker/stats/PrometheusMetricsTest.java

@@ -65,12 +65,7 @@
 import org.apache.pulsar.broker.service.persistent.PersistentSubscription;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
 import org.apache.pulsar.broker.stats.prometheus.PrometheusMetricsGenerator;
-import org.apache.pulsar.client.api.Consumer;
-import org.apache.pulsar.client.api.MessageRoutingMode;
-import org.apache.pulsar.client.api.Producer;
-import org.apache.pulsar.client.api.PulsarClient;
-import org.apache.pulsar.client.api.PulsarClientException;
-import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.api.*;
 import org.apache.pulsar.compaction.Compactor;
 import org.awaitility.Awaitility;
 import org.mockito.Mockito;
@@ -1522,6 +1517,83 @@ public void testSplitTopicAndPartitionLabel() throws Exception {
         consumer2.close();
     }
 
+
+    @Test
+    public void testMetadataStoreStats() throws Exception {
+        String ns = "prop/ns-abc1";
+        admin.namespaces().createNamespace(ns);
+
+        String topic = "persistent://prop/ns-abc1/metadata-store-" + UUID.randomUUID();
+        String subName = "my-sub1";
+
+        @Cleanup
+        Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+                .topic(topic).create();
+        @Cleanup
+        Consumer<String> consumer = pulsarClient.newConsumer(Schema.STRING)
+                .topic(topic).subscriptionName(subName).subscribe();
+
+        for (int i = 0; i < 100; i++) {
+            producer.newMessage().value(UUID.randomUUID().toString()).send();
+        }
+
+        for (;;) {
+            Message<String> message = consumer.receive(10, TimeUnit.SECONDS);
+            if (message == null) {
+                break;
+            }
+            consumer.acknowledge(message);
+        }
+
+        ByteArrayOutputStream output = new ByteArrayOutputStream();
+        PrometheusMetricsGenerator.generate(pulsar, false, false, false, false, output);
+        String metricsStr = output.toString();
+        Multimap<String, Metric> metricsMap = parseMetrics(metricsStr);
+
+        Collection<Metric> getOpsFailed = metricsMap.get("pulsar_metadata_store_get_ops_failed" + "_total");
+        Collection<Metric> delOpsFailed = metricsMap.get("pulsar_metadata_store_del_ops_failed" + "_total");
+        Collection<Metric> putOpsFailed = metricsMap.get("pulsar_metadata_store_put_ops_failed" + "_total");
+
+        Collection<Metric> getOpsLatency = metricsMap.get("pulsar_metadata_store_get_ops_latency" + "_sum");
+        Collection<Metric> delOpsLatency = metricsMap.get("pulsar_metadata_store_del_ops_latency" + "_sum");
+        Collection<Metric> putOpsLatency = metricsMap.get("pulsar_metadata_store_put_ops_latency" + "_sum");
+
+        Collection<Metric> putBytes = metricsMap.get("pulsar_metadata_store_put_bytes" + "_total");
+
+        Assert.assertTrue(getOpsFailed.size() > 0);
+        Assert.assertTrue(delOpsFailed.size() > 0);
+        Assert.assertTrue(putOpsFailed.size() > 0);
+        Assert.assertTrue(getOpsLatency.size() > 0);
+        Assert.assertTrue(delOpsLatency.size() > 0);
+        Assert.assertTrue(putOpsLatency.size() > 0);
+        Assert.assertTrue(putBytes.size() > 0);
+
+        for (Metric m : getOpsFailed) {
+            Assert.assertEquals(m.tags.get("cluster"), "test");
+        }
+        for (Metric m : delOpsFailed) {
+            Assert.assertEquals(m.tags.get("cluster"), "test");
+        }
+        for (Metric m : putOpsFailed) {
+            Assert.assertEquals(m.tags.get("cluster"), "test");
+        }
+        for (Metric m : getOpsLatency) {
+            Assert.assertEquals(m.tags.get("cluster"), "test");
+            Assert.assertTrue(m.value > 0);
+        }
+        for (Metric m : delOpsLatency) {
+            Assert.assertEquals(m.tags.get("cluster"), "test");
+        }
+        for (Metric m : putOpsLatency) {
+            Assert.assertEquals(m.tags.get("cluster"), "test");
+            Assert.assertTrue(m.value > 0);
+        }
+        for (Metric m : putBytes) {
+            Assert.assertEquals(m.tags.get("cluster"), "test");
+            Assert.assertTrue(m.value > 0);
+        }
+    }
+
     private void compareCompactionStateCount(List<Metric> cm, double count) {
         assertEquals(cm.size(), 1);
         assertEquals(cm.get(0).tags.get("cluster"), "test");

pulsar-metadata/pom.xml

@@ -99,6 +99,11 @@
       <artifactId>caffeine</artifactId>
     </dependency>
 
+    <dependency>
+      <groupId>io.prometheus</groupId>
+      <artifactId>simpleclient</artifactId>
+    </dependency>
+
   </dependencies>
 
   <build>

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/AbstractMetadataStore.java

@@ -58,6 +58,7 @@
 import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
 import org.apache.pulsar.metadata.api.extended.SessionEvent;
 import org.apache.pulsar.metadata.cache.impl.MetadataCacheImpl;
+import org.apache.pulsar.metadata.impl.stats.MetadataStoreStats;
 
 @Slf4j
 public abstract class AbstractMetadataStore implements MetadataStoreExtended, Consumer<Notification> {
@@ -70,6 +71,7 @@ public abstract class AbstractMetadataStore implements MetadataStoreExtended, Co
     private final AsyncLoadingCache<String, List<String>> childrenCache;
     private final AsyncLoadingCache<String, Boolean> existsCache;
     private final CopyOnWriteArrayList<MetadataCacheImpl<?>> metadataCaches = new CopyOnWriteArrayList<>();
+    private final MetadataStoreStats stats;
 
     // We don't strictly need to use 'volatile' here because we don't need the precise consistent semantic. Instead,
     // we want to avoid the overhead of 'volatile'.
@@ -126,6 +128,8 @@ public CompletableFuture<Boolean> asyncReload(String key, Boolean oldValue,
                         }
                     }
                 });
+
+        this.stats = MetadataStoreStats.create();
     }
 
     protected void registerSyncLister(Optional<MetadataEventSynchronizer> synchronizer) {
@@ -235,10 +239,20 @@ public <T> MetadataCache<T> getMetadataCache(MetadataSerde<T> serde) {
 
     @Override
     public CompletableFuture<Optional<GetResult>> get(String path) {
+        long start = System.currentTimeMillis();
         if (!isValidPath(path)) {
-            return FutureUtil.failedFuture(new MetadataStoreException.InvalidPathException(path));
+            stats.recordGetOpsFailed();
+            return FutureUtil
+                    .failedFuture(new MetadataStoreException.InvalidPathException(path));
         }
-        return storeGet(path);
+        return storeGet(path)
+                .whenComplete((v, t) -> {
+                    if (t != null) {
+                        stats.recordGetOpsFailed();
+                    } else {
+                        stats.recordGetOpsLatency(System.currentTimeMillis() - start);
+                    }
+                });
     }
 
     protected abstract CompletableFuture<Optional<GetResult>> storeGet(String path);
@@ -313,17 +327,33 @@ public void accept(Notification n) {
 
     @Override
     public final CompletableFuture<Void> delete(String path, Optional<Long> expectedVersion) {
+        long start = System.currentTimeMillis();
         if (!isValidPath(path)) {
+            stats.recordDelOpsFailed();
             return FutureUtil.failedFuture(new MetadataStoreException.InvalidPathException(path));
         }
         if (getMetadataEventSynchronizer().isPresent()) {
             MetadataEvent event = new MetadataEvent(path, null, new HashSet<>(),
                     expectedVersion.orElse(null), Instant.now().toEpochMilli(),
                     getMetadataEventSynchronizer().get().getClusterName(), NotificationType.Deleted);
             return getMetadataEventSynchronizer().get().notify(event)
-                    .thenCompose(__ -> deleteInternal(path, expectedVersion));
+                    .thenCompose(__ -> deleteInternal(path, expectedVersion))
+                    .whenComplete((v, t) -> {
+                        if (null != t) {
+                            stats.recordDelOpsFailed();
+                        } else {
+                            stats.recordDelOpsLatency(System.currentTimeMillis() - start);
+                        }
+                    });
         } else {
-            return deleteInternal(path, expectedVersion);
+            return deleteInternal(path, expectedVersion)
+                    .whenComplete((v, t) -> {
+                        if (null != t) {
+                            stats.recordDelOpsFailed();
+                        } else {
+                            stats.recordDelOpsLatency(System.currentTimeMillis() - start);
+                        }
+                    });
         }
     }
 
@@ -363,7 +393,9 @@ protected abstract CompletableFuture<Stat> storePut(String path, byte[] data, Op
     @Override
     public final CompletableFuture<Stat> put(String path, byte[] data, Optional<Long> optExpectedVersion,
             EnumSet<CreateOption> options) {
+        long start = System.currentTimeMillis();
         if (!isValidPath(path)) {
+            stats.recordPutOpsFailed();
             return FutureUtil.failedFuture(new MetadataStoreException.InvalidPathException(path));
         }
         HashSet<CreateOption> ops = new HashSet<>(options);
@@ -374,9 +406,25 @@ public final CompletableFuture<Stat> put(String path, byte[] data, Optional<Long
                     Instant.now().toEpochMilli(), getMetadataEventSynchronizer().get().getClusterName(),
                     NotificationType.Modified);
             return getMetadataEventSynchronizer().get().notify(event)
-                    .thenCompose(__ -> putInternal(path, data, optExpectedVersion, options));
+                    .thenCompose(__ -> putInternal(path, data, optExpectedVersion, options))
+                    .whenComplete((v, t) -> {
+                        if (t != null) {
+                            stats.recordPutOpsFailed();
+                        } else {
+                            int len = data == null ? 0 : data.length;
+                            stats.recordPutOpsLatency(System.currentTimeMillis() - start, len);
+                        }
+                    });
         } else {
-            return putInternal(path, data, optExpectedVersion, options);
+            return putInternal(path, data, optExpectedVersion, options)
+                    .whenComplete((v, t) -> {
+                        if (t != null) {
+                            stats.recordPutOpsFailed();
+                        } else {
+                            int len = data == null ? 0 : data.length;
+                            stats.recordPutOpsLatency(System.currentTimeMillis() - start, len);
+                        }
+                    });
         }
 
     }

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/stats/MetadataStoreStats.java

@@ -0,0 +1,92 @@
+/**
+ * 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
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * 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.metadata.impl.stats;
+
+import io.prometheus.client.Counter;
+import io.prometheus.client.Histogram;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+public final class MetadataStoreStats {
+    private static final AtomicBoolean INITIALIZED = new AtomicBoolean(false);
+    private static final double[] BUCKETS = new double[]{1, 3, 5, 10, 20, 50, 100, 200, 500, 1000, 2000, 5000};
+
+    private final Histogram getOpsLatency;
+    private final Histogram delOpsLatency;
+    private final Histogram putOpsLatency;
+    private final Counter getFailedCounter;
+    private final Counter delFailedCounter;
+    private final Counter putFailedCounter;
+    private final Counter putBytesCounter;
+
+    private MetadataStoreStats() {
+        getOpsLatency = Histogram.build("pulsar_metadata_store_get_ops_latency", "-")
+                .buckets(BUCKETS)
+                .register();
+        delOpsLatency = Histogram.build("pulsar_metadata_store_del_ops_latency", "-")
+                .buckets(BUCKETS)
+                .register();
+        putOpsLatency = Histogram.build("pulsar_metadata_store_put_ops_latency", "-")
+                .buckets(BUCKETS)
+                .register();
+
+        getFailedCounter = Counter.build("pulsar_metadata_store_get_ops_failed", "-")
+                .register();
+        delFailedCounter = Counter.build("pulsar_metadata_store_del_ops_failed", "-")
+                .register();
+        putFailedCounter = Counter.build("pulsar_metadata_store_put_ops_failed", "-")
+                .register();
+        putBytesCounter = Counter.build("pulsar_metadata_store_put_bytes", "-")
+                .register();
+    }
+
+    public void recordGetOpsLatency(long millis) {
+        this.getOpsLatency.observe(millis);
+    }
+
+    public void recordDelOpsLatency(long millis) {
+        this.delOpsLatency.observe(millis);
+    }
+
+    public void recordPutOpsLatency(long millis, int bytes) {
+        this.putOpsLatency.observe(millis);
+        this.putBytesCounter.inc(bytes);
+    }
+
+    public void recordGetOpsFailed() {
+        this.getFailedCounter.inc();
+    }
+
+    public void recordDelOpsFailed() {
+        this.delFailedCounter.inc();
+    }
+
+    public void recordPutOpsFailed() {
+        this.putFailedCounter.inc();
+    }
+
+    private static MetadataStoreStats instance;
+
+    public static MetadataStoreStats create() {
+        if (INITIALIZED.compareAndSet(false, true)) {
+            instance = new MetadataStoreStats();
+        }
+
+        return instance;
+    }
+}

pulsar-metadata/src/main/java/org/apache/pulsar/metadata/impl/stats/package-info.java

@@ -0,0 +1,19 @@
+/**
+ * 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.metadata.impl.stats;