loadbalancer-experimental: Thread through the EWMA half life in XdsHealthTracker

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/OutlierDetectorConfig.java

@@ -31,6 +31,7 @@
  */
 final class OutlierDetectorConfig {
 
+    private final Duration ewmaHalfLife;
     private final int consecutive5xx;
     private final Duration interval;
     private final Duration baseEjectionTime;
@@ -55,7 +56,8 @@ final class OutlierDetectorConfig {
     private final Duration maxEjectionTimeJitter;
     private final boolean successfulActiveHealthCheckUnejectHost;
 
-    OutlierDetectorConfig(final int consecutive5xx, final Duration interval, final Duration baseEjectionTime,
+    OutlierDetectorConfig(final Duration ewmaHalfLife,
+                          final int consecutive5xx, final Duration interval, final Duration baseEjectionTime,
                           final int maxEjectionPercentage, final int enforcingConsecutive5xx,
                           final int enforcingSuccessRate, final int successRateMinimumHosts,
                           final int successRateRequestVolume, final int successRateStdevFactor,
@@ -66,6 +68,7 @@ final class OutlierDetectorConfig {
                           final int enforcingFailurePercentageLocalOrigin, final int failurePercentageMinimumHosts,
                           final int failurePercentageRequestVolume, final Duration maxEjectionTime,
                           final Duration maxEjectionTimeJitter, final boolean successfulActiveHealthCheckUnejectHost) {
+        this.ewmaHalfLife = requireNonNull(ewmaHalfLife, "ewmaHalfLife");
         this.consecutive5xx = consecutive5xx;
         this.interval = requireNonNull(interval, "interval");
         this.baseEjectionTime = requireNonNull(baseEjectionTime, "baseEjectionTime");
@@ -91,6 +94,16 @@ final class OutlierDetectorConfig {
         this.successfulActiveHealthCheckUnejectHost = successfulActiveHealthCheckUnejectHost;
     }
 
+    /**
+     * The Exponentially Weighted Moving Average (EWMA) half-life.
+     * In the context of an exponentially weighted moving average, the half-life means the time during which
+     * historical data has the same weight as a new sample.
+     * @return the Exponentially Weighted Moving Average (EWMA) half-life.
+     */
+    public Duration ewmaHalfLife() {
+        return ewmaHalfLife;
+    }
+
     /**
      * The number of consecutive failures before the attempt to suspect the host.
      * @return the number of consecutive failures before the attempt to suspect the host.
@@ -299,6 +312,7 @@ public boolean successfulActiveHealthCheckUnejectHost() {
      * A builder for {@link OutlierDetectorConfig} instances.
      */
     public static class Builder {
+        private Duration ewmaHalfLife = Duration.ofSeconds(10);
         private int consecutive5xx = 5;
 
         private Duration interval = Duration.ofSeconds(10);
@@ -346,7 +360,8 @@ public static class Builder {
         private boolean successfulActiveHealthCheckUnejectHost = true;
 
         OutlierDetectorConfig build() {
-            return new OutlierDetectorConfig(consecutive5xx, interval, baseEjectionTime,
+            return new OutlierDetectorConfig(ewmaHalfLife, consecutive5xx,
+                    interval, baseEjectionTime,
                     maxEjectionPercentage, enforcingConsecutive5xx,
                     enforcingSuccessRate, successRateMinimumHosts,
                     successRateRequestVolume, successRateStdevFactor,
@@ -360,6 +375,21 @@ OutlierDetectorConfig build() {
                     successfulActiveHealthCheckUnejectHost);
         }
 
+        /**
+         * Set the Exponentially Weighted Moving Average (EWMA) half-life.
+         * In the context of an exponentially weighted moving average, the half-life means the time during which
+         * historical data has the same weight as a new sample.
+         * Defaults to 10 seconds.
+         * @param ewmaHalfLife the half-life for latency data.
+         * @return {@code this}
+         */
+        public Builder ewmaHalfLife(final Duration ewmaHalfLife) {
+            requireNonNull(ewmaHalfLife, "ewmaHalfLife");
+            ensureNonNegative(ewmaHalfLife.toNanos(), "ewmaHalfLife");
+            this.ewmaHalfLife = ewmaHalfLife;
+            return this;
+        }
+
         /**
          * Set the threshold for consecutive failures before a host is ejected.
          * Defaults to 5.

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

@@ -23,6 +23,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashSet;
@@ -79,7 +80,7 @@ final class XdsHealthChecker<ResolvedAddress> implements HealthChecker<ResolvedA
 
     @Override
     public HealthIndicator newHealthIndicator(ResolvedAddress address, HostObserver hostObserver) {
-        XdsHealthIndicator result = new XdsHealthIndicatorImpl(address, hostObserver);
+        XdsHealthIndicator result = new XdsHealthIndicatorImpl(address, kernel.config.ewmaHalfLife(), hostObserver);
         sequentialExecutor.execute(() -> indicators.add(result));
         indicatorCount.incrementAndGet();
         return result;
@@ -100,8 +101,8 @@ public void cancel() {
 
     private final class XdsHealthIndicatorImpl extends XdsHealthIndicator<ResolvedAddress> {
 
-        XdsHealthIndicatorImpl(final ResolvedAddress address, HostObserver hostObserver) {
-            super(sequentialExecutor, executor, address, lbDescription, hostObserver);
+        XdsHealthIndicatorImpl(final ResolvedAddress address, Duration ewmaHalfLife, HostObserver hostObserver) {
+            super(sequentialExecutor, executor, ewmaHalfLife, address, lbDescription, hostObserver);
         }
 
         @Override

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

@@ -21,6 +21,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.time.Duration;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -57,8 +58,9 @@ abstract class XdsHealthIndicator<ResolvedAddress> extends DefaultRequestTracker
     private volatile Long evictedUntilNanos;
 
     XdsHealthIndicator(final SequentialExecutor sequentialExecutor, final Executor executor,
-                       final ResolvedAddress address, final String lbDescription, final HostObserver hostObserver) {
-        super(1);
+                       final Duration ewmaHalfLife, final ResolvedAddress address, String lbDescription,
+                       final HostObserver hostObserver) {
+        super(requireNonNull(ewmaHalfLife, "ewmaHalfLife").toNanos());
         this.sequentialExecutor = requireNonNull(sequentialExecutor, "sequentialExecutor");
         this.executor = requireNonNull(executor, "executor");
         assert executor instanceof NormalizedTimeSourceExecutor;

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

@@ -212,7 +212,7 @@ private class TestIndicator extends XdsHealthIndicator<String> {
         boolean mayEjectHost = true;
 
         TestIndicator(final OutlierDetectorConfig config) {
-            super(sequentialExecutor, new NormalizedTimeSourceExecutor(testExecutor), "address",
+            super(sequentialExecutor, new NormalizedTimeSourceExecutor(testExecutor), Duration.ofSeconds(10), "address",
                     "description", NoopLoadBalancerObserver.<String>instance().hostObserver("address"));
             this.config = config;
         }