Merge branch 'master' of https://github.com/grpc/grpc-java into child-channel-plugin

Author: AgraVator

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

@@ -49,6 +49,9 @@
  */
 public class DelayedClientCall<ReqT, RespT> extends ClientCall<ReqT, RespT> {
   private static final Logger logger = Logger.getLogger(DelayedClientCall.class.getName());
+
+  /** A string describing what this call is waiting on. */
+  private final String bufferContext;
   /**
    * A timer to monitor the initial deadline. The timer must be cancelled on transition to the real
    * call.
@@ -76,7 +79,11 @@ public class DelayedClientCall<ReqT, RespT> extends ClientCall<ReqT, RespT> {
   private DelayedListener<RespT> delayedListener;
 
   protected DelayedClientCall(
-      Executor callExecutor, ScheduledExecutorService scheduler, @Nullable Deadline deadline) {
+      String bufferContext,
+      Executor callExecutor,
+      ScheduledExecutorService scheduler,
+      @Nullable Deadline deadline) {
+    this.bufferContext = checkNotNull(bufferContext, "bufferContext");
     this.callExecutor = checkNotNull(callExecutor, "callExecutor");
     checkNotNull(scheduler, "scheduler");
     context = Context.current();
@@ -143,7 +150,8 @@ public void run() {
         }
         buf.append(seconds);
         buf.append(String.format(Locale.US, ".%09d", nanos));
-        buf.append("s");
+        buf.append("s waiting for ");
+        buf.append(bufferContext);
         cancel(
             Status.DEADLINE_EXCEEDED.withDescription(buf.toString()),
             // We should not cancel the call if the realCall is set because there could be a

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

@@ -998,9 +998,12 @@ private final class PendingCall<ReqT, RespT> extends DelayedClientCall<ReqT, Res
       final CallOptions callOptions;
       private final long callCreationTime;
 
-      PendingCall(
-          Context context, MethodDescriptor<ReqT, RespT> method, CallOptions callOptions) {
-        super(getCallExecutor(callOptions), scheduledExecutor, callOptions.getDeadline());
+      PendingCall(Context context, MethodDescriptor<ReqT, RespT> method, CallOptions callOptions) {
+        super(
+            "name_resolver",
+            getCallExecutor(callOptions),
+            scheduledExecutor,
+            callOptions.getDeadline());
         this.context = context;
         this.method = method;
         this.callOptions = callOptions;

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

@@ -66,8 +66,8 @@ public class DelayedClientCallTest {
 
   @Test
   public void allMethodsForwarded() throws Exception {
-    DelayedClientCall<String, Integer> delayedClientCall =
-        new DelayedClientCall<>(callExecutor, fakeClock.getScheduledExecutorService(), null);
+    DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
+        "test", callExecutor, fakeClock.getScheduledExecutorService(), null);
     callMeMaybe(delayedClientCall.setCall(mockRealCall));
     ForwardingTestUtil.testMethodsForwarded(
         ClientCall.class,
@@ -94,18 +94,22 @@ public Object get(Method method, int argPos, Class<?> clazz) {
   @Test
   public void deadlineExceededWhileCallIsStartedButStillPending() {
     DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
-        callExecutor, fakeClock.getScheduledExecutorService(), Deadline.after(10, SECONDS));
+        "tESt", callExecutor, fakeClock.getScheduledExecutorService(),
+        Deadline.after(10, SECONDS, fakeClock.getDeadlineTicker()));
 
     delayedClientCall.start(listener, new Metadata());
     fakeClock.forwardTime(10, SECONDS);
     verify(listener).onClose(statusCaptor.capture(), any(Metadata.class));
     assertThat(statusCaptor.getValue().getCode()).isEqualTo(Status.Code.DEADLINE_EXCEEDED);
+    assertThat(statusCaptor.getValue().getDescription())
+        .isEqualTo("Deadline CallOptions was exceeded after 10.000000000s waiting for tESt");
   }
 
   @Test
   public void listenerEventsPropagated() {
     DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
-        callExecutor, fakeClock.getScheduledExecutorService(), Deadline.after(10, SECONDS));
+        "test", callExecutor, fakeClock.getScheduledExecutorService(),
+        Deadline.after(10, SECONDS, fakeClock.getDeadlineTicker()));
     delayedClientCall.start(listener, new Metadata());
     callMeMaybe(delayedClientCall.setCall(mockRealCall));
     @SuppressWarnings("unchecked")
@@ -130,7 +134,7 @@ public void listenerEventsPropagated() {
   @Test
   public void setCallThenStart() {
     DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
-        callExecutor, fakeClock.getScheduledExecutorService(), null);
+        "test", callExecutor, fakeClock.getScheduledExecutorService(), null);
     callMeMaybe(delayedClientCall.setCall(mockRealCall));
     delayedClientCall.start(listener, new Metadata());
     delayedClientCall.request(1);
@@ -146,7 +150,7 @@ public void setCallThenStart() {
   @Test
   public void startThenSetCall() {
     DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
-        callExecutor, fakeClock.getScheduledExecutorService(), null);
+        "test", callExecutor, fakeClock.getScheduledExecutorService(), null);
     delayedClientCall.start(listener, new Metadata());
     delayedClientCall.request(1);
     Runnable r = delayedClientCall.setCall(mockRealCall);
@@ -167,7 +171,7 @@ public void startThenSetCall() {
   @SuppressWarnings("unchecked")
   public void cancelThenSetCall() {
     DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
-        callExecutor, fakeClock.getScheduledExecutorService(), null);
+        "test", callExecutor, fakeClock.getScheduledExecutorService(), null);
     delayedClientCall.start(listener, new Metadata());
     delayedClientCall.request(1);
     delayedClientCall.cancel("cancel", new StatusException(Status.CANCELLED));
@@ -183,7 +187,7 @@ public void cancelThenSetCall() {
   @SuppressWarnings("unchecked")
   public void setCallThenCancel() {
     DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
-        callExecutor, fakeClock.getScheduledExecutorService(), null);
+        "test", callExecutor, fakeClock.getScheduledExecutorService(), null);
     delayedClientCall.start(listener, new Metadata());
     delayedClientCall.request(1);
     Runnable r = delayedClientCall.setCall(mockRealCall);
@@ -206,7 +210,8 @@ public void delayedCallsRunUnderContext() throws Exception {
     Object goldenValue = new Object();
     DelayedClientCall<String, Integer> delayedClientCall =
         Context.current().withValue(contextKey, goldenValue).call(() ->
-            new DelayedClientCall<>(callExecutor, fakeClock.getScheduledExecutorService(), null));
+            new DelayedClientCall<>(
+                "test", callExecutor, fakeClock.getScheduledExecutorService(), null));
     AtomicReference<Context> readyContext = new AtomicReference<>();
     delayedClientCall.start(new ClientCall.Listener<Integer>() {
       @Override public void onReady() {
@@ -232,7 +237,7 @@ public void delayedCallsRunUnderContext() throws Exception {
   @Test
   public void listenerThrowsInPendingCallback_cancelsRealCall() {
     DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
-        callExecutor, fakeClock.getScheduledExecutorService(), null);
+        "test", callExecutor, fakeClock.getScheduledExecutorService(), null);
     final RuntimeException boom = new RuntimeException("boom");
     ClientCall.Listener<Integer> throwingListener = new ClientCall.Listener<Integer>() {
       @Override
@@ -261,7 +266,7 @@ public void start(Listener<Integer> listener, Metadata metadata) {
   @Test
   public void listenerThrowsInPendingOnHeaders_cancelsRealCall() {
     DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
-        callExecutor, fakeClock.getScheduledExecutorService(), null);
+        "test", callExecutor, fakeClock.getScheduledExecutorService(), null);
     final RuntimeException boom = new RuntimeException("boom");
     ClientCall.Listener<Integer> throwingListener = new ClientCall.Listener<Integer>() {
       @Override
@@ -286,7 +291,7 @@ public void start(Listener<Integer> listener, Metadata metadata) {
   @Test
   public void listenerThrowsInPendingOnReady_cancelsRealCall() {
     DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
-        callExecutor, fakeClock.getScheduledExecutorService(), null);
+        "test", callExecutor, fakeClock.getScheduledExecutorService(), null);
     final RuntimeException boom = new RuntimeException("boom");
     ClientCall.Listener<Integer> throwingListener = new ClientCall.Listener<Integer>() {
       @Override
@@ -311,7 +316,7 @@ public void start(Listener<Integer> listener, Metadata metadata) {
   @Test
   public void onCloseExceptionCaughtAndLogged() {
     DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
-        callExecutor, fakeClock.getScheduledExecutorService(), null);
+        "test", callExecutor, fakeClock.getScheduledExecutorService(), null);
     final RuntimeException boom = new RuntimeException("boom");
     final AtomicReference<Status> observed = new AtomicReference<>();
     ClientCall.Listener<Integer> throwingListener = new ClientCall.Listener<Integer>() {
@@ -339,7 +344,7 @@ public void start(Listener<Integer> listener, Metadata metadata) {
   @Test
   public void listenerThrowsInPassThroughOnMessage_cancelsRealCall() {
     DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
-        callExecutor, fakeClock.getScheduledExecutorService(), null);
+        "test", callExecutor, fakeClock.getScheduledExecutorService(), null);
     final RuntimeException boom = new RuntimeException("boom");
     ClientCall.Listener<Integer> throwingListener = new ClientCall.Listener<Integer>() {
       @Override
@@ -362,7 +367,7 @@ public void onMessage(Integer msg) {
   @Test
   public void listenerThrowsInPassThroughOnHeaders_cancelsRealCall() {
     DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
-        callExecutor, fakeClock.getScheduledExecutorService(), null);
+        "test", callExecutor, fakeClock.getScheduledExecutorService(), null);
     final RuntimeException boom = new RuntimeException("boom");
     ClientCall.Listener<Integer> throwingListener = new ClientCall.Listener<Integer>() {
       @Override
@@ -385,7 +390,7 @@ public void onHeaders(Metadata headers) {
   @Test
   public void listenerThrowsInPassThroughOnReady_cancelsRealCall() {
     DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
-        callExecutor, fakeClock.getScheduledExecutorService(), null);
+        "test", callExecutor, fakeClock.getScheduledExecutorService(), null);
     final RuntimeException boom = new RuntimeException("boom");
     ClientCall.Listener<Integer> throwingListener = new ClientCall.Listener<Integer>() {
       @Override
@@ -408,7 +413,7 @@ public void onReady() {
   @Test
   public void listenerThrowsInPassThrough_subsequentCallbacksSwallowedAndOnCloseOverridden() {
     DelayedClientCall<String, Integer> delayedClientCall = new DelayedClientCall<>(
-        callExecutor, fakeClock.getScheduledExecutorService(), null);
+        "test", callExecutor, fakeClock.getScheduledExecutorService(), null);
     final RuntimeException boom = new RuntimeException("boom");
     final AtomicReference<Integer> lastMessage = new AtomicReference<>();
     final AtomicReference<Status> closeStatus = new AtomicReference<>();

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

@@ -25,16 +25,19 @@
 import com.google.common.primitives.UnsignedInts;
 import com.google.errorprone.annotations.CheckReturnValue;
 import io.grpc.Attributes;
+import io.grpc.ConnectivityState;
 import io.grpc.EquivalentAddressGroup;
 import io.grpc.HttpConnectProxiedSocketAddress;
 import io.grpc.InternalEquivalentAddressGroup;
 import io.grpc.InternalLogId;
 import io.grpc.LoadBalancer;
 import io.grpc.LoadBalancerProvider;
 import io.grpc.LoadBalancerRegistry;
+import io.grpc.NameResolver;
 import io.grpc.Status;
 import io.grpc.StatusOr;
 import io.grpc.internal.GrpcUtil;
+import io.grpc.util.ForwardingLoadBalancerHelper;
 import io.grpc.util.GracefulSwitchLoadBalancer;
 import io.grpc.util.OutlierDetectionLoadBalancer.OutlierDetectionLoadBalancerConfig;
 import io.grpc.xds.CdsLoadBalancerProvider.CdsConfig;
@@ -83,15 +86,17 @@ final class CdsLoadBalancer2 extends LoadBalancer {
   private final Helper helper;
   private final LoadBalancerRegistry lbRegistry;
   private final ClusterState clusterState = new ClusterState();
+  private final CdsLbHelper cdsLbHelper = new CdsLbHelper();
   private GracefulSwitchLoadBalancer delegate;
+  private boolean addBackendServicePickDetailsLabel;
   // Following fields are effectively final.
   private String clusterName;
   private Subscription clusterSubscription;
 
   CdsLoadBalancer2(Helper helper, LoadBalancerRegistry lbRegistry) {
     this.helper = checkNotNull(helper, "helper");
     this.lbRegistry = checkNotNull(lbRegistry, "lbRegistry");
-    this.delegate = new GracefulSwitchLoadBalancer(helper);
+    this.delegate = new GracefulSwitchLoadBalancer(cdsLbHelper);
     logger = XdsLogger.withLogId(InternalLogId.allocate("cds-lb", helper.getAuthority()));
     logger.log(XdsLogLevel.INFO, "Created");
   }
@@ -126,6 +131,7 @@ public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
     XdsClusterConfig clusterConfig = clusterConfigOr.getValue();
 
     if (clusterConfig.getChildren() instanceof EndpointConfig) {
+      addBackendServicePickDetailsLabel = true;
       StatusOr<EdsUpdate> edsUpdate = getEdsUpdate(xdsConfig, clusterName);
       StatusOr<ClusterResolutionResult> statusOrResult = clusterState.edsUpdateToResult(
           clusterName,
@@ -156,8 +162,12 @@ public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
           resolvedAddresses.toBuilder()
             .setLoadBalancingPolicyConfig(gracefulConfig)
             .setAddresses(Collections.unmodifiableList(addresses))
+            .setAttributes(resolvedAddresses.getAttributes().toBuilder()
+                .set(NameResolver.ATTR_BACKEND_SERVICE, clusterName)
+                .build())
             .build());
     } else if (clusterConfig.getChildren() instanceof AggregateConfig) {
+      addBackendServicePickDetailsLabel = false;
       Map<String, PriorityChildConfig> priorityChildConfigs = new HashMap<>();
       List<String> leafClusters = ((AggregateConfig) clusterConfig.getChildren()).getLeafNames();
       for (String childCluster: leafClusters) {
@@ -196,7 +206,8 @@ public void handleNameResolutionError(Status error) {
   public void shutdown() {
     logger.log(XdsLogLevel.INFO, "Shutdown");
     delegate.shutdown();
-    delegate = new GracefulSwitchLoadBalancer(helper);
+    delegate = new GracefulSwitchLoadBalancer(cdsLbHelper);
+    addBackendServicePickDetailsLabel = false;
     if (clusterSubscription != null) {
       clusterSubscription.close();
       clusterSubscription = null;
@@ -206,6 +217,7 @@ public void shutdown() {
   @CheckReturnValue // don't forget to return up the stack after the fail call
   private Status fail(Status error) {
     delegate.shutdown();
+    addBackendServicePickDetailsLabel = false;
     helper.updateBalancingState(
         TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(error)));
     return Status.OK; // XdsNameResolver isn't a polling NR, so this value doesn't matter
@@ -215,6 +227,38 @@ private String errorPrefix() {
     return "CdsLb for " + clusterName + ": ";
   }
 
+  private final class CdsLbHelper extends ForwardingLoadBalancerHelper {
+    @Override
+    public void updateBalancingState(ConnectivityState newState, SubchannelPicker newPicker) {
+      if (addBackendServicePickDetailsLabel) {
+        newPicker = new BackendServiceMetricLabelSubchannelPicker(newPicker, clusterName);
+      }
+      delegate().updateBalancingState(newState, newPicker);
+    }
+
+    @Override
+    protected Helper delegate() {
+      return helper;
+    }
+  }
+
+  private static final class BackendServiceMetricLabelSubchannelPicker extends SubchannelPicker {
+    private final SubchannelPicker delegate;
+    private final String backendService;
+
+    private BackendServiceMetricLabelSubchannelPicker(
+        SubchannelPicker delegate, String backendService) {
+      this.delegate = checkNotNull(delegate, "delegate");
+      this.backendService = checkNotNull(backendService, "backendService");
+    }
+
+    @Override
+    public PickResult pickSubchannel(PickSubchannelArgs args) {
+      args.getPickDetailsConsumer().addOptionalLabel("grpc.lb.backend_service", backendService);
+      return delegate.pickSubchannel(args);
+    }
+  }
+
   /**
    * The number of bits assigned to the fractional part of fixed-point values. We normalize weights
    * to a fixed-point number between 0 and 1, representing that item's proportion of traffic (1 ==

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

@@ -33,7 +33,6 @@
 import io.grpc.InternalLogId;
 import io.grpc.LoadBalancer;
 import io.grpc.Metadata;
-import io.grpc.NameResolver;
 import io.grpc.Status;
 import io.grpc.internal.ForwardingClientStreamTracer;
 import io.grpc.internal.GrpcUtil;
@@ -154,9 +153,6 @@ public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
 
     return childSwitchLb.acceptResolvedAddresses(
         resolvedAddresses.toBuilder()
-            .setAttributes(attributes.toBuilder()
-              .set(NameResolver.ATTR_BACKEND_SERVICE, cluster)
-              .build())
             .setLoadBalancingPolicyConfig(config.childConfig)
             .build());
   }
@@ -409,7 +405,6 @@ private RequestLimitingSubchannelPicker(SubchannelPicker delegate,
       public PickResult pickSubchannel(PickSubchannelArgs args) {
         args.getCallOptions().getOption(ClusterImplLoadBalancerProvider.FILTER_METADATA_CONSUMER)
             .accept(filterMetadata);
-        args.getPickDetailsConsumer().addOptionalLabel("grpc.lb.backend_service", cluster);
         for (DropOverload dropOverload : dropPolicies) {
           int rand = random.nextInt(1_000_000);
           if (rand < dropOverload.dropsPerMillion()) {