opentelemetry, rls, xds: Refine delay attributes and add comprehensive nested priority/RLS delay tests (Proposal A121)

Author: AgraVator

opentelemetry/src/main/java/io/grpc/opentelemetry/GrpcOpenTelemetry.java

@@ -364,7 +364,7 @@ && isMetricEnabled("grpc.client.attempt.delay.duration", enableMetrics, disableD
    * Checks whether experimental client attempt and call delay observability is globally enabled.
    *
    * <p>Guarded strictly by the {@code GRPC_EXPERIMENTAL_ENABLE_DELAY_OBSERVABILITY} environment
-   * variable or JVM system property (defaults to {@code false}). When disabled, delay spans and
+   * variable (defaults to {@code false}). When disabled, delay spans and
    * duration histograms are suppressed to avoid runtime overhead.
    */
   static boolean isDelayObservabilityEnabled() {

opentelemetry/src/main/java/io/grpc/opentelemetry/OpenTelemetryMetricsModule.java

@@ -50,6 +50,7 @@
 import io.grpc.internal.StatsTraceContext.ServerCallMethodListener;
 import io.grpc.opentelemetry.GrpcOpenTelemetry.TargetFilter;
 import io.opentelemetry.api.baggage.Baggage;
+import io.opentelemetry.api.common.Attributes;
 import io.opentelemetry.api.common.AttributesBuilder;
 import io.opentelemetry.context.Context;
 import java.util.ArrayList;
@@ -251,7 +252,7 @@ public void recordAttemptDelayEnd() {
         activeDelayStopwatch = null;
         activeDelayType = null;
         if (module.resource.clientAttemptDelayCounter() != null) {
-          AttributesBuilder builder = io.opentelemetry.api.common.Attributes.builder()
+          AttributesBuilder builder = Attributes.builder()
               .put(METHOD_KEY, fullMethodName)
               .put(TARGET_KEY, target)
               .put("grpc.delay_type", delayType);

opentelemetry/src/test/java/io/grpc/opentelemetry/OpenTelemetryTracingModuleTest.java

@@ -42,8 +42,13 @@
 import io.grpc.ClientInterceptor;
 import io.grpc.ClientInterceptors;
 import io.grpc.ClientStreamTracer;
+import io.grpc.ConnectivityState;
 import io.grpc.KnownLength;
+import io.grpc.LoadBalancer;
+import io.grpc.LoadBalancerProvider;
+import io.grpc.LoadBalancerRegistry;
 import io.grpc.ManagedChannel;
+import io.grpc.ManagedChannelBuilder;
 import io.grpc.Metadata;
 import io.grpc.MethodDescriptor;
 import io.grpc.NoopServerCall;

rls/src/test/java/io/grpc/rls/CachingRlsLbClientTest.java

@@ -27,6 +27,7 @@
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.argThat;
 import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.atLeastOnce;
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.times;
@@ -625,6 +626,52 @@ public void rls_pendingLookup_returnsDelayAttributes() throws Exception {
     assertThat(pickResult.getDelayReason()).contains("Route Lookup Service query pending");
   }
 
+  @Test
+  public void rls_childPolicyDelayed_returnsDelayAttributes() throws Exception {
+    setUpRlsLbClient();
+    RlsProtoData.RouteLookupRequestKey routeLookupRequestKey =
+        RlsProtoData.RouteLookupRequestKey.create(
+            ImmutableMap.of(
+                "server", "bigtable.googleapis.com", "service-key", "service1",
+                "method-key", "create"));
+    rlsServerImpl.setLookupTable(
+        ImmutableMap.of(
+            routeLookupRequestKey,
+            RouteLookupResponse.create(
+                ImmutableList.of("target1"), "header")));
+
+    CachedRouteLookupResponse resp = getInSyncContext(routeLookupRequestKey);
+    assertThat(resp.hasData()).isFalse();
+    fakeClock.forwardTime(SERVER_LATENCY_MILLIS, TimeUnit.MILLISECONDS);
+
+    resp = getInSyncContext(routeLookupRequestKey);
+    assertThat(resp.hasData()).isTrue();
+
+    LbPolicyConfiguration.ChildPolicyWrapper wrapper = resp.getChildPolicyWrapper();
+    wrapper.getHelper().updateBalancingState(ConnectivityState.CONNECTING, new SubchannelPicker() {
+      @Override
+      public PickResult pickSubchannel(LoadBalancer.PickSubchannelArgs args) {
+        return PickResult.withNoResult("connecting", "TCP handshake in progress");
+      }
+    });
+
+    ArgumentCaptor<SubchannelPicker> pickerCaptor =
+        ArgumentCaptor.forClass(SubchannelPicker.class);
+    verify(helper, atLeastOnce())
+        .updateBalancingState(any(), pickerCaptor.capture());
+    PickResult pickResult = pickerCaptor.getValue().pickSubchannel(
+        new PickSubchannelArgsImpl(
+            TestMethodDescriptors.voidMethod().toBuilder()
+                .setFullMethodName("service1/create").build(),
+            new Metadata(),
+            CallOptions.DEFAULT,
+            new PickDetailsConsumer() {}));
+    assertThat(pickResult.hasResult()).isFalse();
+    assertThat(pickResult.getDelayType()).isEqualTo("connecting");
+    assertThat(pickResult.getDelayReason()).isEqualTo(
+        "RLS child (target1) delayed: TCP handshake in progress");
+  }
+
   @Test
   public void timeout_not_changing_picked_subchannel() throws Exception {
     setUpRlsLbClient();

xds/src/test/java/io/grpc/xds/PriorityLoadBalancerTest.java

@@ -1060,6 +1060,41 @@ public PickResult pickSubchannel(PickSubchannelArgs args) {
         .isEqualTo("waiting on priority group p0 (child_reason)");
   }
 
+  @Test
+  public void priorityPicker_nestedPriorities_composesTokens() throws Exception {
+    PriorityChildConfig priorityChildConfig0 =
+        new PriorityChildConfig(newChildConfig(fooLbProvider, new Object()), true);
+    PriorityLbConfig priorityLbConfig =
+        new PriorityLbConfig(ImmutableMap.of("p0", priorityChildConfig0), ImmutableList.of("p0"));
+
+    priorityLb.acceptResolvedAddresses(
+        ResolvedAddresses.newBuilder()
+            .setAddresses(ImmutableList.<EquivalentAddressGroup>of())
+            .setLoadBalancingPolicyConfig(priorityLbConfig)
+            .build());
+
+    Helper helper0 = Iterables.getOnlyElement(fooHelpers); // priority p0
+
+    SubchannelPicker nestedChildPicker = new SubchannelPicker() {
+      @Override
+      public PickResult pickSubchannel(PickSubchannelArgs args) {
+        return PickResult.withNoResult("p1:connecting",
+            "waiting on priority group p1 (child_reason)");
+      }
+    };
+    helper0.updateBalancingState(CONNECTING, nestedChildPicker);
+
+    verify(helper, atLeastOnce())
+        .updateBalancingState(eq(CONNECTING), pickerCaptor.capture());
+
+    SubchannelPicker priorityPicker = pickerCaptor.getValue();
+    PickResult result = priorityPicker.pickSubchannel(mock(PickSubchannelArgs.class));
+
+    assertThat(result.getDelayType()).isEqualTo("p0:p1:connecting");
+    assertThat(result.getDelayReason()).isEqualTo(
+        "waiting on priority group p0 (waiting on priority group p1 (child_reason))");
+  }
+
   private void assertLatestConnectivityState(ConnectivityState expectedState) {
     verify(helper, atLeastOnce())
         .updateBalancingState(connectivityStateCaptor.capture(), pickerCaptor.capture());