Adds slow start config implementation

Signed-off-by: anurag.ag <anuragagarwal561994@users.noreply.github.com>

Author: anuragagarwal561994

xds/src/main/java/io/grpc/xds/WeightedRoundRobinLoadBalancer.java

@@ -90,6 +90,7 @@ final class WeightedRoundRobinLoadBalancer extends MultiChildLoadBalancer {
   private static final LongCounterMetricInstrument RR_FALLBACK_COUNTER;
   private static final LongCounterMetricInstrument ENDPOINT_WEIGHT_NOT_YET_USEABLE_COUNTER;
   private static final LongCounterMetricInstrument ENDPOINT_WEIGHT_STALE_COUNTER;
+  private static final LongCounterMetricInstrument ENDPOINT_SLOW_START_COUNTER;
   private static final DoubleHistogramMetricInstrument ENDPOINT_WEIGHTS_HISTOGRAM;
   private static final Logger log = Logger.getLogger(
       WeightedRoundRobinLoadBalancer.class.getName());
@@ -133,6 +134,14 @@ final class WeightedRoundRobinLoadBalancer extends MultiChildLoadBalancer {
         Lists.newArrayList("grpc.target"),
         Lists.newArrayList("grpc.lb.locality", "grpc.lb.backend_service"),
         false);
+    ENDPOINT_SLOW_START_COUNTER = metricInstrumentRegistry.registerLongCounter(
+        "grpc.lb.wrr.endpoints_in_slow_start",
+        "EXPERIMENTAL. Number of endpoints from each scheduler update that "
+            + "are in slow start window",
+        "{endpoint}",
+        Lists.newArrayList("grpc.target"),
+        Lists.newArrayList("grpc.lb.locality", "grpc.lb.backend_service"),
+        false);
     ENDPOINT_WEIGHTS_HISTOGRAM = metricInstrumentRegistry.registerDoubleHistogram(
         "grpc.lb.wrr.endpoint_weights",
         "EXPERIMENTAL. The histogram buckets will be endpoint weight ranges.",
@@ -243,16 +252,21 @@ private SubchannelPicker createReadyPicker(Collection<ChildLbState> activeList)
   private void updateWeight(WeightedRoundRobinPicker picker) {
     Helper helper = getHelper();
     float[] newWeights = new float[picker.children.size()];
+    float[] newScales = new float[picker.children.size()];
     AtomicInteger staleEndpoints = new AtomicInteger();
     AtomicInteger notYetUsableEndpoints = new AtomicInteger();
+    AtomicInteger slowStartEndpoints = new AtomicInteger();
     for (int i = 0; i < picker.children.size(); i++) {
       double newWeight = ((WeightedChildLbState) picker.children.get(i)).getWeight(staleEndpoints,
           notYetUsableEndpoints);
+      double newScale = ((WeightedChildLbState) picker.children.get(i))
+          .getScale(slowStartEndpoints);
       helper.getMetricRecorder()
           .recordDoubleHistogram(ENDPOINT_WEIGHTS_HISTOGRAM, newWeight,
               ImmutableList.of(helper.getChannelTarget()),
               ImmutableList.of(locality, backendService));
       newWeights[i] = newWeight > 0 ? (float) newWeight : 0.0f;
+      newScales[i] = newScale > 0 ? (float) newScale : 1.0f;
     }
 
     if (staleEndpoints.get() > 0) {
@@ -267,7 +281,13 @@ private void updateWeight(WeightedRoundRobinPicker picker) {
               ImmutableList.of(helper.getChannelTarget()),
               ImmutableList.of(locality, backendService));
     }
-    boolean weightsEffective = picker.updateWeight(newWeights);
+    if (slowStartEndpoints.get() > 0) {
+      helper.getMetricRecorder()
+          .addLongCounter(ENDPOINT_SLOW_START_COUNTER, slowStartEndpoints.get(),
+              ImmutableList.of(helper.getChannelTarget()),
+              ImmutableList.of(locality, backendService));
+    }
+    boolean weightsEffective = picker.updateWeight(newWeights, newScales);
     if (!weightsEffective) {
       helper.getMetricRecorder()
           .addLongCounter(RR_FALLBACK_COUNTER, 1, ImmutableList.of(helper.getChannelTarget()),
@@ -289,6 +309,7 @@ final class WeightedChildLbState extends ChildLbState {
     private final Set<WrrSubchannel> subchannels = new HashSet<>();
     private volatile long lastUpdated;
     private volatile long nonEmptySince;
+    private volatile long readySince;
     private volatile double weight = 0;
 
     private OrcaReportListener orcaReportListener;
@@ -320,6 +341,25 @@ private double getWeight(AtomicInteger staleEndpoints, AtomicInteger notYetUsabl
       }
     }
 
+    private double getScale(AtomicInteger slowStartEndpoints) {
+      if (config == null || config.slowStartConfig == null) {
+        return 1;
+      }
+      long slowStartWindowNanos = config.slowStartConfig.slowStartWindowNanos;
+      if (slowStartWindowNanos <= 0) {
+        return 1;
+      }
+      long now = ticker.nanoTime();
+      if (now - readySince >= slowStartWindowNanos) {
+        return 1;
+      } else {
+        slowStartEndpoints.incrementAndGet();
+        double timeFactor = Math.max(now - readySince, 1.0) / slowStartWindowNanos;
+        double weightPercent = Math.pow(timeFactor, 1.0 / config.slowStartConfig.aggression);
+        return Math.max(config.slowStartConfig.minWeightPercent / 100.0, weightPercent);
+      }
+    }
+
     public void addSubchannel(WrrSubchannel wrrSubchannel) {
       subchannels.add(wrrSubchannel);
     }
@@ -439,6 +479,7 @@ public void start(SubchannelStateListener listener) {
         public void onSubchannelState(ConnectivityStateInfo newState) {
           if (newState.getState().equals(ConnectivityState.READY)) {
             owner.nonEmptySince = infTime;
+            owner.readySince = ticker.nanoTime();
           }
           listener.onSubchannelState(newState);
         }
@@ -517,8 +558,8 @@ public PickResult pickSubchannel(PickSubchannelArgs args) {
     }
 
     /** Returns {@code true} if weights are different than round_robin. */
-    private boolean updateWeight(float[] newWeights) {
-      this.scheduler = new StaticStrideScheduler(newWeights, sequence);
+    private boolean updateWeight(float[] newWeights, float[] newScales) {
+      this.scheduler = new StaticStrideScheduler(newWeights, newScales, sequence);
       return !this.scheduler.usesRoundRobin();
     }
 
@@ -604,7 +645,7 @@ static final class StaticStrideScheduler {
     private static final double K_MAX_RATIO = 10;
     private static final double K_MIN_RATIO = 0.1;
 
-    StaticStrideScheduler(float[] weights, AtomicInteger sequence) {
+    StaticStrideScheduler(float[] weights, float[] scales, AtomicInteger sequence) {
       checkArgument(weights.length >= 1, "Couldn't build scheduler: requires at least one weight");
       int numChannels = weights.length;
       int numWeightedChannels = 0;
@@ -643,12 +684,14 @@ static final class StaticStrideScheduler {
       int weightLowerBound = (int) Math.ceil(scalingFactor * unscaledMeanWeight * K_MIN_RATIO);
       short[] scaledWeights = new short[numChannels];
       for (int i = 0; i < numChannels; i++) {
+        double curScalingFactor = scalingFactor * scales[i];
+        int weight;
         if (weights[i] <= 0) {
-          scaledWeights[i] = (short) Math.round(scalingFactor * unscaledMeanWeight);
+          weight = (int) Math.round(curScalingFactor * unscaledMeanWeight);
         } else {
-          int weight = (int) Math.round(scalingFactor * Math.min(weights[i], unscaledMaxWeight));
-          scaledWeights[i] = (short) Math.max(weight, weightLowerBound);
+          weight = (int) Math.round(curScalingFactor * Math.min(weights[i], unscaledMaxWeight));
         }
+        scaledWeights[i] = (short) Math.max(weight, weightLowerBound);
       }
 
       this.scaledWeights = scaledWeights;