Client watcher changes

android/src/main/java/io/grpc/android/UdsChannelBuilder.java

@@ -21,6 +21,7 @@
 import io.grpc.ExperimentalApi;
 import io.grpc.InsecureChannelCredentials;
 import io.grpc.ManagedChannelBuilder;
+import io.grpc.internal.GrpcUtil;
 import java.lang.reflect.InvocationTargetException;
 import javax.annotation.Nullable;
 import javax.net.SocketFactory;
@@ -81,7 +82,7 @@ public static ManagedChannelBuilder<?> forPath(String path, Namespace namespace)
       OKHTTP_CHANNEL_BUILDER_CLASS
           .getMethod("socketFactory", SocketFactory.class)
           .invoke(builder, new UdsSocketFactory(path, namespace));
-      return builder;
+      return builder.proxyDetector(GrpcUtil.NOOP_PROXY_DETECTOR);
     } catch (IllegalAccessException e) {
       throw new RuntimeException("Failed to create OkHttpChannelBuilder", e);
     } catch (NoSuchMethodException e) {

binder/src/main/java/io/grpc/binder/internal/BinderClientTransport.java

@@ -87,13 +87,6 @@ public final class BinderClientTransport extends BinderTransport
   @GuardedBy("this")
   private ScheduledFuture<?> readyTimeoutFuture; // != null iff timeout scheduled.
 
-  @GuardedBy("this")
-  @Nullable
-  private ListenableFuture<Status> authResultFuture; // null before we check auth.
-
-  @GuardedBy("this")
-  @Nullable
-  private ListenableFuture<Status> preAuthResultFuture; // null before we pre-auth.
 
   /**
    * Constructs a new transport instance.
@@ -193,7 +186,8 @@ private void preAuthorize(ServiceInfo serviceInfo) {
     // unauthorized server a chance to run, but the connection will still fail by SecurityPolicy
     // check later in handshake. Pre-auth remains effective at mitigating abuse because malware
     // can't typically control the exact timing of its installation.
-    preAuthResultFuture = checkServerAuthorizationAsync(serviceInfo.applicationInfo.uid);
+    ListenableFuture<Status> preAuthResultFuture =
+        register(checkServerAuthorizationAsync(serviceInfo.applicationInfo.uid));
     Futures.addCallback(
         preAuthResultFuture,
         new FutureCallback<Status>() {
@@ -314,12 +308,6 @@ void notifyTerminated() {
       readyTimeoutFuture.cancel(false);
       readyTimeoutFuture = null;
     }
-    if (preAuthResultFuture != null) {
-      preAuthResultFuture.cancel(false); // No effect if already complete.
-    }
-    if (authResultFuture != null) {
-      authResultFuture.cancel(false); // No effect if already complete.
-    }
     serviceBinding.unbind();
     clientTransportListener.transportTerminated();
   }
@@ -339,7 +327,8 @@ protected void handleSetupTransport(Parcel parcel) {
       } else {
         restrictIncomingBinderToCallsFrom(remoteUid);
         attributes = setSecurityAttrs(attributes, remoteUid);
-        authResultFuture = checkServerAuthorizationAsync(remoteUid);
+        ListenableFuture<Status> authResultFuture =
+            register(checkServerAuthorizationAsync(remoteUid));
         Futures.addCallback(
             authResultFuture,
             new FutureCallback<Status>() {
@@ -398,6 +387,7 @@ protected void handlePingResponse(Parcel parcel) {
     pingTracker.onPingResponse(parcel.readInt());
   }
 
+
   private static ClientStream newFailingClientStream(
       Status failure, Attributes attributes, Metadata headers, ClientStreamTracer[] tracers) {
     StatsTraceContext statsTraceContext =

binder/src/main/java/io/grpc/binder/internal/BinderTransport.java

@@ -45,9 +45,11 @@
 import java.util.ArrayList;
 import java.util.Iterator;
 import java.util.LinkedHashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -166,6 +168,9 @@ protected enum TransportState {
   @GuardedBy("this")
   private final LinkedHashSet<Integer> callIdsToNotifyWhenReady = new LinkedHashSet<>();
 
+  @GuardedBy("this")
+  private final List<Future<?>> ownedFutures = new ArrayList<>(); // To cancel upon terminate.
+
   @GuardedBy("this")
   protected Attributes attributes;
 
@@ -249,6 +254,13 @@ void releaseExecutors() {
     executorServicePool.returnObject(scheduledExecutorService);
   }
 
+  // Registers the specified future for eventual safe cancellation upon shutdown/terminate.
+  @GuardedBy("this")
+  protected final <T extends Future<?>> T register(T future) {
+    ownedFutures.add(future);
+    return future;
+  }
+
   @GuardedBy("this")
   boolean inState(TransportState transportState) {
     return this.transportState == transportState;
@@ -299,6 +311,8 @@ final void shutdownInternal(Status shutdownStatus, boolean forceTerminate) {
       sendShutdownTransaction();
       ArrayList<Inbound<?>> calls = new ArrayList<>(ongoingCalls.values());
       ongoingCalls.clear();
+      ArrayList<Future<?>> futuresToCancel = new ArrayList<>(ownedFutures);
+      ownedFutures.clear();
       scheduledExecutorService.execute(
           () -> {
             for (Inbound<?> inbound : calls) {
@@ -310,6 +324,11 @@ final void shutdownInternal(Status shutdownStatus, boolean forceTerminate) {
               notifyTerminated();
             }
             releaseExecutors();
+
+            for (Future<?> future : futuresToCancel) {
+              // Not holding any locks here just in case some listener runs on a direct Executor.
+              future.cancel(false); // No effect if already isDone().
+            }
           });
     }
   }

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

@@ -122,15 +122,17 @@ public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
             result.maxConcurrentRequests(),
             result.upstreamTlsContext(),
             result.filterMetadata(),
-            result.outlierDetection());
+            result.outlierDetection(),
+            result.backendMetricPropagation());
       } else {
         instance = DiscoveryMechanism.forLogicalDns(
             clusterName,
             result.dnsHostName(),
             result.lrsServerInfo(),
             result.maxConcurrentRequests(),
             result.upstreamTlsContext(),
-            result.filterMetadata());
+            result.filterMetadata(),
+            result.backendMetricPropagation());
       }
       gracefulConfig = GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(
           lbRegistry.getProvider(CLUSTER_RESOLVER_POLICY_NAME),

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

@@ -17,6 +17,7 @@
 package io.grpc.xds;
 
 import static com.google.common.base.Preconditions.checkNotNull;
+import static io.grpc.xds.client.LoadStatsManager2.isEnabledOrcaLrsPropagation;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.MoreObjects;
@@ -46,6 +47,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;
@@ -149,6 +151,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()
@@ -209,6 +212,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");
@@ -321,7 +326,7 @@ private ClusterLocality createClusterLocalityFromAttributes(Attributes addressAt
           (lrsServerInfo == null)
               ? null
               : xdsClient.addClusterLocalityStats(lrsServerInfo, cluster,
-                  edsServiceName, locality);
+                  edsServiceName, locality, backendMetricPropagation);
 
       return new ClusterLocality(localityStats, localityName);
     }
@@ -371,6 +376,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;
@@ -506,11 +516,19 @@ 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) {
+      if (isEnabledOrcaLrsPropagation) {
+        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

@@ -336,7 +336,7 @@ private static Map<String, PriorityChildConfig> generatePriorityChildConfigs(
           new ClusterImplConfig(
               discovery.cluster, discovery.edsServiceName, discovery.lrsServerInfo,
               discovery.maxConcurrentRequests, dropOverloads, endpointLbConfig,
-              discovery.tlsContext, discovery.filterMetadata);
+              discovery.tlsContext, discovery.filterMetadata, discovery.backendMetricPropagation);
       LoadBalancerProvider clusterImplLbProvider =
           lbRegistry.getProvider(XdsLbPolicies.CLUSTER_IMPL_POLICY_NAME);
       Object priorityChildPolicy = GracefulSwitchLoadBalancer.createLoadBalancingPolicyConfig(

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

@@ -30,6 +30,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.Map;
 import java.util.Objects;
@@ -152,6 +153,8 @@ static final class DiscoveryMechanism {
       @Nullable
       final OutlierDetection outlierDetection;
       final Map<String, Struct> filterMetadata;
+      @Nullable
+      final BackendMetricPropagation backendMetricPropagation;
 
       enum Type {
         EDS,
@@ -161,7 +164,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;
@@ -171,27 +175,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