Merge remote-tracking branch 'origin/main' into bl_anderson/configurable_connection_selection

Author: bryce-anderson

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

@@ -84,15 +84,15 @@ private enum State {
     @Nullable
     private final HealthIndicator healthIndicator;
     private final ConnectionPoolStrategy<C> connectionPoolStrategy;
-    private final LoadBalancerObserver.HostObserver<Addr> hostObserver;
+    private final LoadBalancerObserver.HostObserver hostObserver;
     private final ConnectionFactory<Addr, ? extends C> connectionFactory;
     private final ListenableAsyncCloseable closeable;
     private volatile ConnState connState = new ConnState(emptyList(), State.ACTIVE, 0, null);
 
     DefaultHost(final String lbDescription, final Addr address,
                 final ConnectionPoolStrategy<C> connectionPoolStrategy,
                 final ConnectionFactory<Addr, ? extends C> connectionFactory,
-                final HostObserver<Addr> hostObserver, final @Nullable HealthCheckConfig healthCheckConfig,
+                final HostObserver hostObserver, final @Nullable HealthCheckConfig healthCheckConfig,
                 final @Nullable HealthIndicator healthIndicator) {
         this.lbDescription = requireNonNull(lbDescription, "lbDescription");
         this.address = requireNonNull(address, "address");
@@ -104,7 +104,6 @@ private enum State {
         this.healthCheckConfig = healthCheckConfig;
         this.hostObserver = requireNonNull(hostObserver, "hostObserver");
         this.closeable = toAsyncCloseable(this::doClose);
-        hostObserver.onHostCreated(address);
     }
 
     @Override
@@ -124,7 +123,7 @@ public boolean markActiveIfNotClosed() {
             return oldConnState;
         });
         if (oldState.state == State.EXPIRED) {
-            hostObserver.onExpiredHostRevived(address, oldState.connections.size());
+            hostObserver.onExpiredHostRevived(oldState.connections.size());
         }
         return oldState.state != State.CLOSED;
     }
@@ -161,7 +160,7 @@ public boolean markExpired() {
             Object nextState = oldState.connections.isEmpty() ? State.CLOSED : State.EXPIRED;
             if (connStateUpdater.compareAndSet(this, oldState, oldState.toExpired())) {
                 cancelIfHealthCheck(oldState);
-                hostObserver.onHostMarkedExpired(address, oldState.connections.size());
+                hostObserver.onHostMarkedExpired(oldState.connections.size());
                 if (nextState == State.CLOSED) {
                     // Trigger the callback to remove the host from usedHosts array.
                     this.closeAsync().subscribe();
@@ -261,7 +260,7 @@ private void markHealthy(final HealthCheck originalHealthCheckState) {
         }
         // Only if the previous state was a healthcheck should we notify the observer.
         if (oldState.isUnhealthy()) {
-            hostObserver.onHostRevived(address);
+            hostObserver.onHostRevived();
         }
     }
 
@@ -292,7 +291,7 @@ private void onConnectionError(Throwable cause) {
                                     "{} time(s) in a row. Error counting threshold reached, marking this host as " +
                                     "UNHEALTHY for the selection algorithm and triggering background health-checking.",
                             lbDescription, address, healthCheckConfig.failedThreshold, cause);
-                    hostObserver.onHostMarkedUnhealthy(address, cause);
+                    hostObserver.onHostMarkedUnhealthy(cause);
                     nextState.healthCheck.schedule(cause);
                 }
                 break;
@@ -336,7 +335,7 @@ private boolean addConnection(final C connection, final @Nullable HealthCheck cu
                         cancelIfHealthCheck(previous);
                     }
                     // If we transitioned from unhealth to healthy we need to let the observer know.
-                    hostObserver.onHostRevived(address);
+                    hostObserver.onHostRevived();
                 }
                 break;
             }
@@ -373,7 +372,7 @@ private boolean addConnection(final C connection, final @Nullable HealthCheck cu
                             // in the next iteration.
                             && connStateUpdater.compareAndSet(this, currentConnState, nextState.toClosed())) {
                         closeAsync().subscribe();
-                        hostObserver.onExpiredHostRemoved(address, nextState.connections.size());
+                        hostObserver.onExpiredHostRemoved(nextState.connections.size());
                         break;
                     }
                 } else {
@@ -426,9 +425,9 @@ private Completable doClose(final boolean graceful) {
             LOGGER.debug("{}: closing {} connection(s) {}gracefully to the closed address: {}.",
                     lbDescription, oldState.connections.size(), graceful ? "" : "un", address);
             if (oldState.state == State.ACTIVE) {
-                hostObserver.onActiveHostRemoved(address, oldState.connections.size());
+                hostObserver.onActiveHostRemoved(oldState.connections.size());
             } else if (oldState.state == State.EXPIRED) {
-                hostObserver.onExpiredHostRemoved(address, oldState.connections.size());
+                hostObserver.onExpiredHostRemoved(oldState.connections.size());
             }
             final List<C> connections = oldState.connections;
             return (connections.isEmpty() ? completed() :

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

@@ -46,8 +46,8 @@
 import java.util.Map.Entry;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.atomic.AtomicLongFieldUpdater;
+import java.util.function.Function;
 import java.util.function.Predicate;
-import java.util.function.Supplier;
 import javax.annotation.Nullable;
 
 import static io.servicetalk.client.api.LoadBalancerReadyEvent.LOAD_BALANCER_NOT_READY_EVENT;
@@ -133,7 +133,7 @@ final class DefaultLoadBalancer<ResolvedAddress, C extends LoadBalancedConnectio
             final ConnectionFactory<ResolvedAddress, ? extends C> connectionFactory,
             final LoadBalancerObserver<ResolvedAddress> loadBalancerObserver,
             @Nullable final HealthCheckConfig healthCheckConfig,
-            @Nullable final Supplier<HealthChecker<ResolvedAddress>> healthCheckerFactory) {
+            @Nullable final Function<String, HealthChecker<ResolvedAddress>> healthCheckerFactory) {
         this.targetResource = requireNonNull(targetResourceName);
         this.lbDescription = makeDescription(id, targetResource);
         this.hostSelector = requireNonNull(hostSelector, "hostSelector");
@@ -150,7 +150,7 @@ final class DefaultLoadBalancer<ResolvedAddress, C extends LoadBalancedConnectio
         // Maintain a Subscriber so signals are always delivered to replay and new Subscribers get the latest signal.
         eventStream.ignoreElements().subscribe();
         subscribeToEvents(false);
-        this.healthChecker = healthCheckerFactory == null ? null : healthCheckerFactory.get();
+        this.healthChecker = healthCheckerFactory == null ? null : healthCheckerFactory.apply(lbDescription);
     }
 
     private void subscribeToEvents(boolean resubscribe) {
@@ -383,10 +383,12 @@ private void sequentialOnNext(Collection<? extends ServiceDiscovererEvent<Resolv
         }
 
         private Host<ResolvedAddress, C> createHost(ResolvedAddress addr) {
+            final LoadBalancerObserver.HostObserver hostObserver = loadBalancerObserver.hostObserver(addr);
             // All hosts will share the healthcheck config of the parent RR loadbalancer.
-            final HealthIndicator indicator = healthChecker == null ? null : healthChecker.newHealthIndicator(addr);
+            final HealthIndicator indicator = healthChecker == null ?
+                    null : healthChecker.newHealthIndicator(addr, hostObserver);
             final Host<ResolvedAddress, C> host = new DefaultHost<>(lbDescription, addr, connectionPoolStrategy,
-                    connectionFactory, loadBalancerObserver.hostObserver(), healthCheckConfig, indicator);
+                    connectionFactory, loadBalancerObserver.hostObserver(addr), healthCheckConfig, indicator);
             host.onClose().afterFinally(() ->
                     sequentialExecutor.execute(() -> {
                         final List<Host<ResolvedAddress, C>> currentHosts = usedHosts;

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

@@ -26,7 +26,7 @@
 import java.time.Duration;
 import java.util.Collection;
 import java.util.Collections;
-import java.util.function.Supplier;
+import java.util.function.Function;
 import javax.annotation.Nullable;
 
 import static io.servicetalk.loadbalancer.HealthCheckConfig.DEFAULT_HEALTH_CHECK_FAILED_CONNECTIONS_THRESHOLD;
@@ -135,13 +135,13 @@ public LoadBalancerFactory<ResolvedAddress, C> build() {
         }
         final LoadBalancerObserver<ResolvedAddress> loadBalancerObserver = this.loadBalancerObserver != null ?
                 this.loadBalancerObserver : NoopLoadBalancerObserver.instance();
-        Supplier<HealthChecker<ResolvedAddress>> healthCheckerSupplier;
+        Function<String, HealthChecker<ResolvedAddress>> healthCheckerSupplier;
         if (healthCheckerFactory == null) {
             healthCheckerSupplier = null;
         } else {
             final Executor executor = getExecutor();
-            healthCheckerSupplier = () -> healthCheckerFactory.newHealthChecker(executor,
-                    loadBalancerObserver.hostObserver());
+            healthCheckerSupplier = (String lbDescrption) ->
+                    healthCheckerFactory.newHealthChecker(executor, lbDescrption);
         }
 
         return new DefaultLoadBalancerFactory<>(id, loadBalancingPolicy, healthCheckConfig,
@@ -155,15 +155,15 @@ private static final class DefaultLoadBalancerFactory<ResolvedAddress, C extends
         private final LoadBalancingPolicy<ResolvedAddress, C> loadBalancingPolicy;
         private final LoadBalancerObserver<ResolvedAddress> loadBalancerObserver;
         @Nullable
-        private final Supplier<HealthChecker<ResolvedAddress>> healthCheckerFactory;
+        private final Function<String, HealthChecker<ResolvedAddress>> healthCheckerFactory;
         @Nullable
         private final HealthCheckConfig healthCheckConfig;
         private final ConnectionPoolStrategyFactory<C> connectionPoolStrategyFactory;
 
         DefaultLoadBalancerFactory(final String id, final LoadBalancingPolicy<ResolvedAddress, C> loadBalancingPolicy,
                                    final HealthCheckConfig healthCheckConfig,
                                    final LoadBalancerObserver<ResolvedAddress> loadBalancerObserver,
-                                   final Supplier<HealthChecker<ResolvedAddress>> healthCheckerFactory,
+                                   final Function<String, HealthChecker<ResolvedAddress>> healthCheckerFactory,
                                    final ConnectionPoolStrategyFactory<C> connectionPoolStrategyFactory) {
             this.id = requireNonNull(id, "id");
             this.loadBalancingPolicy = requireNonNull(loadBalancingPolicy, "loadBalancingPolicy");

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

@@ -20,6 +20,7 @@
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.function.IntBinaryOperator;
 
+import static io.servicetalk.utils.internal.NumberUtils.ensurePositive;
 import static java.lang.Integer.MAX_VALUE;
 import static java.lang.Integer.MIN_VALUE;
 import static java.lang.Math.ceil;
@@ -66,9 +67,7 @@ abstract class DefaultRequestTracker implements RequestTracker, ScoreSupplier {
     }
 
     DefaultRequestTracker(final long halfLifeNanos, final long cancelPenalty, final long errorPenalty) {
-        if (halfLifeNanos <= 0) {
-            throw new IllegalArgumentException("halfLifeNanos: " + halfLifeNanos + " (expected >0)");
-        }
+        ensurePositive(halfLifeNanos, "halfLifeNanos");
         this.invTau = Math.pow((halfLifeNanos / log(2)), -1);
         this.cancelPenalty = cancelPenalty;
         this.errorPenalty = errorPenalty;

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

@@ -30,5 +30,5 @@ interface HealthChecker<ResolvedAddress> extends Cancellable {
      * @param address the resolved address of the destination.
      * @return new {@link HealthIndicator}.
      */
-    HealthIndicator newHealthIndicator(ResolvedAddress address);
+    HealthIndicator newHealthIndicator(ResolvedAddress address, LoadBalancerObserver.HostObserver hostObserver);
 }

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

@@ -16,7 +16,6 @@
 package io.servicetalk.loadbalancer;
 
 import io.servicetalk.concurrent.api.Executor;
-import io.servicetalk.loadbalancer.LoadBalancerObserver.HostObserver;
 
 /**
  * A factory of {@link HealthChecker} instances. The factory will be used by load balancer
@@ -27,9 +26,8 @@ interface HealthCheckerFactory<ResolvedAddress> {
     /**
      * Create a new {@link HealthChecker}.
      * @param executor the {@link Executor} to use for scheduling tasks and obtaining the current time.
-     * @param hostObserver a {@link HostObserver} to notify of
-     *                     relevant host events.
+     * @param lbDescription a description of the load balancer for logging purposes.
      * @return a new {@link HealthChecker}.
      */
-    HealthChecker newHealthChecker(Executor executor, HostObserver<ResolvedAddress> hostObserver);
+    HealthChecker<ResolvedAddress> newHealthChecker(Executor executor, String lbDescription);
 }

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

@@ -31,7 +31,7 @@ interface LoadBalancerObserver<ResolvedAddress> {
      * Get a {@link HostObserver}.
      * @return a {@link HostObserver}.
      */
-    HostObserver<ResolvedAddress> hostObserver();
+    HostObserver hostObserver(ResolvedAddress resolvedAddress);
 
     /**
      * Callback for when connection selection fails due to no hosts being available.
@@ -51,55 +51,42 @@ void onServiceDiscoveryEvent(Collection<? extends ServiceDiscovererEvent<Resolve
 
     /**
      * An observer for {@link Host} events.
-     * @param <ResolvedAddress> the type of the resolved address.
      */
-    interface HostObserver<ResolvedAddress> {
+    interface HostObserver {
 
         /**
          * Callback for when an active host is marked expired.
-         * @param address the resolved address.
          * @param connectionCount the number of active connections for the host.
          */
-        void onHostMarkedExpired(ResolvedAddress address, int connectionCount);
+        void onHostMarkedExpired(int connectionCount);
 
         /**
          * Callback for when a host is removed by service discovery.
-         * @param address the resolved address.
          * @param connectionCount the number of open connections when the host was removed.
          */
-        void onActiveHostRemoved(ResolvedAddress address, int connectionCount);
+        void onActiveHostRemoved(int connectionCount);
 
         /**
          * Callback for when an expired host is returned to an active state.
-         * @param address the resolved address.
          * @param connectionCount the number of active connections when the host was revived.
          */
-        void onExpiredHostRevived(ResolvedAddress address, int connectionCount);
+        void onExpiredHostRevived(int connectionCount);
 
         /**
          * Callback for when an expired host is removed.
-         * @param address the resolved address.
          * @param connectionCount the number of open connections when the host was removed.
          */
-        void onExpiredHostRemoved(ResolvedAddress address, int connectionCount);
-
-        /**
-         * Callback for when a host is created.
-         * @param address the resolved address.
-         */
-        void onHostCreated(ResolvedAddress address);
+        void onExpiredHostRemoved(int connectionCount);
 
         /**
          * Callback for when a {@link Host} transitions from healthy to unhealthy.
-         * @param address the resolved address.
          * @param cause the most recent cause of the transition.
          */
-        void onHostMarkedUnhealthy(ResolvedAddress address, @Nullable Throwable cause);
+        void onHostMarkedUnhealthy(@Nullable Throwable cause);
 
         /**
          * Callback for when a {@link Host} transitions from unhealthy to healthy.
-         * @param address the resolved address.
          */
-        void onHostRevived(ResolvedAddress address);
+        void onHostRevived();
     }
 }

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

@@ -29,8 +29,8 @@ private NoopLoadBalancerObserver() {
     }
 
     @Override
-    public HostObserver<ResolvedAddress> hostObserver() {
-        return (HostObserver<ResolvedAddress>) NoopHostObserver.INSTANCE;
+    public HostObserver hostObserver(ResolvedAddress resolvedAddress) {
+        return NoopHostObserver.INSTANCE;
     }
 
     @Override
@@ -49,46 +49,40 @@ public void onServiceDiscoveryEvent(Collection<? extends ServiceDiscovererEvent<
         // noop
     }
 
-    private static final class NoopHostObserver<ResolvedAddress> implements
-            LoadBalancerObserver.HostObserver<ResolvedAddress> {
+    private static final class NoopHostObserver implements LoadBalancerObserver.HostObserver {
 
-        private static final HostObserver<Object> INSTANCE = new NoopHostObserver<>();
+        private static final HostObserver INSTANCE = new NoopHostObserver();
 
         private NoopHostObserver() {
         }
 
         @Override
-        public void onHostMarkedExpired(ResolvedAddress resolvedAddress, int connectionCount) {
+        public void onHostMarkedExpired(int connectionCount) {
             // noop
         }
 
         @Override
-        public void onExpiredHostRemoved(ResolvedAddress resolvedAddress, int connectionCount) {
+        public void onExpiredHostRemoved(int connectionCount) {
             // noop
         }
 
         @Override
-        public void onExpiredHostRevived(ResolvedAddress resolvedAddress, int connectionCount) {
+        public void onExpiredHostRevived(int connectionCount) {
             // noop
         }
 
         @Override
-        public void onActiveHostRemoved(ResolvedAddress resolvedAddress, int connectionCount) {
+        public void onActiveHostRemoved(int connectionCount) {
             // noop
         }
 
         @Override
-        public void onHostCreated(ResolvedAddress resolvedAddress) {
+        public void onHostMarkedUnhealthy(Throwable cause) {
             // noop
         }
 
         @Override
-        public void onHostMarkedUnhealthy(ResolvedAddress address, Throwable cause) {
-            // noop
-        }
-
-        @Override
-        public void onHostRevived(ResolvedAddress address) {
+        public void onHostRevived() {
             // noop
         }
     }