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

pulsar-broker/pom.xml

@@ -484,6 +484,11 @@
       <version>${project.version}</version>
     </dependency>
 
+    <dependency>
+      <groupId>io.opentelemetry</groupId>
+      <artifactId>opentelemetry-sdk-testing</artifactId>
+      <scope>test</scope>
+    </dependency>
   </dependencies>
 
   <build>

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

@@ -249,7 +249,8 @@ public class PulsarService implements AutoCloseable, ShutdownService {
     private final Timer brokerClientSharedTimer;
 
     private MetricsGenerator metricsGenerator;
-    private PulsarBrokerOpenTelemetry openTelemetry;
+    @VisibleForTesting
+    protected PulsarBrokerOpenTelemetry openTelemetry;
 
     private TransactionMetadataStoreService transactionMetadataStoreService;
     private TransactionBufferProvider transactionBufferProvider;
@@ -312,6 +313,8 @@ public PulsarService(ServiceConfiguration config,
         TransactionBatchedWriteValidator.validate(config);
         this.config = config;
 
+        this.openTelemetry = new PulsarBrokerOpenTelemetry(config);
+
         // validate `advertisedAddress`, `advertisedListeners`, `internalListenerName`
         this.advertisedListeners = MultipleListenerValidator.validateAndAnalysisAdvertisedListener(config);
 
@@ -902,7 +905,6 @@ public void start() throws PulsarServerException {
             }
 
             this.metricsGenerator = new MetricsGenerator(this);
-            this.openTelemetry = new PulsarBrokerOpenTelemetry(config);
 
             // Initialize the message protocol handlers.
             // start the protocol handlers only after the broker is ready,

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

@@ -25,6 +25,8 @@
 import static org.apache.commons.lang3.StringUtils.isNotBlank;
 import static org.apache.pulsar.common.naming.NamespaceName.SYSTEM_NAMESPACE;
 import com.google.common.hash.Hashing;
+import io.opentelemetry.api.metrics.DoubleHistogram;
+import io.opentelemetry.api.metrics.LongCounter;
 import io.prometheus.client.Counter;
 import java.net.URI;
 import java.net.URL;
@@ -146,18 +148,24 @@ public class NamespaceService implements AutoCloseable {
 
     private final RedirectManager redirectManager;
 
-
+    /** @deprecated by {@link #lookupRedirectsCounter} */
     private static final Counter lookupRedirects = Counter.build("pulsar_broker_lookup_redirects", "-").register();
+    private final LongCounter lookupRedirectsCounter;
+    /** @deprecated by {@link #lookupFailuresCounter} */
     private static final Counter lookupFailures = Counter.build("pulsar_broker_lookup_failures", "-").register();
+    private final LongCounter lookupFailuresCounter;
+    /** @deprecated by {@link #lookupAnswersCounter} */
     private static final Counter lookupAnswers = Counter.build("pulsar_broker_lookup_answers", "-").register();
+    private final LongCounter lookupAnswersCounter;
 
+    /** @deprecated by {@link #lookupLatencyHistogram} */
     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;
 
     /**
      * Default constructor.
@@ -175,6 +183,25 @@ public NamespaceService(PulsarService pulsar) {
         this.bundleSplitListeners = new CopyOnWriteArrayList<>();
         this.localBrokerDataCache = pulsar.getLocalMetadataStore().getMetadataCache(LocalBrokerData.class);
         this.redirectManager = new RedirectManager(pulsar);
+
+        var meter = pulsar.getOpenTelemetry().getMeter();
+        this.lookupRedirectsCounter = meter
+                .counterBuilder("pulsar.broker.lookup.redirect")
+                .setDescription("The number of lookup redirected requests")
+                .build();
+        this.lookupFailuresCounter = meter
+                .counterBuilder("pulsar.broker.lookup.failure")
+                .setDescription("The number of lookup failures")
+                .build();
+        this.lookupAnswersCounter = meter
+                .counterBuilder("pulsar.broker.lookup.answer")
+                .setDescription("The number of lookup responses (i.e. not redirected requests)")
+                .build();
+        this.lookupLatencyHistogram = meter
+                .histogramBuilder("pulsar.broker.lookup.latency")
+                .setDescription("Lookup request latency")
+                .setUnit("s")
+                .build();
     }
 
     public void initialize() {
@@ -205,16 +232,21 @@ public CompletableFuture<Optional<LookupResult>> getBrokerServiceUrlAsync(TopicN
                 });
 
         future.thenAccept(optResult -> {
-            lookupLatency.observe(System.nanoTime() - startTime, TimeUnit.NANOSECONDS);
+            var latencyNs = System.nanoTime() - startTime;
+            lookupLatency.observe(latencyNs, TimeUnit.NANOSECONDS);
+            lookupLatencyHistogram.record(latencyNs / 1_000_000_000.0);
             if (optResult.isPresent()) {
                 if (optResult.get().isRedirect()) {
                     lookupRedirects.inc();
+                    lookupRedirectsCounter.add(1);
                 } else {
                     lookupAnswers.inc();
+                    lookupAnswersCounter.add(1);
                 }
             }
         }).exceptionally(ex -> {
             lookupFailures.inc();
+            lookupFailuresCounter.add(1);
             return null;
         });
 

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

@@ -38,6 +38,7 @@
 import io.netty.channel.socket.SocketChannel;
 import io.netty.handler.ssl.SslContext;
 import io.netty.util.concurrent.DefaultThreadFactory;
+import io.opentelemetry.api.metrics.ObservableLongUpDownCounter;
 import io.prometheus.client.Histogram;
 import java.io.Closeable;
 import java.io.IOException;
@@ -241,8 +242,13 @@ public class BrokerService implements Closeable {
     protected final AtomicReference<Semaphore> lookupRequestSemaphore;
     protected final AtomicReference<Semaphore> topicLoadRequestSemaphore;
 
+    /** @deprecated by {@link #pendingLookupRequestsCounter} */
     private final ObserverGauge pendingLookupRequests;
+    private final ObservableLongUpDownCounter pendingLookupRequestsCounter;
+
+    /** @deprecated by {@link #pendingTopicLoadRequestsCounter} */
     private final ObserverGauge pendingTopicLoadRequests;
+    private final ObservableLongUpDownCounter pendingTopicLoadRequestsCounter;
 
     private final ScheduledExecutorService inactivityMonitor;
     private final ScheduledExecutorService messageExpiryMonitor;
@@ -403,15 +409,25 @@ public BrokerService(PulsarService pulsar, EventLoopGroup eventLoopGroup) throws
         this.defaultServerBootstrap = defaultServerBootstrap();
 
         this.pendingLookupRequests = ObserverGauge.build("pulsar_broker_lookup_pending_requests", "-")
-                .supplier(() -> pulsar.getConfig().getMaxConcurrentLookupRequest()
-                        - lookupRequestSemaphore.get().availablePermits())
+                .supplier(this::getPendingLookupRequest)
                 .register();
+        this.pendingLookupRequestsCounter = pulsar.getOpenTelemetry().getMeter()
+                .upDownCounterBuilder("pulsar.broker.lookup.pending.request")
+                .setDescription("The number of pending lookup requests in the broker. "
+                        + "When it reaches threshold \"maxConcurrentLookupRequest\" defined in broker.conf, "
+                        + "new requests are rejected.")
+                .buildWithCallback(measurement -> measurement.record(getPendingLookupRequest()));
 
         this.pendingTopicLoadRequests = ObserverGauge.build(
                 "pulsar_broker_topic_load_pending_requests", "-")
-                .supplier(() -> pulsar.getConfig().getMaxConcurrentTopicLoadRequest()
-                        - topicLoadRequestSemaphore.get().availablePermits())
+                .supplier(this::getPendingTopicLoadRequests)
                 .register();
+        this.pendingTopicLoadRequestsCounter = pulsar.getOpenTelemetry().getMeter()
+                .upDownCounterBuilder("pulsar.broker.topic.load.pending.request")
+                .setDescription("The number of pending topic load operations in the broker. "
+                        + "When it reaches threshold \"maxConcurrentTopicLoadRequest\" defined in broker.conf, "
+                        + "new requests are rejected.")
+                .buildWithCallback(measurement -> measurement.record(getPendingTopicLoadRequests()));
 
         this.brokerEntryMetadataInterceptors = BrokerEntryMetadataUtils
                 .loadBrokerEntryMetadataInterceptors(pulsar.getConfiguration().getBrokerEntryMetadataInterceptors(),
@@ -423,6 +439,15 @@ public BrokerService(PulsarService pulsar, EventLoopGroup eventLoopGroup) throws
         this.bundlesQuotas = new BundlesQuotas(pulsar);
     }
 
+    private int getPendingLookupRequest() {
+        return pulsar.getConfig().getMaxConcurrentLookupRequest() - lookupRequestSemaphore.get().availablePermits();
+    }
+
+    private int getPendingTopicLoadRequests() {
+        return pulsar.getConfig().getMaxConcurrentTopicLoadRequest()
+                - topicLoadRequestSemaphore.get().availablePermits();
+    }
+
     public void addTopicEventListener(TopicEventsListener... listeners) {
         topicEventsDispatcher.addTopicEventListener(listeners);
         getTopics().keys().forEach(topic ->
@@ -780,6 +805,8 @@ public CompletableFuture<Void> closeAsync() {
                                     log.warn("Error in closing authenticationService", e);
                                 }
                                 pulsarStats.close();
+                                pendingTopicLoadRequestsCounter.close();
+                                pendingLookupRequestsCounter.close();
                                 try {
                                     delayedDeliveryTrackerFactory.close();
                                 } catch (Exception e) {

pulsar-broker/src/main/java/org/apache/pulsar/broker/stats/PulsarBrokerOpenTelemetry.java

@@ -18,8 +18,11 @@
  */
 package org.apache.pulsar.broker.stats;
 
+import com.google.common.annotations.VisibleForTesting;
 import io.opentelemetry.api.metrics.Meter;
+import io.opentelemetry.sdk.autoconfigure.AutoConfiguredOpenTelemetrySdkBuilder;
 import java.io.Closeable;
+import java.util.function.Consumer;
 import lombok.Getter;
 import org.apache.pulsar.PulsarVersion;
 import org.apache.pulsar.broker.ServiceConfiguration;
@@ -34,10 +37,17 @@ public class PulsarBrokerOpenTelemetry implements Closeable {
     private final Meter meter;
 
     public PulsarBrokerOpenTelemetry(ServiceConfiguration config) {
+        this(config, null);
+    }
+
+    @VisibleForTesting
+    public PulsarBrokerOpenTelemetry(ServiceConfiguration config,
+                                     Consumer<AutoConfiguredOpenTelemetrySdkBuilder> builderCustomizer) {
         openTelemetryService = OpenTelemetryService.builder()
                 .clusterName(config.getClusterName())
                 .serviceName(SERVICE_NAME)
                 .serviceVersion(PulsarVersion.getVersion())
+                .sdkBuilderConsumer(builderCustomizer)
                 .build();
         meter = openTelemetryService.getOpenTelemetry().getMeter("org.apache.pulsar.broker");
     }

pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java

@@ -448,19 +448,27 @@ protected PulsarTestContext.Builder createPulsarTestContextBuilder(ServiceConfig
         return builder;
     }
 
+    protected PulsarTestContext createAdditionalPulsarTestContext(ServiceConfiguration conf) throws Exception {
+        return createAdditionalPulsarTestContext(conf, null);
+    }
     /**
      * This method can be used in test classes for creating additional PulsarTestContext instances
      * that share the same mock ZooKeeper and BookKeeper instances as the main PulsarTestContext instance.
      *
      * @param conf the ServiceConfiguration instance to use
+     * @param builderCustomizer a consumer that can be used to customize the builder configuration
      * @return the PulsarTestContext instance
      * @throws Exception if an error occurs
      */
-    protected PulsarTestContext createAdditionalPulsarTestContext(ServiceConfiguration conf) throws Exception {
-        return createPulsarTestContextBuilder(conf)
+    protected PulsarTestContext createAdditionalPulsarTestContext(ServiceConfiguration conf,
+                                              Consumer<PulsarTestContext.Builder> builderCustomizer) throws Exception {
+        var builder = createPulsarTestContextBuilder(conf)
                 .reuseMockBookkeeperAndMetadataStores(pulsarTestContext)
-                .reuseSpyConfig(pulsarTestContext)
-                .build();
+                .reuseSpyConfig(pulsarTestContext);
+        if (builderCustomizer != null) {
+            builderCustomizer.accept(builder);
+        }
+        return builder.build();
     }
 
     protected void waitForZooKeeperWatchers() {

pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/PulsarTestContext.java

@@ -21,11 +21,13 @@
 
 import com.google.common.util.concurrent.MoreExecutors;
 import io.netty.channel.EventLoopGroup;
+import io.opentelemetry.sdk.testing.exporter.InMemoryMetricReader;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.function.Consumer;
 import java.util.function.Function;
@@ -52,6 +54,7 @@
 import org.apache.pulsar.broker.resources.TopicResources;
 import org.apache.pulsar.broker.service.BrokerService;
 import org.apache.pulsar.broker.service.ServerCnx;
+import org.apache.pulsar.broker.stats.PulsarBrokerOpenTelemetry;
 import org.apache.pulsar.broker.storage.ManagedLedgerStorage;
 import org.apache.pulsar.common.util.GracefulExecutorServicesShutdown;
 import org.apache.pulsar.common.util.PortManager;
@@ -64,6 +67,7 @@
 import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
 import org.apache.pulsar.metadata.impl.MetadataStoreFactoryImpl;
 import org.apache.pulsar.metadata.impl.ZKMetadataStore;
+import org.apache.pulsar.opentelemetry.OpenTelemetryService;
 import org.apache.zookeeper.CreateMode;
 import org.apache.zookeeper.MockZooKeeper;
 import org.apache.zookeeper.MockZooKeeperSession;
@@ -160,6 +164,8 @@ public class PulsarTestContext implements AutoCloseable {
 
     private final boolean preallocatePorts;
 
+    private final boolean enableOpenTelemetry;
+    private final InMemoryMetricReader openTelemetryMetricReader;
 
     public ManagedLedgerFactory getManagedLedgerFactory() {
         return managedLedgerClientFactory.getManagedLedgerFactory();
@@ -727,11 +733,24 @@ protected void initializePulsarServices(SpyConfig spyConfig, Builder builder) {
                     .equals(PulsarCompactionServiceFactory.class.getName())) {
                 compactionServiceFactory = new MockPulsarCompactionServiceFactory(spyConfig, builder.compactor);
             }
+            PulsarBrokerOpenTelemetry pulsarBrokerOpenTelemetry;
+            if (builder.enableOpenTelemetry) {
+                var reader = InMemoryMetricReader.create();
+                pulsarBrokerOpenTelemetry = new PulsarBrokerOpenTelemetry(builder.config, builderCustomizer -> {
+                    builderCustomizer.addMeterProviderCustomizer(
+                            (meterProviderBuilder, __) -> meterProviderBuilder.registerMetricReader(reader));
+                    builderCustomizer.addPropertiesSupplier(
+                            () -> Map.of(OpenTelemetryService.OTEL_SDK_DISABLED_KEY, "false"));
+                });
+                openTelemetryMetricReader(reader);
+            } else {
+                pulsarBrokerOpenTelemetry = null;
+            }
             PulsarService pulsarService = spyConfig.getPulsarService()
                     .spy(StartableTestPulsarService.class, spyConfig, builder.config, builder.localMetadataStore,
                             builder.configurationMetadataStore, compactionServiceFactory,
                             builder.brokerInterceptor,
-                            bookKeeperClientFactory, builder.brokerServiceCustomizer);
+                            bookKeeperClientFactory, builder.brokerServiceCustomizer, pulsarBrokerOpenTelemetry);
             if (compactionServiceFactory != null) {
                 compactionServiceFactory.initialize(pulsarService);
             }

pulsar-broker/src/test/java/org/apache/pulsar/broker/testcontext/StartableTestPulsarService.java

@@ -28,6 +28,7 @@
 import org.apache.pulsar.broker.intercept.BrokerInterceptor;
 import org.apache.pulsar.broker.namespace.NamespaceService;
 import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.stats.PulsarBrokerOpenTelemetry;
 import org.apache.pulsar.compaction.CompactionServiceFactory;
 import org.apache.pulsar.metadata.api.extended.MetadataStoreExtended;
 
@@ -44,10 +45,16 @@ public StartableTestPulsarService(SpyConfig spyConfig, ServiceConfiguration conf
                                       CompactionServiceFactory compactionServiceFactory,
                                       BrokerInterceptor brokerInterceptor,
                                       BookKeeperClientFactory bookKeeperClientFactory,
-                                      Function<BrokerService, BrokerService> brokerServiceCustomizer) {
+                                      Function<BrokerService, BrokerService> brokerServiceCustomizer,
+                                      PulsarBrokerOpenTelemetry openTelemetry) {
         super(spyConfig, config, localMetadataStore, configurationMetadataStore, compactionServiceFactory,
                 brokerInterceptor, bookKeeperClientFactory);
         this.brokerServiceCustomizer = brokerServiceCustomizer;
+        if (openTelemetry != null) {
+            // Replace existing OpenTelemetry wrapper class.
+            this.openTelemetry.close();
+            this.openTelemetry = openTelemetry;
+        }
     }
 
     @Override
@@ -59,4 +66,4 @@ protected BrokerService newBrokerService(PulsarService pulsar) throws Exception
     public Supplier<NamespaceService> getNamespaceServiceProvider() throws PulsarServerException {
         return () -> spyConfig.getNamespaceService().spy(NamespaceService.class, this);
     }
-}
+}