[feat][misc] PIP-264: Implement topic lookup metrics using OpenTelemetry

pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java

@@ -24,7 +24,6 @@
 import static java.util.concurrent.TimeUnit.SECONDS;
 import static org.apache.commons.lang3.StringUtils.isNotBlank;
 import static org.apache.pulsar.common.naming.NamespaceName.SYSTEM_NAMESPACE;
-import com.google.common.annotations.VisibleForTesting;
 import com.google.common.hash.Hashing;
 import io.opentelemetry.api.common.AttributeKey;
 import io.opentelemetry.api.common.Attributes;
@@ -106,7 +105,6 @@
 import org.apache.pulsar.common.util.collections.ConcurrentOpenHashMap;
 import org.apache.pulsar.metadata.api.MetadataCache;
 import org.apache.pulsar.metadata.api.MetadataStoreException;
-import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes;
 import org.apache.pulsar.opentelemetry.annotations.PulsarDeprecatedMetric;
 import org.apache.pulsar.policies.data.loadbalancer.AdvertisedListener;
 import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData;
@@ -153,35 +151,37 @@ public class NamespaceService implements AutoCloseable {
 
     private final RedirectManager redirectManager;
 
-    @PulsarDeprecatedMetric(newMetricName = "pulsar.broker.lookup.request.duration")
+    public static final String LOOKUP_REQUEST_DURATION_METRIC_NAME = "pulsar.broker.request.topic.lookup.duration";
+
+    private static final AttributeKey<String> PULSAR_LOOKUP_RESPONSE_TYPE =
+            AttributeKey.stringKey("pulsar.lookup.response.type");
+    public static final Attributes PULSAR_LOOKUP_RESPONSE_BROKER_ATTRIBUTES = Attributes.builder()
+            .put(PULSAR_LOOKUP_RESPONSE_TYPE, "broker")
+            .build();
+    public static final Attributes PULSAR_LOOKUP_RESPONSE_REDIRECT_ATTRIBUTES = Attributes.builder()
+            .put(PULSAR_LOOKUP_RESPONSE_TYPE, "redirect")
+            .build();
+    public static final Attributes PULSAR_LOOKUP_RESPONSE_FAILURE_ATTRIBUTES = Attributes.builder()
+            .put(PULSAR_LOOKUP_RESPONSE_TYPE, "failure")
+            .build();
+
+    @PulsarDeprecatedMetric(newMetricName = LOOKUP_REQUEST_DURATION_METRIC_NAME)
     private static final Counter lookupRedirects = Counter.build("pulsar_broker_lookup_redirects", "-").register();
 
-    @PulsarDeprecatedMetric(newMetricName = "pulsar.broker.lookup.request.duration")
+    @PulsarDeprecatedMetric(newMetricName = LOOKUP_REQUEST_DURATION_METRIC_NAME)
     private static final Counter lookupFailures = Counter.build("pulsar_broker_lookup_failures", "-").register();
 
-    @PulsarDeprecatedMetric(newMetricName = "pulsar.broker.lookup.request.duration")
+    @PulsarDeprecatedMetric(newMetricName = LOOKUP_REQUEST_DURATION_METRIC_NAME)
     private static final Counter lookupAnswers = Counter.build("pulsar_broker_lookup_answers", "-").register();
 
-    @PulsarDeprecatedMetric(newMetricName = "pulsar.broker.lookup.request.duration")
+    @PulsarDeprecatedMetric(newMetricName = LOOKUP_REQUEST_DURATION_METRIC_NAME)
     private static final Summary lookupLatency = Summary.build("pulsar_broker_lookup", "-")
             .quantile(0.50)
             .quantile(0.99)
             .quantile(0.999)
             .quantile(1.0)
             .register();
     private final DoubleHistogram lookupLatencyHistogram;
-    private static final AttributeKey<String> PULSAR_LOOKUP_RESPONSE_TYPE =
-            AttributeKey.stringKey("pulsar.lookup.response.type");
-    @VisibleForTesting
-    public static final Attributes PULSAR_LOOKUP_RESPONSE_BROKER_ATTRIBUTES = Attributes.builder()
-            .putAll(OpenTelemetryAttributes.PULSAR_RESPONSE_STATUS_SUCCESS)
-            .put(PULSAR_LOOKUP_RESPONSE_TYPE, "broker")
-            .build();
-    @VisibleForTesting
-    public static final Attributes PULSAR_LOOKUP_RESPONSE_REDIRECT_ATTRIBUTES = Attributes.builder()
-            .putAll(OpenTelemetryAttributes.PULSAR_RESPONSE_STATUS_SUCCESS)
-            .put(PULSAR_LOOKUP_RESPONSE_TYPE, "redirect")
-            .build();
 
     /**
      * Default constructor.
@@ -201,7 +201,7 @@ public NamespaceService(PulsarService pulsar) {
         this.redirectManager = new RedirectManager(pulsar);
 
         this.lookupLatencyHistogram = pulsar.getOpenTelemetry().getMeter()
-                .histogramBuilder("pulsar.broker.lookup.request.duration")
+                .histogramBuilder(LOOKUP_REQUEST_DURATION_METRIC_NAME)
                 .setDescription("Lookup request duration")
                 .setUnit("s")
                 .build();
@@ -249,11 +249,11 @@ public CompletableFuture<Optional<LookupResult>> getBrokerServiceUrlAsync(TopicN
                     }
                 } else {
                     // No lookup result, default to reporting as failure.
-                    attributes = OpenTelemetryAttributes.PULSAR_RESPONSE_STATUS_FAILURE;
+                    attributes = PULSAR_LOOKUP_RESPONSE_FAILURE_ATTRIBUTES;
                 }
             } else {
                 lookupFailures.inc();
-                attributes = OpenTelemetryAttributes.PULSAR_RESPONSE_STATUS_FAILURE;
+                attributes = PULSAR_LOOKUP_RESPONSE_FAILURE_ATTRIBUTES;
             }
             lookupLatencyHistogram.record(MetricsUtil.convertToSeconds(latencyNs, TimeUnit.NANOSECONDS), attributes);
         });

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

@@ -38,7 +38,6 @@
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.ssl.SslContext;
 import io.netty.util.concurrent.DefaultThreadFactory;
-import io.opentelemetry.api.metrics.ObservableLongGauge;
 import io.opentelemetry.api.metrics.ObservableLongUpDownCounter;
 import io.prometheus.client.Histogram;
 import java.io.Closeable;
@@ -244,15 +243,19 @@ public class BrokerService implements Closeable {
     protected final AtomicReference<Semaphore> lookupRequestSemaphore;
     protected final AtomicReference<Semaphore> topicLoadRequestSemaphore;
 
-    @PulsarDeprecatedMetric(newMetricName = "pulsar.broker.topic.lookup.operation.pending.usage")
+    public static final String TOPIC_LOOKUP_USAGE_METRIC_NAME = "pulsar.broker.request.topic.lookup.concurrent.usage";
+    public static final String TOPIC_LOOKUP_LIMIT_METRIC_NAME = "pulsar.broker.request.topic.lookup.concurrent.limit";
+    @PulsarDeprecatedMetric(newMetricName = TOPIC_LOOKUP_USAGE_METRIC_NAME)
     private final ObserverGauge pendingLookupRequests;
     private final ObservableLongUpDownCounter pendingLookupOperationsCounter;
-    private final ObservableLongGauge pendingLookupOperationsLimitGauge;
+    private final ObservableLongUpDownCounter pendingLookupOperationsLimitCounter;
 
-    @PulsarDeprecatedMetric(newMetricName = "pulsar.broker.topic.load.operation.pending.usage")
+    public static final String TOPIC_LOAD_USAGE_METRIC_NAME = "pulsar.broker.topic.load.concurrent.usage";
+    public static final String TOPIC_LOAD_LIMIT_METRIC_NAME = "pulsar.broker.topic.load.concurrent.limit";
+    @PulsarDeprecatedMetric(newMetricName = TOPIC_LOAD_USAGE_METRIC_NAME)
     private final ObserverGauge pendingTopicLoadRequests;
     private final ObservableLongUpDownCounter pendingTopicLoadOperationsCounter;
-    private final ObservableLongGauge pendingTopicLoadOperationsLimitGauge;
+    private final ObservableLongUpDownCounter pendingTopicLoadOperationsLimitCounter;
 
     private final ScheduledExecutorService inactivityMonitor;
     private final ScheduledExecutorService messageExpiryMonitor;
@@ -415,15 +418,14 @@ public BrokerService(PulsarService pulsar, EventLoopGroup eventLoopGroup) throws
                 .supplier(this::getPendingLookupRequest)
                 .register();
         this.pendingLookupOperationsCounter = pulsar.getOpenTelemetry().getMeter()
-                .upDownCounterBuilder("pulsar.broker.topic.lookup.operation.pending.usage")
+                .upDownCounterBuilder(TOPIC_LOOKUP_USAGE_METRIC_NAME)
                 .setDescription("The number of pending lookup operations in the broker. "
                         + "When it reaches threshold \"maxConcurrentLookupRequest\" defined in broker.conf, "
                         + "new requests are rejected.")
                 .setUnit("{operation}")
                 .buildWithCallback(measurement -> measurement.record(getPendingLookupRequest()));
-        this.pendingLookupOperationsLimitGauge = pulsar.getOpenTelemetry().getMeter()
-                .gaugeBuilder("pulsar.broker.topic.lookup.operation.pending.limit")
-                .ofLongs()
+        this.pendingLookupOperationsLimitCounter = pulsar.getOpenTelemetry().getMeter()
+                .upDownCounterBuilder(TOPIC_LOOKUP_LIMIT_METRIC_NAME)
                 .setDescription("The maximum number of pending lookup operations in the broker. "
                         + "Equal to \"maxConcurrentLookupRequest\" defined in broker.conf.")
                 .setUnit("{operation}")
@@ -435,15 +437,14 @@ public BrokerService(PulsarService pulsar, EventLoopGroup eventLoopGroup) throws
                 .supplier(this::getPendingTopicLoadRequests)
                 .register();
         this.pendingTopicLoadOperationsCounter = pulsar.getOpenTelemetry().getMeter()
-                .upDownCounterBuilder("pulsar.broker.topic.load.operation.pending.usage")
+                .upDownCounterBuilder(TOPIC_LOAD_USAGE_METRIC_NAME)
                 .setDescription("The number of pending topic load operations in the broker. "
                         + "When it reaches threshold \"maxConcurrentTopicLoadRequest\" defined in broker.conf, "
                         + "new requests are rejected.")
                 .setUnit("{operation}")
                 .buildWithCallback(measurement -> measurement.record(getPendingTopicLoadRequests()));
-        this.pendingTopicLoadOperationsLimitGauge = pulsar.getOpenTelemetry().getMeter()
-                .gaugeBuilder("pulsar.broker.topic.load.operation.pending.limit")
-                .ofLongs()
+        this.pendingTopicLoadOperationsLimitCounter = pulsar.getOpenTelemetry().getMeter()
+                .upDownCounterBuilder(TOPIC_LOAD_LIMIT_METRIC_NAME)
                 .setDescription("The maximum number of pending topic load operations in the broker. "
                         + "Equal to \"maxConcurrentTopicLoadRequest\" defined in broker.conf.")
                 .setUnit("{operation}")

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/BrokerServiceThrottlingTest.java

@@ -82,14 +82,14 @@ protected void customizeMainPulsarTestContextBuilder(PulsarTestContext.Builder b
     public void testThrottlingLookupRequestSemaphore() throws Exception {
         var lookupRequestSemaphore = pulsar.getBrokerService().lookupRequestSemaphore;
         var configName = "maxConcurrentLookupRequest";
-        var metricName = "pulsar.broker.topic.lookup.operation.pending.limit";
+        var metricName = BrokerService.TOPIC_LOOKUP_LIMIT_METRIC_NAME;
         // Validate that the configuration has not been overridden.
         assertThat(admin.brokers().getAllDynamicConfigurations()).doesNotContainKey(configName);
-        assertLongGaugeValue(metricName, 50_000);
+        assertLongSumValue(metricName, 50_000);
         assertThat(lookupRequestSemaphore.get().availablePermits()).isNotEqualTo(0);
         admin.brokers().updateDynamicConfiguration(configName, Integer.toString(0));
         waitAtMost(1, TimeUnit.SECONDS).until(() -> lookupRequestSemaphore.get().availablePermits() == 0);
-        assertLongGaugeValue(metricName, 0);
+        assertLongSumValue(metricName, 0);
     }
 
     /**
@@ -100,22 +100,22 @@ public void testThrottlingLookupRequestSemaphore() throws Exception {
     public void testThrottlingTopicLoadRequestSemaphore() throws Exception {
         var topicLoadRequestSemaphore = pulsar.getBrokerService().topicLoadRequestSemaphore;
         var configName = "maxConcurrentTopicLoadRequest";
-        var metricName = "pulsar.broker.topic.load.operation.pending.limit";
+        var metricName = BrokerService.TOPIC_LOAD_LIMIT_METRIC_NAME;
         // Validate that the configuration has not been overridden.
         assertThat(admin.brokers().getAllDynamicConfigurations()).doesNotContainKey(configName);
-        assertLongGaugeValue(metricName, 5_000);
+        assertLongSumValue(metricName, 5_000);
         assertThat(topicLoadRequestSemaphore.get().availablePermits()).isNotEqualTo(0);
         admin.brokers().updateDynamicConfiguration(configName, Integer.toString(0));
         waitAtMost(1, TimeUnit.SECONDS).until(() -> topicLoadRequestSemaphore.get().availablePermits() == 0);
-        assertLongGaugeValue(metricName, 0);
+        assertLongSumValue(metricName, 0);
     }
 
-    private void assertLongGaugeValue(String metricName, int value) {
+    private void assertLongSumValue(String metricName, int value) {
         assertThat(pulsarTestContext.getOpenTelemetryMetricReader().collectAllMetrics())
                 .anySatisfy(metric -> assertThat(metric)
                         .hasName(metricName)
-                        .hasLongGaugeSatisfying(
-                                gauge -> gauge.hasPointsSatisfying(point -> point.hasValue(value))));
+                        .hasLongSumSatisfying(
+                                sum -> sum.hasPointsSatisfying(point -> point.hasValue(value))));
     }
 
     /**

pulsar-broker/src/test/java/org/apache/pulsar/client/api/BrokerServiceLookupTest.java

@@ -19,6 +19,7 @@
 package org.apache.pulsar.client.api;
 
 import static io.opentelemetry.sdk.testing.assertj.OpenTelemetryAssertions.assertThat;
+import static org.apache.pulsar.broker.namespace.NamespaceService.LOOKUP_REQUEST_DURATION_METRIC_NAME;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doReturn;
@@ -94,7 +95,6 @@
 import org.apache.pulsar.common.policies.data.TenantInfoImpl;
 import org.apache.pulsar.common.util.ObjectMapperFactory;
 import org.apache.pulsar.common.util.SecurityUtility;
-import org.apache.pulsar.opentelemetry.OpenTelemetryAttributes;
 import org.apache.pulsar.policies.data.loadbalancer.LocalBrokerData;
 import org.apache.pulsar.policies.data.loadbalancer.NamespaceBundleStats;
 import org.apache.zookeeper.KeeperException;
@@ -223,9 +223,8 @@ public void testMultipleBrokerLookup() throws Exception {
         topicLoadRequestSemaphoreField.set(pulsar2.getBrokerService(),
                 new AtomicReference<>(topicLoadRequestSemaphoreSpy));
 
-        var metricName = "pulsar.broker.lookup.request.duration";
         assertThat(pulsarTestContext.getOpenTelemetryMetricReader().collectAllMetrics())
-                .noneSatisfy(metric -> assertThat(metric).hasName("pulsar.broker.lookup.request.duration"));
+                .noneSatisfy(metric -> assertThat(metric).hasName(LOOKUP_REQUEST_DURATION_METRIC_NAME));
 
         /**** started broker-2 ****/
         @Cleanup
@@ -237,15 +236,15 @@ public void testMultipleBrokerLookup() throws Exception {
         cdlAfterLookupSemaphoreAcquire.await();
         assertThat(metricReader.collectAllMetrics())
                 .anySatisfy(metric -> assertThat(metric)
-                        .hasName("pulsar.broker.topic.lookup.operation.pending.usage")
+                        .hasName(BrokerService.TOPIC_LOOKUP_USAGE_METRIC_NAME)
                         .hasLongSumSatisfying(
                                 sum -> sum.hasPointsSatisfying(point -> point.hasValue(1))));
         cdlLookupSemaphoreVerification.countDown();
 
         cdlAfterTopicLoadSemaphoreAcquire.await();
         assertThat(pulsarTestContext2.getOpenTelemetryMetricReader().collectAllMetrics())
                 .anySatisfy(metric -> assertThat(metric)
-                        .hasName("pulsar.broker.topic.load.operation.pending.usage")
+                        .hasName(BrokerService.TOPIC_LOAD_USAGE_METRIC_NAME)
                         .hasLongSumSatisfying(
                                 sum -> sum.hasPointsSatisfying(point -> point.hasValue(1))));
         cdlTopicLoadSemaphoreVerification.countDown();
@@ -274,7 +273,7 @@ public void testMultipleBrokerLookup() throws Exception {
         var metrics = metricReader.collectAllMetrics();
         assertThat(metrics)
                 .anySatisfy(metric -> assertThat(metric)
-                        .hasName(metricName)
+                        .hasName(LOOKUP_REQUEST_DURATION_METRIC_NAME)
                         .hasHistogramSatisfying(histogram -> histogram.hasPointsSatisfying(
                                 point -> point
                                         .hasAttributes(NamespaceService.PULSAR_LOOKUP_RESPONSE_REDIRECT_ATTRIBUTES)
@@ -1201,14 +1200,13 @@ public void testLookupConnectionNotCloseIfGetUnloadingExOrMetadataEx() throws Ex
         assertTrue(lookupService instanceof BinaryProtoLookupService);
         ClientCnx lookupConnection = pulsarClientImpl.getCnxPool().getConnection(lookupService.resolveHost()).join();
 
-        var metricName = "pulsar.broker.lookup.request.duration";
         var metricReader = pulsarTestContext.getOpenTelemetryMetricReader();
         assertThat(metricReader.collectAllMetrics())
                 .noneSatisfy(metric -> assertThat(metric)
-                        .hasName(metricName)
+                        .hasName(LOOKUP_REQUEST_DURATION_METRIC_NAME)
                         .hasHistogramSatisfying(histogram -> histogram.hasPointsSatisfying(
                                 point -> point
-                                        .hasAttributes(OpenTelemetryAttributes.PULSAR_RESPONSE_STATUS_FAILURE),
+                                        .hasAttributes(NamespaceService.PULSAR_LOOKUP_RESPONSE_FAILURE_ATTRIBUTES),
                                 point -> point
                                         .hasAttributes(NamespaceService.PULSAR_LOOKUP_RESPONSE_BROKER_ATTRIBUTES))));
 
@@ -1224,10 +1222,10 @@ public void testLookupConnectionNotCloseIfGetUnloadingExOrMetadataEx() throws Ex
 
         assertThat(metricReader.collectAllMetrics())
                 .anySatisfy(metric -> assertThat(metric)
-                        .hasName(metricName)
+                        .hasName(LOOKUP_REQUEST_DURATION_METRIC_NAME)
                         .hasHistogramSatisfying(histogram -> histogram.hasPointsSatisfying(
                                 point -> point
-                                        .hasAttributes(OpenTelemetryAttributes.PULSAR_RESPONSE_STATUS_FAILURE)
+                                        .hasAttributes(NamespaceService.PULSAR_LOOKUP_RESPONSE_FAILURE_ATTRIBUTES)
                                         .hasCount(1),
                                 point -> point
                                         .hasAttributes(NamespaceService.PULSAR_LOOKUP_RESPONSE_BROKER_ATTRIBUTES))));

pulsar-opentelemetry/src/main/java/org/apache/pulsar/opentelemetry/OpenTelemetryAttributes.java

@@ -19,7 +19,6 @@
 package org.apache.pulsar.opentelemetry;
 
 import io.opentelemetry.api.common.AttributeKey;
-import io.opentelemetry.api.common.Attributes;
 
 /**
  * Common OpenTelemetry attributes to be used by Pulsar components.
@@ -30,8 +29,4 @@ public interface OpenTelemetryAttributes {
      * {@link OpenTelemetryService}.
      */
     AttributeKey<String> PULSAR_CLUSTER = AttributeKey.stringKey("pulsar.cluster");
-
-    AttributeKey<String> PULSAR_RESPONSE_STATUS = AttributeKey.stringKey("pulsar.response.status");
-    Attributes PULSAR_RESPONSE_STATUS_SUCCESS = Attributes.of(PULSAR_RESPONSE_STATUS, "success");
-    Attributes PULSAR_RESPONSE_STATUS_FAILURE = Attributes.of(PULSAR_RESPONSE_STATUS, "failure");
 }