opentelemetry: Implement dual Load Balancer delay spans and metrics

AgraVator · AgraVator · commit 6a3572bec7ed · 2026-06-22T14:12:07.000Z
diff --git a/api/src/main/java/io/grpc/LoadBalancer.java b/api/src/main/java/io/grpc/LoadBalancer.java
@@ -697,7 +697,9 @@ public PickResult copyWithSubchannel(Subchannel subchannel) {
      */
     public PickResult copyWithStreamTracerFactory(
         @Nullable ClientStreamTracer.Factory streamTracerFactory) {
-      return new PickResult(subchannel, streamTracerFactory, status, drop, authorityOverride, delayType, delayReason);
+      return new PickResult(
+          subchannel, streamTracerFactory, status, drop, authorityOverride, delayType,
+          delayReason);
     }
 
     /**
diff --git a/core/src/main/java/io/grpc/internal/DelayedClientTransport.java b/core/src/main/java/io/grpc/internal/DelayedClientTransport.java
@@ -158,8 +158,8 @@ public final ClientStream newStream(
         synchronized (lock) {
           PickerState newerState = pickerState;
           if (state == newerState) {
-            String delayType = determineQueuingDelayType(pickResult, callOptions.isWaitForReady());
-            String delayReason = determineQueuingDelayReason(pickResult, callOptions.isWaitForReady());
+            String delayType = determineQueuingDelayType(pickResult);
+            String delayReason = determineQueuingDelayReason(pickResult);
             return createPendingStream(args, tracers, pickResult, delayType, delayReason);
           }
           state = newerState;
@@ -320,10 +320,8 @@ final void reprocess(@Nullable SubchannelPicker picker) {
         }
         toRemove.add(stream);
       } else { // stay pending
-        String delayType = determineQueuingDelayType(
-            pickResult, stream.args.getCallOptions().isWaitForReady());
-        String delayReason = determineQueuingDelayReason(
-            pickResult, stream.args.getCallOptions().isWaitForReady());
+        String delayType = determineQueuingDelayType(pickResult);
+        String delayReason = determineQueuingDelayReason(pickResult);
         stream.updateDelay(delayType, delayReason);
       }
     }
@@ -366,8 +364,7 @@ public InternalLogId getLogId() {
     return logId;
   }
 
-  private static String determineQueuingDelayType(
-      @Nullable PickResult pickResult, boolean isWaitForReady) {
+  private static String determineQueuingDelayType(@Nullable PickResult pickResult) {
     if (pickResult == null) {
       return "client_channel_init";
     }
@@ -383,8 +380,7 @@ private static String determineQueuingDelayType(
     return "client_channel_init";
   }
 
-  private static String determineQueuingDelayReason(
-      @Nullable PickResult pickResult, boolean isWaitForReady) {
+  private static String determineQueuingDelayReason(@Nullable PickResult pickResult) {
     if (pickResult == null) {
       return "client channel: created LB policy.";
     }
diff --git a/core/src/main/java/io/grpc/internal/PickFirstLeafLoadBalancer.java b/core/src/main/java/io/grpc/internal/PickFirstLeafLoadBalancer.java
@@ -167,7 +167,10 @@ public Status acceptResolvedAddresses(ResolvedAddresses resolvedAddresses) {
     if (noOldAddrs) {
       // Make tests happy; they don't properly assume starting in CONNECTING
       rawConnectivityState = CONNECTING;
-      updateBalancingState(CONNECTING, new FixedResultPicker(PickResult.withNoResult()));
+      updateBalancingState(
+          CONNECTING,
+          new FixedResultPicker(
+              PickResult.withNoResult("connecting", "pick_first: address list updated")));
     }
 
     if (rawConnectivityState == READY) {
@@ -333,7 +336,10 @@ void processSubchannelState(SubchannelData subchannelData, ConnectivityStateInfo
 
       case CONNECTING:
         rawConnectivityState = CONNECTING;
-        updateBalancingState(CONNECTING, new FixedResultPicker(PickResult.withNoResult()));
+        updateBalancingState(
+            CONNECTING,
+            new FixedResultPicker(
+                PickResult.withNoResult("connecting", "pick_first: attempting to connect")));
         break;
 
       case READY:
@@ -653,7 +659,8 @@ public PickResult pickSubchannel(PickSubchannelArgs args) {
       if (connectionRequested.compareAndSet(false, true)) {
         helper.getSynchronizationContext().execute(pickFirstLeafLoadBalancer::requestConnection);
       }
-      return PickResult.withNoResult();
+      return PickResult.withNoResult(
+          "connecting", "pick_first: requesting connection");
     }
   }
 
diff --git a/core/src/main/java/io/grpc/internal/PickFirstLoadBalancer.java b/core/src/main/java/io/grpc/internal/PickFirstLoadBalancer.java
@@ -180,7 +180,8 @@ public PickResult pickSubchannel(PickSubchannelArgs args) {
       if (connectionRequested.compareAndSet(false, true)) {
         helper.getSynchronizationContext().execute(PickFirstLoadBalancer.this::requestConnection);
       }
-      return PickResult.withNoResult();
+      return PickResult.withNoResult(
+          "connecting", "pick_first: requesting connection");
     }
   }
 
diff --git a/core/src/test/java/io/grpc/internal/DelayedClientTransportTest.java b/core/src/test/java/io/grpc/internal/DelayedClientTransportTest.java
@@ -888,7 +888,8 @@ public void streamDelayMetrics_channelFallback_subchannelStateMismatch() {
     ClientStreamTracer mockTracer = mock(ClientStreamTracer.class);
     ClientStreamTracer[] customTracers = new ClientStreamTracer[] { mockTracer };
 
-    io.grpc.LoadBalancer.Subchannel disconnectedSubchannel = mock(io.grpc.LoadBalancer.Subchannel.class);
+    io.grpc.LoadBalancer.Subchannel disconnectedSubchannel =
+        mock(io.grpc.LoadBalancer.Subchannel.class);
     when(disconnectedSubchannel.getInternalSubchannel())
         .thenReturn(newTransportProvider(null));
 
diff --git a/grpclb/src/main/java/io/grpc/grpclb/GrpclbState.java b/grpclb/src/main/java/io/grpc/grpclb/GrpclbState.java
@@ -124,7 +124,8 @@ final class GrpclbState {
   static final RoundRobinEntry BUFFER_ENTRY = new RoundRobinEntry() {
       @Override
       public PickResult picked(PickSubchannelArgs args) {
-        return PickResult.withNoResult();
+        return PickResult.withNoResult(
+            "connecting", "grpclb: waiting for backend server list");
       }
 
       @Override
diff --git a/opentelemetry/src/main/java/io/grpc/opentelemetry/GrpcOpenTelemetry.java b/opentelemetry/src/main/java/io/grpc/opentelemetry/GrpcOpenTelemetry.java
@@ -232,6 +232,16 @@ static OpenTelemetryMetricsResource createMetricInstruments(Meter meter,
               .build());
     }
 
+    if (isMetricEnabled("grpc.client.attempt.delay", enableMetrics, disableDefault)) {
+      builder.clientAttemptDelayCounter(
+          meter.histogramBuilder(
+                  "grpc.client.attempt.delay")
+              .setUnit("s")
+              .setDescription("Time taken to complete a client call attempt delay")
+              .setExplicitBucketBoundariesAdvice(LATENCY_BUCKETS)
+              .build());
+    }
+
     if (isMetricEnabled("grpc.client.attempt.sent_total_compressed_message_size", enableMetrics,
         disableDefault)) {
       builder.clientTotalSentCompressedMessageSizeCounter(
diff --git a/opentelemetry/src/main/java/io/grpc/opentelemetry/OpenTelemetryMetricsModule.java b/opentelemetry/src/main/java/io/grpc/opentelemetry/OpenTelemetryMetricsModule.java
@@ -203,6 +203,8 @@ private static final class ClientTracer extends ClientStreamTracer {
     volatile String backendService;
     long attemptNanos;
     Code statusCode;
+    @Nullable private volatile Stopwatch activeDelayStopwatch;
+    @Nullable private volatile String activeDelayType;
 
     ClientTracer(CallAttemptsTracerFactory attemptsState, OpenTelemetryMetricsModule module,
         StreamInfo info, String target, String fullMethodName,
@@ -216,6 +218,59 @@ private static final class ClientTracer extends ClientStreamTracer {
       this.stopwatch = module.stopwatchSupplier.get().start();
     }
 
+    @Override
+    public void streamCreated(io.grpc.Attributes transportAtts, Metadata headers) {
+      delayEnded();
+    }
+
+    @Override
+    public void delayTypeStarted(String delayType) {
+      delayEnded();
+      activeDelayType = delayType;
+      activeDelayStopwatch = module.stopwatchSupplier.get().start();
+    }
+
+    @Override
+    public void delayEnded() {
+      Stopwatch delayStopwatch = activeDelayStopwatch;
+      String delayType = activeDelayType;
+      if (delayStopwatch != null && delayType != null) {
+        delayStopwatch.stop();
+        long delayNanos = delayStopwatch.elapsed(TimeUnit.NANOSECONDS);
+        activeDelayStopwatch = null;
+        activeDelayType = null;
+        if (module.resource.clientAttemptDelayCounter() != null) {
+          AttributesBuilder builder = io.opentelemetry.api.common.Attributes.builder()
+              .put(METHOD_KEY, fullMethodName)
+              .put(TARGET_KEY, target)
+              .put("grpc.delay_type", delayType);
+          if (module.localityEnabled) {
+            String savedLocality = locality;
+            if (savedLocality == null) {
+              savedLocality = "";
+            }
+            builder.put(LOCALITY_KEY, savedLocality);
+          }
+          if (module.backendServiceEnabled) {
+            String savedBackendService = backendService;
+            if (savedBackendService == null) {
+              savedBackendService = "";
+            }
+            builder.put(BACKEND_SERVICE_KEY, savedBackendService);
+          }
+          if (module.customLabelEnabled) {
+            builder.put(
+                CUSTOM_LABEL_KEY, info.getCallOptions().getOption(Grpc.CALL_OPTION_CUSTOM_LABEL));
+          }
+          for (OpenTelemetryPlugin.ClientStreamPlugin plugin : streamPlugins) {
+            plugin.addLabels(builder);
+          }
+          module.resource.clientAttemptDelayCounter()
+              .record(delayNanos * SECONDS_PER_NANO, builder.build(), attemptsState.otelContext);
+        }
+      }
+    }
+
     @Override
     public void inboundHeaders(Metadata headers) {
       for (OpenTelemetryPlugin.ClientStreamPlugin plugin : streamPlugins) {
@@ -262,6 +317,7 @@ public void inboundTrailers(Metadata trailers) {
 
     @Override
     public void streamClosed(Status status) {
+      delayEnded();
       stopwatch.stop();
       attemptNanos = stopwatch.elapsed(TimeUnit.NANOSECONDS);
       Deadline deadline = info.getCallOptions().getDeadline();
diff --git a/opentelemetry/src/main/java/io/grpc/opentelemetry/OpenTelemetryMetricsResource.java b/opentelemetry/src/main/java/io/grpc/opentelemetry/OpenTelemetryMetricsResource.java
@@ -35,6 +35,9 @@ abstract class OpenTelemetryMetricsResource {
   @Nullable
   abstract DoubleHistogram clientAttemptDurationCounter();
 
+  @Nullable
+  abstract DoubleHistogram clientAttemptDelayCounter();
+
   @Nullable
   abstract LongHistogram clientTotalSentCompressedMessageSizeCounter();
 
@@ -79,6 +82,8 @@ abstract static class Builder {
 
     abstract Builder clientAttemptDurationCounter(DoubleHistogram counter);
 
+    abstract Builder clientAttemptDelayCounter(DoubleHistogram counter);
+
     abstract Builder clientTotalSentCompressedMessageSizeCounter(LongHistogram counter);
 
     abstract Builder clientTotalReceivedCompressedMessageSizeCounter(
diff --git a/opentelemetry/src/main/java/io/grpc/opentelemetry/OpenTelemetryTracingModule.java b/opentelemetry/src/main/java/io/grpc/opentelemetry/OpenTelemetryTracingModule.java
@@ -192,6 +192,7 @@ private final class ClientTracer extends ClientStreamTracer {
     private final Span parentSpan;
     volatile int seqNo;
     boolean isPendingStream;
+    @Nullable private volatile Span activeDelaySpan;
 
     ClientTracer(Span span, Span parentSpan) {
       this.span = checkNotNull(span, "span");
@@ -200,6 +201,7 @@ private final class ClientTracer extends ClientStreamTracer {
 
     @Override
     public void streamCreated(Attributes transportAtts, Metadata headers) {
+      delayEnded();
       contextPropagators.getTextMapPropagator().inject(Context.current().with(span), headers,
           metadataSetter);
       if (isPendingStream) {
@@ -212,6 +214,36 @@ public void createPendingStream() {
       isPendingStream = true;
     }
 
+    @Override
+    public void delayTypeStarted(String delayType) {
+      if (activeDelaySpan != null) {
+        activeDelaySpan.end();
+      }
+      activeDelaySpan = otelTracer.spanBuilder("Attempt Delay: " + delayType)
+          .setParent(Context.current().with(span))
+          .setAttribute("grpc.delay_type", delayType)
+          .startSpan();
+    }
+
+    @Override
+    public void delayReasonAttached(String delayReason) {
+      Span delaySpan = activeDelaySpan;
+      if (delaySpan != null) {
+        delaySpan.addEvent(delayReason);
+      } else {
+        span.addEvent("delay_reason: " + delayReason);
+      }
+    }
+
+    @Override
+    public void delayEnded() {
+      Span delaySpan = activeDelaySpan;
+      if (delaySpan != null) {
+        delaySpan.end();
+        activeDelaySpan = null;
+      }
+    }
+
     @Override
     public void outboundMessageSent(
         int seqNo, long optionalWireSize, long optionalUncompressedSize) {
@@ -238,6 +270,7 @@ public void inboundUncompressedSize(long bytes) {
 
     @Override
     public void streamClosed(io.grpc.Status status) {
+      delayEnded();
       endSpanWithStatus(span, status);
     }
   }
diff --git a/opentelemetry/src/test/java/io/grpc/opentelemetry/OpenTelemetryTracingModuleTest.java b/opentelemetry/src/test/java/io/grpc/opentelemetry/OpenTelemetryTracingModuleTest.java
@@ -274,6 +274,31 @@ public void clientBasicTracingMocking() {
     inOrder.verifyNoMoreInteractions();
   }
 
+  @Test
+  public void clientDelayTracingMocking() {
+    Span mockDelaySpan = mock(Span.class);
+    when(mockSpanBuilder.setAttribute(
+            org.mockito.ArgumentMatchers.anyString(),
+            org.mockito.ArgumentMatchers.anyString()))
+        .thenReturn(mockSpanBuilder);
+    when(mockSpanBuilder.startSpan()).thenReturn(mockAttemptSpan, mockDelaySpan);
+
+    OpenTelemetryTracingModule tracingModule = new OpenTelemetryTracingModule(mockOpenTelemetry);
+    CallAttemptsTracerFactory callTracer =
+        tracingModule.newClientCallTracer(mockClientSpan, method);
+    ClientStreamTracer clientStreamTracer =
+        callTracer.newClientStreamTracer(STREAM_INFO, new Metadata());
+
+    clientStreamTracer.delayTypeStarted("connecting");
+    clientStreamTracer.delayReasonAttached("pick_first: attempting to connect");
+    clientStreamTracer.delayEnded();
+
+    verify(mockTracer).spanBuilder(eq("Attempt Delay: connecting"));
+    verify(mockSpanBuilder).setAttribute(eq("grpc.delay_type"), eq("connecting"));
+    verify(mockDelaySpan).addEvent(eq("pick_first: attempting to connect"));
+    verify(mockDelaySpan).end();
+  }
+
   @Test
   public void clientBasicTracingRule() {
     OpenTelemetryTracingModule tracingModule = new OpenTelemetryTracingModule(
@@ -381,6 +406,34 @@ public void clientBasicTracingRule() {
     assertEquals(attemptSpanData.hasEnded(), true);
   }
 
+  @Test
+  public void clientDelayTracingRule() {
+    OpenTelemetryTracingModule tracingModule = new OpenTelemetryTracingModule(
+        openTelemetryRule.getOpenTelemetry());
+    Span clientSpan = tracerRule.spanBuilder("test-client-span").startSpan();
+    CallAttemptsTracerFactory callTracer =
+        tracingModule.newClientCallTracer(clientSpan, method);
+    ClientStreamTracer clientStreamTracer =
+        callTracer.newClientStreamTracer(STREAM_INFO, new Metadata());
+
+    clientStreamTracer.delayTypeStarted("connecting");
+    clientStreamTracer.delayReasonAttached("pick_first: attempting to connect");
+    clientStreamTracer.delayEnded();
+    clientStreamTracer.streamClosed(Status.OK);
+    callTracer.callEnded(Status.OK);
+    clientSpan.end();
+
+    List<SpanData> spans = openTelemetryRule.getSpans();
+    assertEquals(3, spans.size());
+    SpanData delaySpanData = spans.get(0);
+
+    assertEquals("Attempt Delay: connecting", delaySpanData.getName());
+    assertEquals("connecting", delaySpanData.getAttributes().get(
+        io.opentelemetry.api.common.AttributeKey.stringKey("grpc.delay_type")));
+    assertEquals(1, delaySpanData.getEvents().size());
+    assertEquals("pick_first: attempting to connect", delaySpanData.getEvents().get(0).getName());
+  }
+
   @Test
   public void clientInterceptor() {
     testClientInterceptors(false);
diff --git a/util/src/test/java/io/grpc/util/RoundRobinLoadBalancerTest.java b/util/src/test/java/io/grpc/util/RoundRobinLoadBalancerTest.java
@@ -577,6 +577,16 @@ private void deliverSubchannelState(Subchannel subchannel, ConnectivityStateInfo
     testHelperInst.deliverSubchannelState(subchannel, newState);
   }
 
+  @Test
+  public void roundRobin_delayAttributes() {
+    acceptAddresses(servers, affinity);
+    verify(mockHelper, atLeastOnce())
+        .updateBalancingState(eq(CONNECTING), pickerCaptor.capture());
+    PickResult res = pickerCaptor.getValue().pickSubchannel(mockArgs);
+    assertThat(res.getDelayType()).isEqualTo("connecting");
+    assertThat(res.getDelayReason()).contains("attempting to connect");
+  }
+
   private static class FakeSocketAddress extends SocketAddress {
     final String name;
 
diff --git a/xds/src/main/java/io/grpc/xds/ClusterImplLoadBalancer.java b/xds/src/main/java/io/grpc/xds/ClusterImplLoadBalancer.java
@@ -200,7 +200,8 @@ public void shutdown() {
   private final class ClusterImplLbHelper extends ForwardingLoadBalancerHelper {
     private final AtomicLong inFlights;
     private ConnectivityState currentState = ConnectivityState.IDLE;
-    private SubchannelPicker currentPicker = new FixedResultPicker(PickResult.withNoResult());
+    private SubchannelPicker currentPicker = new FixedResultPicker(
+        PickResult.withNoResult("connecting", "cluster_impl: initializing"));
     private List<DropOverload> dropPolicies = Collections.emptyList();
     private long maxConcurrentRequests = DEFAULT_PER_CLUSTER_MAX_CONCURRENT_REQUESTS;
     @Nullable
diff --git a/xds/src/main/java/io/grpc/xds/LazyLoadBalancer.java b/xds/src/main/java/io/grpc/xds/LazyLoadBalancer.java
@@ -107,7 +107,8 @@ private final class LazyPicker extends SubchannelPicker {
       public PickResult pickSubchannel(PickSubchannelArgs args) {
         // activate() is a no-op after shutdown()
         helper.getSynchronizationContext().execute(LazyDelegate.this::activate);
-        return PickResult.withNoResult();
+        return PickResult.withNoResult(
+            "connecting", "lazy: waiting for connection");
       }
     }
   }
diff --git a/xds/src/main/java/io/grpc/xds/LeastRequestLoadBalancer.java b/xds/src/main/java/io/grpc/xds/LeastRequestLoadBalancer.java
diff --git a/xds/src/main/java/io/grpc/xds/WeightedRoundRobinLoadBalancer.java b/xds/src/main/java/io/grpc/xds/WeightedRoundRobinLoadBalancer.java
diff --git a/xds/src/main/java/io/grpc/xds/WeightedTargetLoadBalancer.java b/xds/src/main/java/io/grpc/xds/WeightedTargetLoadBalancer.java

Original file line number	Diff line number	Diff line change
`@@ -158,8 +158,8 @@ public final ClientStream newStream(`
`158`	`158`	`synchronized (lock) {`
`159`	`159`	`PickerState newerState = pickerState;`
`160`	`160`	`if (state == newerState) {`
`161`		`- String delayType = determineQueuingDelayType(pickResult, callOptions.isWaitForReady());`
`162`		`- String delayReason = determineQueuingDelayReason(pickResult, callOptions.isWaitForReady());`
	`161`	`+ String delayType = determineQueuingDelayType(pickResult);`
	`162`	`+ String delayReason = determineQueuingDelayReason(pickResult);`
`163`	`163`	`return createPendingStream(args, tracers, pickResult, delayType, delayReason);`
`164`	`164`	`}`
`165`	`165`	`state = newerState;`
`@@ -320,10 +320,8 @@ final void reprocess(@Nullable SubchannelPicker picker) {`
`320`	`320`	`}`
`321`	`321`	`toRemove.add(stream);`
`322`	`322`	`} else { // stay pending`
`323`		`- String delayType = determineQueuingDelayType(`
`324`		`- pickResult, stream.args.getCallOptions().isWaitForReady());`
`325`		`- String delayReason = determineQueuingDelayReason(`
`326`		`- pickResult, stream.args.getCallOptions().isWaitForReady());`
	`323`	`+ String delayType = determineQueuingDelayType(pickResult);`
	`324`	`+ String delayReason = determineQueuingDelayReason(pickResult);`
`327`	`325`	`stream.updateDelay(delayType, delayReason);`
`328`	`326`	`}`
`329`	`327`	`}`
`@@ -366,8 +364,7 @@ public InternalLogId getLogId() {`
`366`	`364`	`return logId;`
`367`	`365`	`}`
`368`	`366`
`369`		`- private static String determineQueuingDelayType(`
`370`		`- @Nullable PickResult pickResult, boolean isWaitForReady) {`
	`367`	`+ private static String determineQueuingDelayType(@Nullable PickResult pickResult) {`
`371`	`368`	`if (pickResult == null) {`
`372`	`369`	`return "client_channel_init";`
`373`	`370`	`}`
`@@ -383,8 +380,7 @@ private static String determineQueuingDelayType(`
`383`	`380`	`return "client_channel_init";`
`384`	`381`	`}`
`385`	`382`
`386`		`- private static String determineQueuingDelayReason(`
`387`		`- @Nullable PickResult pickResult, boolean isWaitForReady) {`
	`383`	`+ private static String determineQueuingDelayReason(@Nullable PickResult pickResult) {`
`388`	`384`	`if (pickResult == null) {`
`389`	`385`	`return "client channel: created LB policy.";`
`390`	`386`	`}`
Original file line number	Diff line number	Diff line change
`@@ -180,7 +180,8 @@ public PickResult pickSubchannel(PickSubchannelArgs args) {`
`180`	`180`	`if (connectionRequested.compareAndSet(false, true)) {`
`181`	`181`	`helper.getSynchronizationContext().execute(PickFirstLoadBalancer.this::requestConnection);`
`182`	`182`	`}`
`183`		`- return PickResult.withNoResult();`
	`183`	`+ return PickResult.withNoResult(`
	`184`	`+ "connecting", "pick_first: requesting connection");`
`184`	`185`	`}`
`185`	`186`	`}`
`186`	`187`
Original file line number	Diff line number	Diff line change
`@@ -124,7 +124,8 @@ final class GrpclbState {`
`124`	`124`	`static final RoundRobinEntry BUFFER_ENTRY = new RoundRobinEntry() {`
`125`	`125`	`@Override`
`126`	`126`	`public PickResult picked(PickSubchannelArgs args) {`
`127`		`- return PickResult.withNoResult();`
	`127`	`+ return PickResult.withNoResult(`
	`128`	`+ "connecting", "grpclb: waiting for backend server list");`
`128`	`129`	`}`
`129`	`130`
`130`	`131`	`@Override`