xds: ORCA to LRS propagation changes

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

@@ -141,15 +141,17 @@ public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
             result.maxConcurrentRequests(),
             result.upstreamTlsContext(),
             result.filterMetadata(),
-            result.outlierDetection());
+            result.outlierDetection(),
+            result.backendMetricPropagation());
       } else {
         instance = DiscoveryMechanism.forLogicalDns(
             leafName,
             result.dnsHostName(),
             result.lrsServerInfo(),
             result.maxConcurrentRequests(),
             result.upstreamTlsContext(),
-            result.filterMetadata());
+            result.filterMetadata(),
+            result.backendMetricPropagation());
       }
       instances.add(instance);
     }

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

@@ -46,6 +46,7 @@
 import io.grpc.xds.EnvoyServerProtoData.UpstreamTlsContext;
 import io.grpc.xds.ThreadSafeRandom.ThreadSafeRandomImpl;
 import io.grpc.xds.XdsNameResolverProvider.CallCounterProvider;
+import io.grpc.xds.client.BackendMetricPropagation;
 import io.grpc.xds.client.Bootstrapper.ServerInfo;
 import io.grpc.xds.client.LoadStatsManager2.ClusterDropStats;
 import io.grpc.xds.client.LoadStatsManager2.ClusterLocalityStats;
@@ -148,6 +149,7 @@ public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
     childLbHelper.updateMaxConcurrentRequests(config.maxConcurrentRequests);
     childLbHelper.updateSslContextProviderSupplier(config.tlsContext);
     childLbHelper.updateFilterMetadata(config.filterMetadata);
+    childLbHelper.updateBackendMetricPropagation(config.backendMetricPropagation);
 
     childSwitchLb.handleResolvedAddresses(
         resolvedAddresses.toBuilder()
@@ -208,6 +210,8 @@ private final class ClusterImplLbHelper extends ForwardingLoadBalancerHelper {
     private Map<String, Struct> filterMetadata = ImmutableMap.of();
     @Nullable
     private final ServerInfo lrsServerInfo;
+    @Nullable
+    private BackendMetricPropagation backendMetricPropagation;
 
     private ClusterImplLbHelper(AtomicLong inFlights, @Nullable ServerInfo lrsServerInfo) {
       this.inFlights = checkNotNull(inFlights, "inFlights");
@@ -320,7 +324,7 @@ private ClusterLocality createClusterLocalityFromAttributes(Attributes addressAt
           (lrsServerInfo == null)
               ? null
               : xdsClient.addClusterLocalityStats(lrsServerInfo, cluster,
-                  edsServiceName, locality);
+                  edsServiceName, locality, backendMetricPropagation);
 
       return new ClusterLocality(localityStats, localityName);
     }
@@ -370,6 +374,11 @@ private void updateFilterMetadata(Map<String, Struct> filterMetadata) {
       this.filterMetadata = ImmutableMap.copyOf(filterMetadata);
     }
 
+    private void updateBackendMetricPropagation(
+        @Nullable BackendMetricPropagation backendMetricPropagation) {
+      this.backendMetricPropagation = backendMetricPropagation;
+    }
+
     private class RequestLimitingSubchannelPicker extends SubchannelPicker {
       private final SubchannelPicker delegate;
       private final List<DropOverload> dropPolicies;
@@ -505,11 +514,18 @@ private OrcaPerRpcListener(ClusterLocalityStats stats) {
     }
 
     /**
-     * Copies {@link MetricReport#getNamedMetrics()} to {@link ClusterLocalityStats} such that it is
-     * included in the snapshot for the LRS report sent to the LRS server.
+     * Copies ORCA metrics from {@link MetricReport} to {@link ClusterLocalityStats}
+     * such that they are included in the snapshot for the LRS report sent to the LRS server.
+     * This includes both top-level metrics (CPU, memory, application utilization) and named
+     * metrics, filtered according to the backend metric propagation configuration.
      */
     @Override
     public void onLoadReport(MetricReport report) {
+      stats.recordTopLevelMetrics(
+          report.getCpuUtilization(),
+          report.getMemoryUtilization(),
+          report.getApplicationUtilization());
+
       stats.recordBackendLoadMetricStats(report.getNamedMetrics());
     }
   }

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

@@ -31,6 +31,7 @@
 import io.grpc.Status;
 import io.grpc.xds.Endpoints.DropOverload;
 import io.grpc.xds.EnvoyServerProtoData.UpstreamTlsContext;
+import io.grpc.xds.client.BackendMetricPropagation;
 import io.grpc.xds.client.Bootstrapper.ServerInfo;
 import java.util.ArrayList;
 import java.util.Collections;
@@ -98,11 +99,14 @@ static final class ClusterImplConfig {
     // Provides the direct child policy and its config.
     final Object childConfig;
     final Map<String, Struct> filterMetadata;
+    @Nullable
+    final BackendMetricPropagation backendMetricPropagation;
 
     ClusterImplConfig(String cluster, @Nullable String edsServiceName,
         @Nullable ServerInfo lrsServerInfo, @Nullable Long maxConcurrentRequests,
         List<DropOverload> dropCategories, Object childConfig,
-        @Nullable UpstreamTlsContext tlsContext, Map<String, Struct> filterMetadata) {
+        @Nullable UpstreamTlsContext tlsContext, Map<String, Struct> filterMetadata,
+        @Nullable BackendMetricPropagation backendMetricPropagation) {
       this.cluster = checkNotNull(cluster, "cluster");
       this.edsServiceName = edsServiceName;
       this.lrsServerInfo = lrsServerInfo;
@@ -112,6 +116,7 @@ static final class ClusterImplConfig {
       this.dropCategories = Collections.unmodifiableList(
           new ArrayList<>(checkNotNull(dropCategories, "dropCategories")));
       this.childConfig = checkNotNull(childConfig, "childConfig");
+      this.backendMetricPropagation = backendMetricPropagation;
     }
 
     @Override

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

@@ -55,6 +55,7 @@
 import io.grpc.xds.PriorityLoadBalancerProvider.PriorityLbConfig;
 import io.grpc.xds.PriorityLoadBalancerProvider.PriorityLbConfig.PriorityChildConfig;
 import io.grpc.xds.XdsEndpointResource.EdsUpdate;
+import io.grpc.xds.client.BackendMetricPropagation;
 import io.grpc.xds.client.Bootstrapper.ServerInfo;
 import io.grpc.xds.client.Locality;
 import io.grpc.xds.client.XdsClient;
@@ -191,11 +192,12 @@ public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
         if (instance.type == DiscoveryMechanism.Type.EDS) {
           state = new EdsClusterState(instance.cluster, instance.edsServiceName,
               instance.lrsServerInfo, instance.maxConcurrentRequests, instance.tlsContext,
-              instance.filterMetadata, instance.outlierDetection);
+              instance.filterMetadata, instance.outlierDetection,
+              instance.backendMetricPropagation);
         } else {  // logical DNS
           state = new LogicalDnsClusterState(instance.cluster, instance.dnsHostName,
               instance.lrsServerInfo, instance.maxConcurrentRequests, instance.tlsContext,
-              instance.filterMetadata);
+              instance.filterMetadata, instance.backendMetricPropagation);
         }
         clusterStates.put(instance.cluster, state);
         state.start();
@@ -334,6 +336,8 @@ private abstract class ClusterState {
       protected final Map<String, Struct> filterMetadata;
       @Nullable
       protected final OutlierDetection outlierDetection;
+      @Nullable
+      protected final BackendMetricPropagation backendMetricPropagation;
       // Resolution status, may contain most recent error encountered.
       protected Status status = Status.OK;
       // True if has received resolution result.
@@ -346,13 +350,15 @@ private abstract class ClusterState {
 
       private ClusterState(String name, @Nullable ServerInfo lrsServerInfo,
           @Nullable Long maxConcurrentRequests, @Nullable UpstreamTlsContext tlsContext,
-          Map<String, Struct> filterMetadata, @Nullable OutlierDetection outlierDetection) {
+          Map<String, Struct> filterMetadata, @Nullable OutlierDetection outlierDetection,
+          @Nullable BackendMetricPropagation backendMetricPropagation) {
         this.name = name;
         this.lrsServerInfo = lrsServerInfo;
         this.maxConcurrentRequests = maxConcurrentRequests;
         this.tlsContext = tlsContext;
         this.filterMetadata = ImmutableMap.copyOf(filterMetadata);
         this.outlierDetection = outlierDetection;
+        this.backendMetricPropagation = backendMetricPropagation;
       }
 
       abstract void start();
@@ -371,9 +377,10 @@ private final class EdsClusterState extends ClusterState implements ResourceWatc
       private EdsClusterState(String name, @Nullable String edsServiceName,
           @Nullable ServerInfo lrsServerInfo, @Nullable Long maxConcurrentRequests,
           @Nullable UpstreamTlsContext tlsContext, Map<String, Struct> filterMetadata,
-          @Nullable OutlierDetection outlierDetection) {
+          @Nullable OutlierDetection outlierDetection,
+          @Nullable BackendMetricPropagation backendMetricPropagation) {
         super(name, lrsServerInfo, maxConcurrentRequests, tlsContext, filterMetadata,
-            outlierDetection);
+            outlierDetection, backendMetricPropagation);
         this.edsServiceName = edsServiceName;
       }
 
@@ -470,8 +477,8 @@ public void run() {
             Map<String, PriorityChildConfig> priorityChildConfigs =
                 generateEdsBasedPriorityChildConfigs(
                     name, edsServiceName, lrsServerInfo, maxConcurrentRequests, tlsContext,
-                    filterMetadata, outlierDetection, endpointLbConfig, lbRegistry,
-                    prioritizedLocalityWeights, dropOverloads);
+                    filterMetadata, backendMetricPropagation, outlierDetection,
+                    endpointLbConfig, lbRegistry, prioritizedLocalityWeights, dropOverloads);
             status = Status.OK;
             resolved = true;
             result = new ClusterResolutionResult(addresses, priorityChildConfigs,
@@ -585,8 +592,10 @@ private final class LogicalDnsClusterState extends ClusterState {
 
       private LogicalDnsClusterState(String name, String dnsHostName,
           @Nullable ServerInfo lrsServerInfo, @Nullable Long maxConcurrentRequests,
-          @Nullable UpstreamTlsContext tlsContext, Map<String, Struct> filterMetadata) {
-        super(name, lrsServerInfo, maxConcurrentRequests, tlsContext, filterMetadata, null);
+          @Nullable UpstreamTlsContext tlsContext, Map<String, Struct> filterMetadata,
+          @Nullable BackendMetricPropagation backendMetricPropagation) {
+        super(name, lrsServerInfo, maxConcurrentRequests, tlsContext,
+            filterMetadata, null, backendMetricPropagation);
         this.dnsHostName = checkNotNull(dnsHostName, "dnsHostName");
         nameResolverFactory =
             checkNotNull(helper.getNameResolverRegistry().asFactory(), "nameResolverFactory");
@@ -688,7 +697,7 @@ public Status onResult2(final ResolutionResult resolutionResult) {
             }
             PriorityChildConfig priorityChildConfig = generateDnsBasedPriorityChildConfig(
                     name, lrsServerInfo, maxConcurrentRequests, tlsContext, filterMetadata,
-                    lbRegistry, Collections.<DropOverload>emptyList());
+                    backendMetricPropagation, lbRegistry, Collections.<DropOverload>emptyList());
             status = Status.OK;
             resolved = true;
             result = new ClusterResolutionResult(addresses, priorityName, priorityChildConfig);
@@ -772,13 +781,14 @@ private static class ClusterResolutionResult {
   private static PriorityChildConfig generateDnsBasedPriorityChildConfig(
       String cluster, @Nullable ServerInfo lrsServerInfo, @Nullable Long maxConcurrentRequests,
       @Nullable UpstreamTlsContext tlsContext, Map<String, Struct> filterMetadata,
+      @Nullable BackendMetricPropagation backendMetricPropagation,
       LoadBalancerRegistry lbRegistry, List<DropOverload> dropOverloads) {
     // Override endpoint-level LB policy with pick_first for logical DNS cluster.
     Object endpointLbConfig = GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(
         lbRegistry.getProvider("pick_first"), null);
     ClusterImplConfig clusterImplConfig =
         new ClusterImplConfig(cluster, null, lrsServerInfo, maxConcurrentRequests,
-            dropOverloads, endpointLbConfig, tlsContext, filterMetadata);
+            dropOverloads, endpointLbConfig, tlsContext, filterMetadata, backendMetricPropagation);
     LoadBalancerProvider clusterImplLbProvider =
         lbRegistry.getProvider(XdsLbPolicies.CLUSTER_IMPL_POLICY_NAME);
     Object clusterImplPolicy = GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(
@@ -796,14 +806,16 @@ private static Map<String, PriorityChildConfig> generateEdsBasedPriorityChildCon
       String cluster, @Nullable String edsServiceName, @Nullable ServerInfo lrsServerInfo,
       @Nullable Long maxConcurrentRequests, @Nullable UpstreamTlsContext tlsContext,
       Map<String, Struct> filterMetadata,
+      @Nullable BackendMetricPropagation backendMetricPropagation,
       @Nullable OutlierDetection outlierDetection, Object endpointLbConfig,
       LoadBalancerRegistry lbRegistry, Map<String,
       Map<Locality, Integer>> prioritizedLocalityWeights, List<DropOverload> dropOverloads) {
     Map<String, PriorityChildConfig> configs = new HashMap<>();
     for (String priority : prioritizedLocalityWeights.keySet()) {
       ClusterImplConfig clusterImplConfig =
           new ClusterImplConfig(cluster, edsServiceName, lrsServerInfo, maxConcurrentRequests,
-              dropOverloads, endpointLbConfig, tlsContext, filterMetadata);
+              dropOverloads, endpointLbConfig, tlsContext,
+              filterMetadata, backendMetricPropagation);
       LoadBalancerProvider clusterImplLbProvider =
           lbRegistry.getProvider(XdsLbPolicies.CLUSTER_IMPL_POLICY_NAME);
       Object priorityChildPolicy = GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(

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

@@ -29,6 +29,7 @@
 import io.grpc.Status;
 import io.grpc.xds.EnvoyServerProtoData.OutlierDetection;
 import io.grpc.xds.EnvoyServerProtoData.UpstreamTlsContext;
+import io.grpc.xds.client.BackendMetricPropagation;
 import io.grpc.xds.client.Bootstrapper.ServerInfo;
 import java.util.List;
 import java.util.Map;
@@ -137,6 +138,8 @@ static final class DiscoveryMechanism {
       @Nullable
       final OutlierDetection outlierDetection;
       final Map<String, Struct> filterMetadata;
+      @Nullable
+      final BackendMetricPropagation backendMetricPropagation;
 
       enum Type {
         EDS,
@@ -146,7 +149,8 @@ enum Type {
       private DiscoveryMechanism(String cluster, Type type, @Nullable String edsServiceName,
           @Nullable String dnsHostName, @Nullable ServerInfo lrsServerInfo,
           @Nullable Long maxConcurrentRequests, @Nullable UpstreamTlsContext tlsContext,
-          Map<String, Struct> filterMetadata, @Nullable OutlierDetection outlierDetection) {
+          Map<String, Struct> filterMetadata, @Nullable OutlierDetection outlierDetection,
+          @Nullable BackendMetricPropagation backendMetricPropagation) {
         this.cluster = checkNotNull(cluster, "cluster");
         this.type = checkNotNull(type, "type");
         this.edsServiceName = edsServiceName;
@@ -156,27 +160,33 @@ private DiscoveryMechanism(String cluster, Type type, @Nullable String edsServic
         this.tlsContext = tlsContext;
         this.filterMetadata = ImmutableMap.copyOf(checkNotNull(filterMetadata, "filterMetadata"));
         this.outlierDetection = outlierDetection;
+        this.backendMetricPropagation = backendMetricPropagation;
       }
 
       static DiscoveryMechanism forEds(String cluster, @Nullable String edsServiceName,
           @Nullable ServerInfo lrsServerInfo, @Nullable Long maxConcurrentRequests,
           @Nullable UpstreamTlsContext tlsContext, Map<String, Struct> filterMetadata,
-          OutlierDetection outlierDetection) {
-        return new DiscoveryMechanism(cluster, Type.EDS, edsServiceName, null, lrsServerInfo,
-            maxConcurrentRequests, tlsContext, filterMetadata, outlierDetection);
+          OutlierDetection outlierDetection,
+          @Nullable BackendMetricPropagation backendMetricPropagation) {
+        return new DiscoveryMechanism(cluster, Type.EDS, edsServiceName,
+            null, lrsServerInfo, maxConcurrentRequests, tlsContext,
+            filterMetadata, outlierDetection, backendMetricPropagation);
       }
 
       static DiscoveryMechanism forLogicalDns(String cluster, String dnsHostName,
           @Nullable ServerInfo lrsServerInfo, @Nullable Long maxConcurrentRequests,
-          @Nullable UpstreamTlsContext tlsContext, Map<String, Struct> filterMetadata) {
+          @Nullable UpstreamTlsContext tlsContext, Map<String, Struct> filterMetadata,
+          @Nullable BackendMetricPropagation backendMetricPropagation) {
         return new DiscoveryMechanism(cluster, Type.LOGICAL_DNS, null, dnsHostName,
-            lrsServerInfo, maxConcurrentRequests, tlsContext, filterMetadata, null);
+            lrsServerInfo, maxConcurrentRequests, tlsContext, filterMetadata, null,
+            backendMetricPropagation);
       }
 
       @Override
       public int hashCode() {
         return Objects.hash(cluster, type, lrsServerInfo, maxConcurrentRequests, tlsContext,
-            edsServiceName, dnsHostName, filterMetadata, outlierDetection);
+            edsServiceName, dnsHostName, filterMetadata,
+            outlierDetection, backendMetricPropagation);
       }
 
       @Override