loadbalancer-experimental: thread lbDescription into the LoadBalancer…

…Observer (#2936)

Motivation:

We already have a standard resource identifier for use when logging
load balancer concerns: lbDescrption. However, we don't use that for
the LoadBalancerObserver type because we don't know it on construction.

Modifications:

- Change the LoadBalancerBuilder to take a factory of LoadBalancerObservers
  which will consume the lbDescription during load balancer construction.
- Deprecate the older API for a release cycle.

Result:

Better logging.

servicetalk-loadbalancer-experimental-provider/src/main/java/io/servicetalk/loadbalancer/experimental/DefaultHttpLoadBalancerProvider.java

@@ -49,7 +49,7 @@ public final <U, R> SingleAddressHttpClientBuilder<U, R> newBuilder(U address,
         if (config.enabledForServiceName(serviceName)) {
             try {
                 HttpLoadBalancerFactory<R> loadBalancerFactory = new DefaultHttpLoadBalancerFactory<>(
-                        defaultLoadBalancer(serviceName, config));
+                        defaultLoadBalancer(config));
                 builder = builder.loadBalancerFactory(loadBalancerFactory);
                 builder = new LoadBalancerIgnoringBuilder<>(builder, serviceName);
                 LOGGER.info("Enabled DefaultLoadBalancer for service with name {}", serviceName);
@@ -61,10 +61,10 @@ public final <U, R> SingleAddressHttpClientBuilder<U, R> newBuilder(U address,
     }
 
     private <R> LoadBalancerFactory<R, FilterableStreamingHttpLoadBalancedConnection> defaultLoadBalancer(
-            String serviceName, DefaultLoadBalancerProviderConfig config) {
+            DefaultLoadBalancerProviderConfig config) {
         return LoadBalancers.<R, FilterableStreamingHttpLoadBalancedConnection>
                         builder("experimental-load-balancer")
-                .loadBalancerObserver(new DefaultLoadBalancerObserver(serviceName))
+                .loadBalancerObserver(DefaultLoadBalancerObserver::new)
                 // set up the new features.
                 .outlierDetectorConfig(config.outlierDetectorConfig())
                 .loadBalancingPolicy(config.getLoadBalancingPolicy())

servicetalk-loadbalancer-experimental-provider/src/main/java/io/servicetalk/loadbalancer/experimental/DefaultLoadBalancerObserver.java

@@ -31,10 +31,10 @@ final class DefaultLoadBalancerObserver implements LoadBalancerObserver {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(DefaultLoadBalancerObserver.class);
 
-    private final String clientName;
+    private final String lbDescription;
 
-    DefaultLoadBalancerObserver(final String clientName) {
-        this.clientName = requireNonNull(clientName, "clientName");
+    DefaultLoadBalancerObserver(final String lbDescription) {
+        this.lbDescription = requireNonNull(lbDescription, "lbDescription");
     }
 
     @Override
@@ -44,19 +44,19 @@ public HostObserver hostObserver(Object resolvedAddress) {
 
     @Override
     public void onNoHostsAvailable() {
-        LOGGER.debug("{}- onNoHostsAvailable()", clientName);
+        LOGGER.debug("{}- onNoHostsAvailable()", lbDescription);
     }
 
     @Override
     public void onServiceDiscoveryEvent(Collection<? extends ServiceDiscovererEvent<?>> events, int oldHostSetSize,
                                         int newHostSetSize) {
         LOGGER.debug("{}- onServiceDiscoveryEvent(events: {}, oldHostSetSize: {}, newHostSetSize: {})",
-                clientName, events, oldHostSetSize, newHostSetSize);
+                lbDescription, events, oldHostSetSize, newHostSetSize);
     }
 
     @Override
     public void onNoActiveHostsAvailable(int hostSetSize, NoActiveHostException exception) {
-        LOGGER.debug("{}- No active hosts available. Host set size: {}.", clientName, hostSetSize, exception);
+        LOGGER.debug("{}- No active hosts available. Host set size: {}.", lbDescription, hostSetSize, exception);
     }
 
     private final class HostObserverImpl implements HostObserver {
@@ -70,35 +70,35 @@ private final class HostObserverImpl implements HostObserver {
         @Override
         public void onHostMarkedExpired(int connectionCount) {
             LOGGER.debug("{}:{}- onHostMarkedExpired(connectionCount: {})",
-                    clientName, resolvedAddress, connectionCount);
+                    lbDescription, resolvedAddress, connectionCount);
         }
 
         @Override
         public void onActiveHostRemoved(int connectionCount) {
             LOGGER.debug("{}:{}- onActiveHostRemoved(connectionCount: {})",
-                    clientName, resolvedAddress, connectionCount);
+                    lbDescription, resolvedAddress, connectionCount);
         }
 
         @Override
         public void onExpiredHostRevived(int connectionCount) {
             LOGGER.debug("{}:{}- onExpiredHostRevived(connectionCount: {})",
-                    clientName, resolvedAddress, connectionCount);
+                    lbDescription, resolvedAddress, connectionCount);
         }
 
         @Override
         public void onExpiredHostRemoved(int connectionCount) {
             LOGGER.debug("{}:{}- onExpiredHostRemoved(connectionCount: {})",
-                    clientName, resolvedAddress, connectionCount);
+                    lbDescription, resolvedAddress, connectionCount);
         }
 
         @Override
         public void onHostMarkedUnhealthy(@Nullable Throwable cause) {
-            LOGGER.debug("{}:{}- onHostMarkedUnhealthy(ex)", clientName, resolvedAddress, cause);
+            LOGGER.debug("{}:{}- onHostMarkedUnhealthy(ex)", lbDescription, resolvedAddress, cause);
         }
 
         @Override
         public void onHostRevived() {
-            LOGGER.debug("{}:{}- onHostRevived()", clientName, resolvedAddress);
+            LOGGER.debug("{}:{}- onHostRevived()", lbDescription, resolvedAddress);
         }
     }
 }

servicetalk-loadbalancer-experimental/src/main/java/io/servicetalk/loadbalancer/DefaultLoadBalancer.java

@@ -132,7 +132,7 @@ final class DefaultLoadBalancer<ResolvedAddress, C extends LoadBalancedConnectio
             final HostSelector<ResolvedAddress, C> hostSelector,
             final ConnectionPoolStrategy<C> connectionPoolStrategy,
             final ConnectionFactory<ResolvedAddress, ? extends C> connectionFactory,
-            final LoadBalancerObserver loadBalancerObserver,
+            final LoadBalancerObserverFactory loadBalancerObserverFactory,
             @Nullable final HealthCheckConfig healthCheckConfig,
             final Function<String, OutlierDetector<ResolvedAddress, C>> outlierDetectorFactory) {
         this.targetResource = requireNonNull(targetResourceName);
@@ -144,7 +144,8 @@ final class DefaultLoadBalancer<ResolvedAddress, C extends LoadBalancedConnectio
         this.eventStream = fromSource(eventStreamProcessor)
                 .replay(1); // Allow for multiple subscribers and provide new subscribers with last signal.
         this.connectionFactory = requireNonNull(connectionFactory);
-        this.loadBalancerObserver = requireNonNull(loadBalancerObserver, "loadBalancerObserver");
+        this.loadBalancerObserver = requireNonNull(loadBalancerObserverFactory, "loadBalancerObserverFactory")
+                .newObserver(lbDescription);
         this.healthCheckConfig = healthCheckConfig;
         this.sequentialExecutor = new SequentialExecutor((uncaughtException) ->
                 LOGGER.error("{}: Uncaught exception in {}", this, this.getClass().getSimpleName(), uncaughtException));

servicetalk-loadbalancer-experimental/src/main/java/io/servicetalk/loadbalancer/DefaultLoadBalancerBuilder.java

@@ -41,7 +41,7 @@ final class DefaultLoadBalancerBuilder<ResolvedAddress, C extends LoadBalancedCo
     @Nullable
     private Executor backgroundExecutor;
     @Nullable
-    private LoadBalancerObserver loadBalancerObserver;
+    private LoadBalancerObserverFactory loadBalancerObserverFactory;
     private ConnectionPoolStrategyFactory<C> connectionPoolStrategyFactory = defaultConnectionPoolStrategyFactory();
     private OutlierDetectorConfig outlierDetectorConfig = OutlierDetectorConfig.DEFAULT_CONFIG;
 
@@ -60,7 +60,13 @@ public LoadBalancerBuilder<ResolvedAddress, C> loadBalancingPolicy(
     @Override
     public LoadBalancerBuilder<ResolvedAddress, C> loadBalancerObserver(
             @Nullable LoadBalancerObserver loadBalancerObserver) {
-        this.loadBalancerObserver = loadBalancerObserver;
+        return loadBalancerObserver(ignored -> loadBalancerObserver);
+    }
+
+    @Override
+    public LoadBalancerBuilder<ResolvedAddress, C> loadBalancerObserver(
+            @Nullable LoadBalancerObserverFactory loadBalancerObserverFactory) {
+        this.loadBalancerObserverFactory = loadBalancerObserverFactory;
         return this;
     }
 
@@ -100,8 +106,8 @@ public LoadBalancerFactory<ResolvedAddress, C> build() {
                     outlierDetectorConfig.serviceDiscoveryResubscribeInterval(),
                     outlierDetectorConfig.serviceDiscoveryResubscribeJitter());
         }
-        final LoadBalancerObserver loadBalancerObserver = this.loadBalancerObserver != null ?
-                this.loadBalancerObserver : NoopLoadBalancerObserver.instance();
+        final LoadBalancerObserverFactory loadBalancerObserverFactory = this.loadBalancerObserverFactory != null ?
+                this.loadBalancerObserverFactory : ignored -> NoopLoadBalancerObserver.instance();
         final Function<String, OutlierDetector<ResolvedAddress, C>> outlierDetectorFactory;
         if (OutlierDetectorConfig.xDSDisabled(outlierDetectorConfig)) {
             outlierDetectorFactory = (lbDescription) -> new NoopOutlierDetector<>(outlierDetectorConfig, executor);
@@ -110,15 +116,15 @@ public LoadBalancerFactory<ResolvedAddress, C> build() {
                 new XdsOutlierDetector<>(executor, outlierDetectorConfig, lbDescription);
         }
         return new DefaultLoadBalancerFactory<>(id, loadBalancingPolicy, healthCheckConfig,
-                loadBalancerObserver, outlierDetectorFactory, connectionPoolStrategyFactory);
+                loadBalancerObserverFactory, outlierDetectorFactory, connectionPoolStrategyFactory);
     }
 
     static final class DefaultLoadBalancerFactory<ResolvedAddress, C extends LoadBalancedConnection>
             implements LoadBalancerFactory<ResolvedAddress, C> {
 
         private final String id;
         private final LoadBalancingPolicy<ResolvedAddress, C> loadBalancingPolicy;
-        private final LoadBalancerObserver loadBalancerObserver;
+        private final LoadBalancerObserverFactory loadBalancerObserverFactory;
         @Nullable
         private final Function<String, OutlierDetector<ResolvedAddress, C>> outlierDetectorFactory;
         @Nullable
@@ -127,12 +133,13 @@ static final class DefaultLoadBalancerFactory<ResolvedAddress, C extends LoadBal
 
         DefaultLoadBalancerFactory(final String id, final LoadBalancingPolicy<ResolvedAddress, C> loadBalancingPolicy,
                                    @Nullable final HealthCheckConfig healthCheckConfig,
-                                   final LoadBalancerObserver loadBalancerObserver,
+                                   final LoadBalancerObserverFactory loadBalancerObserverFactory,
                                    final Function<String, OutlierDetector<ResolvedAddress, C>> outlierDetectorFactory,
                                    final ConnectionPoolStrategyFactory<C> connectionPoolStrategyFactory) {
             this.id = requireNonNull(id, "id");
             this.loadBalancingPolicy = requireNonNull(loadBalancingPolicy, "loadBalancingPolicy");
-            this.loadBalancerObserver = requireNonNull(loadBalancerObserver, "loadBalancerObserver");
+            this.loadBalancerObserverFactory = requireNonNull(loadBalancerObserverFactory,
+                    "loadBalancerObserverFactory");
             this.connectionPoolStrategyFactory = requireNonNull(
                     connectionPoolStrategyFactory, "connectionPoolStrategyFactory");
             this.healthCheckConfig = healthCheckConfig;
@@ -162,7 +169,7 @@ public LoadBalancer<C> newLoadBalancer(
             return new DefaultLoadBalancer<>(id, targetResource, eventPublisher,
                     loadBalancingPolicy.buildSelector(Collections.emptyList(), targetResource),
                     connectionPoolStrategyFactory.buildStrategy(targetResource), connectionFactory,
-                    loadBalancerObserver, healthCheckConfig, outlierDetectorFactory);
+                    loadBalancerObserverFactory, healthCheckConfig, outlierDetectorFactory);
         }
 
         @Override

servicetalk-loadbalancer-experimental/src/main/java/io/servicetalk/loadbalancer/DelegatingLoadBalancerBuilder.java

@@ -66,6 +66,13 @@ public LoadBalancerBuilder<ResolvedAddress, C> loadBalancerObserver(
         return this;
     }
 
+    @Override
+    public LoadBalancerBuilder<ResolvedAddress, C> loadBalancerObserver(
+            @Nullable LoadBalancerObserverFactory loadBalancerObserverFactory) {
+        delegate = delegate.loadBalancerObserver(loadBalancerObserverFactory);
+        return this;
+    }
+
     @Override
     public LoadBalancerBuilder<ResolvedAddress, C> outlierDetectorConfig(OutlierDetectorConfig outlierDetectorConfig) {
         delegate = delegate.outlierDetectorConfig(outlierDetectorConfig);

servicetalk-loadbalancer-experimental/src/main/java/io/servicetalk/loadbalancer/LoadBalancerBuilder.java

@@ -74,9 +74,24 @@ LoadBalancerBuilder<ResolvedAddress, C> loadBalancingPolicy(
      * Set the {@link LoadBalancerObserver} to use with this load balancer.
      * @param loadBalancerObserver the {@link LoadBalancerObserver} to use, or {@code null} to not use an observer.
      * @return {code this}
+     * @deprecated use the overload that takes a {@link LoadBalancerObserverFactory}
      */
+    @Deprecated // TODO: remove deprecated method after 0.42.45 release.
     LoadBalancerBuilder<ResolvedAddress, C> loadBalancerObserver(@Nullable LoadBalancerObserver loadBalancerObserver);
 
+    /**
+     * Set the {@link LoadBalancerObserverFactory} to use with this load balancer.
+     * @param loadBalancerObserverFactory the {@link LoadBalancerObserverFactory} to use, or {@code null} to not use an
+     *                                    observer.
+     * @return {code this}
+     */
+    // TODO: remove the default implementation after 0.42.45 release.
+    default LoadBalancerBuilder<ResolvedAddress, C> loadBalancerObserver(
+            @Nullable LoadBalancerObserverFactory loadBalancerObserverFactory) {
+        return loadBalancerObserver(loadBalancerObserverFactory == null ? null
+                : loadBalancerObserverFactory.newObserver("<unknown>"));
+    }
+
     /**
      * Set the {@link OutlierDetectorConfig} to use with this load balancer.
      * The outlier detection system works in conjunction with the load balancing policy to attempt to avoid hosts

servicetalk-loadbalancer-experimental/src/main/java/io/servicetalk/loadbalancer/LoadBalancerObserverFactory.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright © 2024 Apple Inc. and the ServiceTalk project authors
+ *
+ * Licensed 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 io.servicetalk.loadbalancer;
+
+/**
+ * Factory of {@link LoadBalancerObserver} instances.
+ */
+@FunctionalInterface
+public interface LoadBalancerObserverFactory {
+
+    /**
+     * Provide a {@link LoadBalancerObserver} to use with a newly created load balancer.
+     * @param lbDescription the description of the load balancer.
+     * @return a {@link LoadBalancerObserver} to use with a newly created load balancer.
+     */
+    LoadBalancerObserver newObserver(String lbDescription);
+}

servicetalk-loadbalancer-experimental/src/test/java/io/servicetalk/loadbalancer/DefaultLoadBalancerTest.java

@@ -190,7 +190,7 @@ TestableLoadBalancer<String, TestLoadBalancedConnection> newTestLoadBalancer(
                 LinearSearchConnectionPoolStrategy.<TestLoadBalancedConnection>factory(DEFAULT_LINEAR_SEARCH_SPACE)
                         .buildStrategy("test-service"),
                 connectionFactory,
-                NoopLoadBalancerObserver.instance(),
+                lbDescription -> NoopLoadBalancerObserver.instance(),
                 null,
                 factory);
     }