add range validation & merge metrics from MetricRecorder and CallMetricRecorder in per-call reporting

Author: danielzhaotongliu

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

@@ -32,6 +32,7 @@
 @ExperimentalApi("https://github.com/grpc/grpc-java/issues/6012")
 @ThreadSafe
 public final class CallMetricRecorder {
+
   private static final CallMetricRecorder NOOP = new CallMetricRecorder().disable();
 
   static final Context.Key<CallMetricRecorder> CONTEXT_KEY =
@@ -65,16 +66,16 @@ public static CallMetricRecorder getCurrent() {
   }
 
   /**
-   * Records a call metric measurement for utilization.
-   * If RPC has already finished, this method is no-op.
+   * Records a call metric measurement for utilization in the range [0, 1]. Values outside the valid
+   * range are rejected. 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.23.0
    */
   public CallMetricRecorder recordUtilizationMetric(String name, double value) {
-    if (disabled) {
+    if (disabled || !MetricRecorderHelper.isUtilizationValid(value)) {
       return this;
     }
     if (utilizationMetrics.get() == null) {
@@ -87,15 +88,15 @@ public CallMetricRecorder recordUtilizationMetric(String name, double value) {
   }
 
   /**
-   * Records a call metric measurement for request cost.
-   * If RPC has already finished, this method is no-op.
+   * Records a call metric measurement for request cost. 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.47.0
-   * @deprecated use {@link #recordRequestCostMetric} instead.
-   *     This method will be removed in the future.
+   * @deprecated use {@link #recordRequestCostMetric} instead. This method will be removed in the
+   * future.
    */
   @Deprecated
   @InlineMe(replacement = "this.recordRequestCostMetric(name, value)")
@@ -104,8 +105,8 @@ public CallMetricRecorder recordCallMetric(String name, double value) {
   }
 
   /**
-   * Records a call metric measurement for request cost.
-   * If RPC has already finished, this method is no-op.
+   * Records a call metric measurement for request cost. If RPC has already finished, this method is
+   * no-op.
    *
    * <p>A latter record will overwrite its former name-sakes.
    *
@@ -126,57 +127,56 @@ public CallMetricRecorder recordRequestCostMetric(String name, double value) {
   }
 
   /**
-   * Records a call metric measurement for CPU utilization.
-   * If RPC has already finished, this method is no-op.
+   * Records a call metric measurement for CPU utilization in the range [0, 1]. Values outside the
+   * valid range are rejected. 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.47.0
    */
   public CallMetricRecorder recordCpuUtilizationMetric(double value) {
-    if (disabled) {
+    if (disabled || !MetricRecorderHelper.isUtilizationValid(value)) {
       return this;
     }
     cpuUtilizationMetric = value;
     return this;
   }
 
   /**
-   * Records a call metric measurement for memory utilization.
-   * If RPC has already finished, this method is no-op.
+   * Records a call metric measurement for memory utilization in the range [0, 1]. Values outside
+   * the valid range are rejected. 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.47.0
    */
   public CallMetricRecorder recordMemoryUtilizationMetric(double value) {
-    if (disabled) {
+    if (disabled || !MetricRecorderHelper.isUtilizationValid(value)) {
       return this;
     }
     memoryUtilizationMetric = value;
     return this;
   }
 
   /**
-   * Records a call metric measurement for qps.
-   * If RPC has already finished, this method is no-op.
+   * Records a call metric measurement for qps in the range [0, inf). Values outside the valid range
+   * are rejected. 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) {
+    if (disabled || !MetricRecorderHelper.isQpsValid(value)) {
       return this;
     }
     qps = value;
     return this;
   }
 
-
   /**
    * Returns all request cost metric values. No more metric values will be recorded after this
    * method is called. Calling this method multiple times returns the same collection of metric

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

@@ -22,11 +22,12 @@
 import java.util.concurrent.ConcurrentHashMap;
 
 /**
- * Implements the service/APIs for Out-of-Band metrics reporting, only for utilization metrics.
- * A user should use the public set-APIs to update the server machine's utilization metrics data.
+ * Implements the service/APIs for Out-of-Band metrics reporting, only for utilization metrics. A
+ * user should use the public set-APIs to update the server machine's utilization metrics data.
  */
 @ExperimentalApi("https://github.com/grpc/grpc-java/issues/9006")
 public final class MetricRecorder {
+
   private volatile ConcurrentHashMap<String, Double> metricsData = new ConcurrentHashMap<>();
   private volatile double cpuUtilization;
   private volatile double memoryUtilization;
@@ -36,17 +37,22 @@ public static MetricRecorder newInstance() {
     return new MetricRecorder();
   }
 
-  private MetricRecorder() {}
+  private MetricRecorder() {
+  }
 
   /**
-   * Update the metrics value corresponding to the specified key.
+   * Update the metrics value in the range [0, 1] corresponding to the specified key. Values outside
+   * the valid range are rejected.
    */
   public void putUtilizationMetric(String key, double value) {
+    if (!MetricRecorderHelper.isUtilizationValid(value)) {
+      return;
+    }
     metricsData.put(key, value);
   }
 
   /**
-   * Replace the whole metrics data using the specified map.
+   * Replace the whole metrics data using the specified map. No range validation.
    */
   public void setAllUtilizationMetrics(Map<String, Double> metrics) {
     metricsData = new ConcurrentHashMap<>(metrics);
@@ -60,9 +66,13 @@ public void removeUtilizationMetric(String key) {
   }
 
   /**
-   * Update the CPU utilization metrics data.
+   * Update the CPU utilization metrics data in the range [0, 1]. Values outside the valid range are
+   * rejected.
    */
   public void setCpuUtilizationMetric(double value) {
+    if (!MetricRecorderHelper.isUtilizationValid(value)) {
+      return;
+    }
     cpuUtilization = value;
   }
 
@@ -74,9 +84,13 @@ public void clearCpuUtilizationMetric() {
   }
 
   /**
-   * Update the memory utilization metrics data.
+   * Update the memory utilization metrics data in the range [0, 1]. Values outside the valid range
+   * are rejected.
    */
   public void setMemoryUtilizationMetric(double value) {
+    if (!MetricRecorderHelper.isUtilizationValid(value)) {
+      return;
+    }
     memoryUtilization = value;
   }
 
@@ -88,9 +102,13 @@ public void clearMemoryUtilizationMetric() {
   }
 
   /**
-   * Update the QPS metrics data.
+   * Update the QPS metrics data in the range [0, inf). Values outside the valid range are
+   * rejected.
    */
   public void setQps(double value) {
+    if (!MetricRecorderHelper.isQpsValid(value)) {
+      return;
+    }
     qps = value;
   }
 
@@ -102,7 +120,7 @@ public void clearQps() {
   }
 
   MetricReport getMetricReport() {
-    return new MetricReport(cpuUtilization, memoryUtilization, qps,
-        Collections.emptyMap(), Collections.unmodifiableMap(metricsData));
+    return new MetricReport(cpuUtilization, memoryUtilization, qps, Collections.emptyMap(),
+        Collections.unmodifiableMap(metricsData));
   }
 }

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

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2023 The gRPC Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.grpc.services;
+
+/**
+ * Utility helper class to check whether values for {@link CallMetricRecorder} and
+ * {@link MetricRecorder} are inside the valid range.
+ */
+final class MetricRecorderHelper {
+
+  /**
+   * Return true if the utilization value is in the range [0, 1] and false otherwise.
+   */
+  static boolean isUtilizationValid(double utilization) {
+    return utilization >= 0.0 && utilization <= 1.0;
+  }
+
+  /**
+   * Return true if the qps value is in the range [0, inf) and false otherwise.
+   */
+  static boolean isQpsValid(double qps) {
+    return qps >= 0.0;
+  }
+
+  // Prevent instantiation.
+  private MetricRecorderHelper() {
+  }
+}

xds/src/main/java/io/grpc/xds/orca/OrcaMetricReportingServerInterceptor.java

@@ -31,13 +31,15 @@
 import io.grpc.protobuf.ProtoUtils;
 import io.grpc.services.CallMetricRecorder;
 import io.grpc.services.InternalCallMetricRecorder;
+import io.grpc.services.InternalMetricRecorder;
+import io.grpc.services.MetricRecorder;
 import io.grpc.services.MetricReport;
 
 /**
  * A {@link ServerInterceptor} that intercepts a {@link ServerCall} by running server-side RPC
  * handling under a {@link Context} that records custom per-request metrics provided by server
- * applications and sends to client side along with the response in the format of Open Request
- * Cost Aggregation (ORCA).
+ * applications and sends to client side along with the response in the format of Open Request Cost
+ * Aggregation (ORCA).
  *
  * @since 1.23.0
  */
@@ -53,6 +55,8 @@ public final class OrcaMetricReportingServerInterceptor implements ServerInterce
           "endpoint-load-metrics-bin",
           ProtoUtils.metadataMarshaller(OrcaLoadReport.getDefaultInstance()));
 
+  private static MetricRecorder metricRecorder;
+
   @VisibleForTesting
   OrcaMetricReportingServerInterceptor() {
   }
@@ -61,6 +65,10 @@ public static OrcaMetricReportingServerInterceptor getInstance() {
     return INSTANCE;
   }
 
+  public static void setMetricRecorder(MetricRecorder mr) {
+    metricRecorder = mr;
+  }
+
   @Override
   public <ReqT, RespT> Listener<ReqT> interceptCall(
       ServerCall<ReqT, RespT> call, Metadata headers, ServerCallHandler<ReqT, RespT> next) {
@@ -77,6 +85,10 @@ public <ReqT, RespT> Listener<ReqT> interceptCall(
           public void close(Status status, Metadata trailers) {
             OrcaLoadReport report = fromInternalReport(
                 InternalCallMetricRecorder.finalizeAndDump2(finalCallMetricRecorder));
+            if (metricRecorder != null) {
+              report = mergeMetrics(report,
+                  fromInternalReport(InternalMetricRecorder.getMetricReport(metricRecorder)));
+            }
             if (!report.equals(OrcaLoadReport.getDefaultInstance())) {
               trailers.put(ORCA_ENDPOINT_LOAD_METRICS_KEY, report);
             }
@@ -94,8 +106,40 @@ private static OrcaLoadReport fromInternalReport(MetricReport internalReport) {
     return OrcaLoadReport.newBuilder()
         .setCpuUtilization(internalReport.getCpuUtilization())
         .setMemUtilization(internalReport.getMemoryUtilization())
+        .setRpsFractional(internalReport.getQps())
         .putAllUtilization(internalReport.getUtilizationMetrics())
         .putAllRequestCost(internalReport.getRequestCostMetrics())
         .build();
   }
+
+  /**
+   * Return a merged {@link OrcaLoadReport} where the metrics from {@link CallMetricRecorder} takes
+   * a higher precedence compared to {@link MetricRecorder}.
+   */
+  private static OrcaLoadReport mergeMetrics(OrcaLoadReport callMetricRecorderReport,
+      OrcaLoadReport metricRecorderReport) {
+    // Merge metrics from the MetricRecorder first since metrics from the CallMetricRecorder takes a
+    // higher precedence.
+    OrcaLoadReport.Builder builder = metricRecorderReport.toBuilder()
+        .putAllUtilization(callMetricRecorderReport.getUtilizationMap())
+        .putAllRequestCost(callMetricRecorderReport.getRequestCostMap());
+    // Overwrite only if the values from CallMetricRecorder are set
+    double cpu = callMetricRecorderReport.getCpuUtilization();
+    if (isReportValueSet(cpu)) {
+      builder.setCpuUtilization(cpu);
+    }
+    double mem = callMetricRecorderReport.getMemUtilization();
+    if (isReportValueSet(mem)) {
+      builder.setMemUtilization(mem);
+    }
+    double rps = callMetricRecorderReport.getRpsFractional();
+    if (isReportValueSet(rps)) {
+      builder.setRpsFractional(rps);
+    }
+    return builder.build();
+  }
+
+  private static boolean isReportValueSet(double value) {
+    return value != 0;
+  }
 }