loadbalancer-experimental: DefaultLoadBalancer logs settings on start…

…up (#3000)

Motivation:

As DefaultLoadBalancer becomes the norm it can be helpful to track its
activation via log messages.

Modifications:

- Emit a log message on startup that includes the lb description and
  important settings.
- Fix a bug where the cancellation penalty was inadvertently used for
  the error penalty.
- Cancel the update stream from the OutlierDetector when the LB closes.

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

@@ -20,6 +20,7 @@
 import io.servicetalk.client.api.NoActiveHostException;
 import io.servicetalk.client.api.NoAvailableHostException;
 import io.servicetalk.client.api.ServiceDiscovererEvent;
+import io.servicetalk.concurrent.Cancellable;
 import io.servicetalk.concurrent.CompletableSource;
 import io.servicetalk.concurrent.PublisherSource.Processor;
 import io.servicetalk.concurrent.PublisherSource.Subscriber;
@@ -109,6 +110,7 @@ final class DefaultLoadBalancer<ResolvedAddress, C extends LoadBalancedConnectio
     private final HealthCheckConfig healthCheckConfig;
     private final HostPriorityStrategy priorityStrategy;
     private final OutlierDetector<ResolvedAddress, C> outlierDetector;
+    private final Cancellable outlierDetectorStatusChangeStream;
     private final LoadBalancerObserver loadBalancerObserver;
     private final ListenableAsyncCloseable asyncCloseable;
 
@@ -167,8 +169,11 @@ final class DefaultLoadBalancer<ResolvedAddress, C extends LoadBalancedConnectio
         subscribeToEvents(false);
 
         // When we get a health-status event we should update the host set.
-        this.outlierDetector.healthStatusChanged().forEach((ignored) ->
+        this.outlierDetectorStatusChangeStream = this.outlierDetector.healthStatusChanged().forEach((ignored) ->
             sequentialExecutor.execute(() -> sequentialUpdateUsedHosts(usedHosts)));
+
+        LOGGER.info("{}: starting load balancer. Load balancing policy: {}, outlier detection: {}", this,
+                loadBalancingPolicy, outlierDetector);
     }
 
     private void subscribeToEvents(boolean resubscribe) {
@@ -194,6 +199,7 @@ private Completable doClose(final boolean graceful) {
                 if (!isClosed) {
                     discoveryCancellable.cancel();
                     eventStreamProcessor.onComplete();
+                    outlierDetectorStatusChangeStream.cancel();
                     outlierDetector.cancel();
                 }
                 isClosed = true;

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

@@ -57,7 +57,7 @@ private final class BasicHealthIndicator extends DefaultRequestTracker
 
         BasicHealthIndicator() {
             super(outlierDetectorConfig.ewmaHalfLife().toNanos(), outlierDetectorConfig.ewmaCancellationPenalty(),
-                    outlierDetectorConfig.ewmaCancellationPenalty(), outlierDetectorConfig.concurrentRequestPenalty());
+                    outlierDetectorConfig.ewmaErrorPenalty(), outlierDetectorConfig.concurrentRequestPenalty());
         }
 
         @Override
@@ -95,4 +95,15 @@ public void setHost(Host<ResolvedAddress, C> host) {
             // noop
         }
     }
+
+    @Override
+    public String toString() {
+        return "NoopOutlierDetector{" +
+                "ewmaHalfLife=" + outlierDetectorConfig.ewmaHalfLife() +
+                ", ewmaCancellationPenalty=" + outlierDetectorConfig.ewmaCancellationPenalty() +
+                ", ewmaErrorPenalty=" + outlierDetectorConfig.ewmaErrorPenalty() +
+                ", concurrentRequestPenalty=" + outlierDetectorConfig.concurrentRequestPenalty() +
+                ", executor=" + executor +
+                '}';
+    }
 }

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

@@ -68,6 +68,7 @@ final class XdsOutlierDetector<ResolvedAddress, C extends LoadBalancedConnection
 
     private static final Logger LOGGER = LoggerFactory.getLogger(XdsOutlierDetector.class);
 
+    private final OutlierDetectorConfig outlierDetectorConfig;
     private final SequentialExecutor sequentialExecutor;
     private final Executor executor;
     private final String lbDescription;
@@ -80,21 +81,18 @@ final class XdsOutlierDetector<ResolvedAddress, C extends LoadBalancedConnection
     // reads and writes are protected by `sequentialExecutor`.
     private int ejectedHostCount;
 
-    XdsOutlierDetector(final Executor executor, final OutlierDetectorConfig config, final String lbDescription,
-                       SequentialExecutor.ExceptionHandler exceptionHandler) {
+    XdsOutlierDetector(final Executor executor, final OutlierDetectorConfig outlierDetectorConfig,
+                       final String lbDescription, SequentialExecutor.ExceptionHandler exceptionHandler) {
         this.sequentialExecutor = new SequentialExecutor(exceptionHandler);
+        this.outlierDetectorConfig = requireNonNull(outlierDetectorConfig, "outlierDetectorConfig");
         this.executor = requireNonNull(executor, "executor");
         this.lbDescription = requireNonNull(lbDescription, "lbDescription");
-        this.kernel = new Kernel(config);
+        this.kernel = new Kernel(outlierDetectorConfig);
     }
 
     XdsOutlierDetector(final Executor executor, final OutlierDetectorConfig config, final String lbDescription) {
-        SequentialExecutor.ExceptionHandler exceptionHandler = (uncaughtException) ->
-                LOGGER.error("{}: Uncaught exception in {}", this, getClass().getSimpleName(), uncaughtException);
-        this.sequentialExecutor = new SequentialExecutor(exceptionHandler);
-        this.executor = requireNonNull(executor, "executor");
-        this.lbDescription = requireNonNull(lbDescription, "lbDescription");
-        this.kernel = new Kernel(config);
+        this(executor, config, lbDescription, (uncaughtException) -> LOGGER.error("{}: Uncaught exception in {}",
+                lbDescription, XdsOutlierDetector.class.getSimpleName(), uncaughtException));
     }
 
     @Override
@@ -130,6 +128,15 @@ int ejectedHostCount() {
         return ejectedHostCount;
     }
 
+    @Override
+    public String toString() {
+        return "XdsOutlierDetector{" +
+                "lbDescription=" + lbDescription +
+                ", outlierDetectorConfig=" + outlierDetectorConfig +
+                ", executor=" + executor +
+                '}';
+    }
+
     private final class XdsHealthIndicatorImpl extends XdsHealthIndicator<ResolvedAddress, C> {
 
         // Protected by `sequentialExecutor`.