services, xds, orca: support EPS in client-side WRR (#10177)

core/src/main/java/io/grpc/internal/JsonUtil.java

@@ -121,6 +121,33 @@ public static Double getNumberAsDouble(Map<String, ?> obj, String key) {
         String.format("value '%s' for key '%s' in '%s' is not a number", value, key, obj));
   }
 
+  /**
+   * Gets a number from an object for the given key.  If the key is not present, this returns null.
+   * If the value does not represent a float, throws an exception.
+   */
+  @Nullable
+  public static Float getNumberAsFloat(Map<String, ?> obj, String key) {
+    assert key != null;
+    if (!obj.containsKey(key)) {
+      return null;
+    }
+    Object value = obj.get(key);
+    if (value instanceof Float) {
+      return (Float) value;
+    }
+    if (value instanceof String) {
+      try {
+        return Float.parseFloat((String) value);
+      } catch (NumberFormatException e) {
+        throw new IllegalArgumentException(
+            String.format("string value '%s' for key '%s' cannot be parsed as a float", value,
+                key));
+      }
+    }
+    throw new IllegalArgumentException(
+        String.format("value %s for key '%s' is not a float", value, key));
+  }
+
   /**
    * Gets a number from an object for the given key, casted to an integer.  If the key is not
    * present, this returns null.  If the value does not represent an integer, throws an exception.

core/src/test/java/io/grpc/internal/JsonUtilTest.java

@@ -38,16 +38,19 @@ public void getNumber() {
     map.put("key_string_nan", "NaN");
     map.put("key_number_5.5", 5.5D);
     map.put("key_string_six", "six");
+    map.put("key_number_7", 7F);
     map.put("key_string_infinity", "Infinity");
     map.put("key_string_minus_infinity", "-Infinity");
     map.put("key_string_exponent", "2.998e8");
     map.put("key_string_minus_zero", "-0");
+    map.put("key_string_boolean", true);
 
     assertThat(JsonUtil.getNumberAsDouble(map, "key_number_1")).isEqualTo(1D);
     assertThat(JsonUtil.getNumberAsInteger(map, "key_number_1")).isEqualTo(1);
     assertThat(JsonUtil.getNumberAsLong(map, "key_number_1")).isEqualTo(1L);
 
     assertThat(JsonUtil.getNumberAsDouble(map, "key_string_2.0")).isEqualTo(2D);
+    assertThat(JsonUtil.getNumberAsFloat(map, "key_string_2.0")).isEqualTo(2F);
     try {
       JsonUtil.getNumberAsInteger(map, "key_string_2.0");
       fail("expecting to throw but did not");
@@ -66,8 +69,10 @@ public void getNumber() {
     assertThat(JsonUtil.getNumberAsDouble(map, "key_string_3")).isEqualTo(3D);
     assertThat(JsonUtil.getNumberAsInteger(map, "key_string_3")).isEqualTo(3);
     assertThat(JsonUtil.getNumberAsLong(map, "key_string_3")).isEqualTo(3L);
+    assertThat(JsonUtil.getNumberAsFloat(map, "key_string_3")).isEqualTo(3F);
 
     assertThat(JsonUtil.getNumberAsDouble(map, "key_string_nan")).isNaN();
+    assertThat(JsonUtil.getNumberAsFloat(map, "key_string_nan")).isNaN();
     try {
       JsonUtil.getNumberAsInteger(map, "key_string_nan");
       fail("expecting to throw but did not");
@@ -118,18 +123,41 @@ public void getNumber() {
       assertThat(e).hasMessageThat().isEqualTo(
           "value 'six' for key 'key_string_six' is not a long integer");
     }
+    try {
+      JsonUtil.getNumberAsFloat(map, "key_string_six");
+      fail("expecting to throw but did not");
+    } catch (RuntimeException e) {
+      assertThat(e).hasMessageThat().isEqualTo(
+          "string value 'six' for key 'key_string_six' cannot be parsed as a float");
+    }
+
+    assertThat(JsonUtil.getNumberAsFloat(map, "key_number_7")).isEqualTo(7F);
 
     assertThat(JsonUtil.getNumberAsDouble(map, "key_string_infinity")).isPositiveInfinity();
     assertThat(JsonUtil.getNumberAsDouble(map, "key_string_minus_infinity")).isNegativeInfinity();
     assertThat(JsonUtil.getNumberAsDouble(map, "key_string_exponent")).isEqualTo(2.998e8D);
 
+    assertThat(JsonUtil.getNumberAsFloat(map, "key_string_infinity")).isPositiveInfinity();
+    assertThat(JsonUtil.getNumberAsFloat(map, "key_string_minus_infinity")).isNegativeInfinity();
+    assertThat(JsonUtil.getNumberAsFloat(map, "key_string_exponent")).isEqualTo(2.998e8F);
+
     assertThat(JsonUtil.getNumberAsDouble(map, "key_string_minus_zero")).isZero();
     assertThat(JsonUtil.getNumberAsInteger(map, "key_string_minus_zero")).isEqualTo(0);
     assertThat(JsonUtil.getNumberAsLong(map, "key_string_minus_zero")).isEqualTo(0L);
+    assertThat(JsonUtil.getNumberAsFloat(map, "key_string_minus_zero")).isZero();
 
     assertThat(JsonUtil.getNumberAsDouble(map, "key_nonexistent")).isNull();
     assertThat(JsonUtil.getNumberAsInteger(map, "key_nonexistent")).isNull();
     assertThat(JsonUtil.getNumberAsLong(map, "key_nonexistent")).isNull();
+    assertThat(JsonUtil.getNumberAsFloat(map, "key_nonexistent")).isNull();
+
+    try {
+      JsonUtil.getNumberAsFloat(map, "key_string_boolean");
+      fail("expecting to throw but did not");
+    } catch (RuntimeException e) {
+      assertThat(e).hasMessageThat().isEqualTo(
+          "value true for key 'key_string_boolean' is not a float");
+    }
   }
 
   @Test

repositories.bzl

@@ -86,10 +86,10 @@ def grpc_java_repositories():
     if not native.existing_rule("com_github_cncf_xds"):
         http_archive(
             name = "com_github_cncf_xds",
-            strip_prefix = "xds-06c439db220b89134a8a49bad41994560d6537c6",
-            sha256 = "41ea212940ab44bf7f8a8b4169cfbc612ed2166dafabc0a56a8820ef665fc6a4",
+            strip_prefix = "xds-32f1caf87195bf3390061c29f18987e51ca56a88",
+            sha256 = "fcd0b50c013452fda9c5e28c131c287b655ebb361271a76ad3bffc08b3ecd82e",
             urls = [
-                "https://github.com/cncf/xds/archive/06c439db220b89134a8a49bad41994560d6537c6.tar.gz",
+                "https://github.com/cncf/xds/archive/32f1caf87195bf3390061c29f18987e51ca56a88.tar.gz",
             ],
         )
     if not native.existing_rule("com_github_grpc_grpc"):

services/src/main/java/io/grpc/services/CallMetricRecorder.java

@@ -44,6 +44,7 @@ public final class CallMetricRecorder {
   private double cpuUtilizationMetric = 0;
   private double memoryUtilizationMetric = 0;
   private double qps = 0;
+  private double eps = 0;
   private volatile boolean disabled;
 
   /**
@@ -160,22 +161,37 @@ public CallMetricRecorder recordMemoryUtilizationMetric(double value) {
   }
 
   /**
-   * Records a call metric measurement for qps in the range [0, inf). Values outside the valid range
-   * are ignored. If RPC has already finished, this method is no-op.
+   * Records a call metric measurement for queries per second (qps) in the range [0, inf). Values
+   * outside the valid range are ignored. If RPC has already finished, this method is no-op.
    *
    * <p>A latter record will overwrite its former name-sakes.
    *
    * @return this recorder object
    * @since 1.54.0
    */
   public CallMetricRecorder recordQpsMetric(double value) {
-    if (disabled || !MetricRecorderHelper.isQpsValid(value)) {
+    if (disabled || !MetricRecorderHelper.isRateValid(value)) {
       return this;
     }
     qps = value;
     return this;
   }
 
+  /**
+   * Records a call metric measurement for errors per second (eps) in the range [0, inf). Values
+   * outside the valid range are ignored. If RPC has already finished, this method is no-op.
+   *
+   * <p>A latter record will overwrite its former name-sakes.
+   *
+   * @return this recorder object
+   */
+  public CallMetricRecorder recordEpsMetric(double value) {
+    if (disabled || !MetricRecorderHelper.isRateValid(value)) {
+      return this;
+    }
+    eps = value;
+    return this;
+  }
 
   /**
    * Returns all request cost metric values. No more metric values will be recorded after this
@@ -205,7 +221,7 @@ MetricReport finalizeAndDump2() {
     if (savedUtilizationMetrics == null) {
       savedUtilizationMetrics = Collections.emptyMap();
     }
-    return new MetricReport(cpuUtilizationMetric, memoryUtilizationMetric, qps,
+    return new MetricReport(cpuUtilizationMetric, memoryUtilizationMetric, qps, eps,
         Collections.unmodifiableMap(savedRequestCostMetrics),
         Collections.unmodifiableMap(savedUtilizationMetrics)
     );

services/src/main/java/io/grpc/services/InternalCallMetricRecorder.java

@@ -46,8 +46,9 @@ public static MetricReport finalizeAndDump2(CallMetricRecorder recorder) {
   }
 
   public static MetricReport createMetricReport(double cpuUtilization, double memoryUtilization,
-      double qps, Map<String, Double> requestCostMetrics, Map<String, Double> utilizationMetrics) {
-    return new MetricReport(cpuUtilization, memoryUtilization, qps, requestCostMetrics,
+      double qps, double eps, Map<String, Double> requestCostMetrics,
+      Map<String, Double> utilizationMetrics) {
+    return new MetricReport(cpuUtilization, memoryUtilization, qps, eps, requestCostMetrics,
         utilizationMetrics);
   }
 }

services/src/main/java/io/grpc/services/MetricRecorder.java

@@ -31,6 +31,7 @@ public final class MetricRecorder {
   private volatile double cpuUtilization;
   private volatile double memoryUtilization;
   private volatile double qps;
+  private volatile double eps;
 
   public static MetricRecorder newInstance() {
     return new MetricRecorder();
@@ -103,7 +104,7 @@ public void clearMemoryUtilizationMetric() {
    * Update the QPS metrics data in the range [0, inf). Values outside the valid range are ignored.
    */
   public void setQpsMetric(double value) {
-    if (!MetricRecorderHelper.isQpsValid(value)) {
+    if (!MetricRecorderHelper.isRateValid(value)) {
       return;
     }
     qps = value;
@@ -116,8 +117,25 @@ public void clearQpsMetric() {
     qps = 0;
   }
 
+  /**
+   * Update the EPS metrics data in the range [0, inf). Values outside the valid range are ignored.
+   */
+  public void setEpsMetric(double value) {
+    if (!MetricRecorderHelper.isRateValid(value)) {
+      return;
+    }
+    this.eps = value;
+  }
+
+  /**
+   * Clear the EPS metrics data.
+   */
+  public void clearEpsMetric() {
+    eps = 0;
+  }
+
   MetricReport getMetricReport() {
-    return new MetricReport(cpuUtilization, memoryUtilization, qps,
+    return new MetricReport(cpuUtilization, memoryUtilization, qps, eps,
         Collections.emptyMap(), Collections.unmodifiableMap(metricsData));
   }
 }

services/src/main/java/io/grpc/services/MetricRecorderHelper.java

@@ -39,10 +39,10 @@ static boolean isCpuUtilizationValid(double utilization) {
   }
 
   /**
-   * Return true if the qps value is in the range [0, inf) and false otherwise.
+   * Return true if a rate value (such as qps or eps) is in the range [0, inf) and false otherwise.
    */
-  static boolean isQpsValid(double qps) {
-    return qps >= 0.0;
+  static boolean isRateValid(double rate) {
+    return rate >= 0.0;
   }
 
   // Prevent instantiation.

services/src/main/java/io/grpc/services/MetricReport.java

@@ -31,15 +31,17 @@ public final class MetricReport {
   private double cpuUtilization;
   private double memoryUtilization;
   private double qps;
+  private double eps;
   private Map<String, Double> requestCostMetrics;
   private Map<String, Double> utilizationMetrics;
 
-  MetricReport(double cpuUtilization, double memoryUtilization, double qps,
+  MetricReport(double cpuUtilization, double memoryUtilization, double qps, double eps,
                Map<String, Double> requestCostMetrics,
                Map<String, Double> utilizationMetrics) {
     this.cpuUtilization = cpuUtilization;
     this.memoryUtilization = memoryUtilization;
     this.qps = qps;
+    this.eps = eps;
     this.requestCostMetrics = checkNotNull(requestCostMetrics, "requestCostMetrics");
     this.utilizationMetrics = checkNotNull(utilizationMetrics, "utilizationMetrics");
   }
@@ -64,6 +66,10 @@ public double getQps() {
     return qps;
   }
 
+  public double getEps() {
+    return eps;
+  }
+
   @Override
   public String toString() {
     return MoreObjects.toStringHelper(this)
@@ -72,6 +78,7 @@ public String toString() {
         .add("requestCost", requestCostMetrics)
         .add("utilization", utilizationMetrics)
         .add("qps", qps)
+        .add("eps", eps)
         .toString();
   }
 }

services/src/test/java/io/grpc/services/CallMetricRecorderTest.java

@@ -47,6 +47,7 @@ public void dumpDumpsAllSavedMetricValues() {
     recorder.recordCpuUtilizationMetric(0.1928);
     recorder.recordMemoryUtilizationMetric(0.474);
     recorder.recordQpsMetric(2522.54);
+    recorder.recordEpsMetric(1.618);
 
     MetricReport dump = recorder.finalizeAndDump2();
     Truth.assertThat(dump.getUtilizationMetrics())
@@ -56,12 +57,16 @@ public void dumpDumpsAllSavedMetricValues() {
     Truth.assertThat(dump.getCpuUtilization()).isEqualTo(0.1928);
     Truth.assertThat(dump.getMemoryUtilization()).isEqualTo(0.474);
     Truth.assertThat(dump.getQps()).isEqualTo(2522.54);
+    Truth.assertThat(dump.getEps()).isEqualTo(1.618);
+    Truth.assertThat(dump.toString()).contains("eps=1.618");
   }
 
   @Test
   public void noMetricsRecordedAfterSnapshot() {
     Map<String, Double> initDump = recorder.finalizeAndDump();
     recorder.recordUtilizationMetric("cost", 0.154353423);
+    recorder.recordQpsMetric(3.14159);
+    recorder.recordEpsMetric(1.618);
     assertThat(recorder.finalizeAndDump()).isEqualTo(initDump);
   }
 
@@ -84,12 +89,14 @@ public void noMetricsRecordedIfUtilizationAndQpsAreLessThanLowerBound() {
     recorder.recordCpuUtilizationMetric(-0.001);
     recorder.recordMemoryUtilizationMetric(-0.001);
     recorder.recordQpsMetric(-0.001);
+    recorder.recordEpsMetric(-0.001);
     recorder.recordUtilizationMetric("util1", -0.001);
 
     MetricReport dump = recorder.finalizeAndDump2();
     Truth.assertThat(dump.getCpuUtilization()).isEqualTo(0);
     Truth.assertThat(dump.getMemoryUtilization()).isEqualTo(0);
     Truth.assertThat(dump.getQps()).isEqualTo(0);
+    Truth.assertThat(dump.getEps()).isEqualTo(0);
     Truth.assertThat(dump.getUtilizationMetrics()).isEmpty();
     Truth.assertThat(dump.getRequestCostMetrics()).isEmpty();
   }
@@ -108,6 +115,8 @@ public void lastValueWinForMetricsWithSameName() {
     recorder.recordUtilizationMetric("util1", 0.843233);
     recorder.recordQpsMetric(1928.3);
     recorder.recordQpsMetric(100.8);
+    recorder.recordEpsMetric(3.14159);
+    recorder.recordEpsMetric(1.618);
 
     MetricReport dump = recorder.finalizeAndDump2();
     Truth.assertThat(dump.getRequestCostMetrics())
@@ -117,6 +126,7 @@ public void lastValueWinForMetricsWithSameName() {
         .containsExactly("util1", 0.843233);
     Truth.assertThat(dump.getCpuUtilization()).isEqualTo(0);
     Truth.assertThat(dump.getQps()).isEqualTo(100.8);
+    Truth.assertThat(dump.getEps()).isEqualTo(1.618);
   }
 
   @Test

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

@@ -89,6 +89,8 @@ class LoadBalancerConfigFactory {
   static final String PICK_FIRST_FIELD_NAME = "pick_first";
   static final String SHUFFLE_ADDRESS_LIST_FIELD_NAME = "shuffleAddressList";
 
+  static final String ERROR_UTILIZATION_PENALTY = "errorUtilizationPenalty";
+
   /**
    * Factory method for creating a new {link LoadBalancerConfigConverter} for a given xDS {@link
    * Cluster}.
@@ -135,7 +137,8 @@ class LoadBalancerConfigFactory {
                                                         String weightExpirationPeriod,
                                                         String oobReportingPeriod,
                                                         Boolean enableOobLoadReport,
-                                                        String weightUpdatePeriod) {
+                                                        String weightUpdatePeriod,
+                                                        Float errorUtilizationPenalty) {
     ImmutableMap.Builder<String, Object> configBuilder = ImmutableMap.builder();
     if (blackoutPeriod != null) {
       configBuilder.put(BLACK_OUT_PERIOD, blackoutPeriod);
@@ -152,6 +155,9 @@ class LoadBalancerConfigFactory {
     if (weightUpdatePeriod != null) {
       configBuilder.put(WEIGHT_UPDATE_PERIOD, weightUpdatePeriod);
     }
+    if (errorUtilizationPenalty != null) {
+      configBuilder.put(ERROR_UTILIZATION_PENALTY, errorUtilizationPenalty);
+    }
     return ImmutableMap.of(WeightedRoundRobinLoadBalancerProvider.SCHEME,
         configBuilder.buildOrThrow());
   }
@@ -291,7 +297,8 @@ static class LoadBalancingPolicyConverter {
                 ? Durations.toString(wrr.getWeightExpirationPeriod()) : null,
             wrr.hasOobReportingPeriod() ? Durations.toString(wrr.getOobReportingPeriod()) : null,
             wrr.hasEnableOobLoadReport() ? wrr.getEnableOobLoadReport().getValue() : null,
-            wrr.hasWeightUpdatePeriod() ? Durations.toString(wrr.getWeightUpdatePeriod()) : null);
+            wrr.hasWeightUpdatePeriod() ? Durations.toString(wrr.getWeightUpdatePeriod()) : null,
+            wrr.hasErrorUtilizationPenalty() ? wrr.getErrorUtilizationPenalty().getValue() : null);
       } catch (IllegalArgumentException ex) {
         throw new ResourceInvalidException("Invalid duration in weighted round robin config: "
             + ex.getMessage());