core, opentelemetry: Implement Attempt-Level RPC Delay Observability (Proposal A121)

api/src/main/java/io/grpc/ClientStreamTracer.java

@@ -57,6 +57,45 @@ public void streamCreated(@Grpc.TransportAttr Attributes transportAttrs, Metadat
   public void createPendingStream() {
   }
 
+  /**
+   * Called when an attempt-level delay segment (such as waiting for a load balancing pick or
+   * connection establishment) starts.
+   *
+   * <p>This method is invoked synchronously on the attempt thread. Implementations should start
+   * internal timers or child tracing spans (named strictly {@code "Attempt Delay"}) carrying the
+   * canonical {@code grpc.delay_type} attribute.
+   *
+   * @param delayType canonical low-cardinality label categorizing the delay (e.g., "connecting")
+   * @param delayReason high-cardinality diagnostic string describing granular runtime conditions
+   * @since 1.82.0
+   */
+  public void recordAttemptDelayStart(String delayType, String delayReason) {
+  }
+
+  /**
+   * Called when an attempt-level delay reason changes while the overall delay type remains
+   * constant (for example, when a priority load balancing policy fails over between tiers).
+   *
+   * <p>Implementations should record structured events (such as {@code "Delay state transition"})
+   * on the active delay span without recreating the span or resetting cumulative timers.
+   *
+   * @param delayReason updated high-cardinality diagnostic string describing new conditions
+   * @since 1.82.0
+   */
+  public void recordAttemptDelayReasonChanged(String delayReason) {
+  }
+
+  /**
+   * Called when an attempt-level delay segment ends upon successful pick or stream creation.
+   *
+   * <p>Implementations should simultaneously close active child tracing spans and record elapsed
+   * duration to the {@code grpc.client.attempt.delay.duration} histogram.
+   *
+   * @since 1.82.0
+   */
+  public void recordAttemptDelayEnd() {
+  }
+
   /**
    * Headers has been sent to the socket.
    */

api/src/main/java/io/grpc/LoadBalancer.java

@@ -547,25 +547,32 @@ public static final class PickResult {
     // True if the result is created by withDrop()
     private final boolean drop;
     @Nullable private final String authorityOverride;
+    @Nullable private final String delayType;
+    @Nullable private final String delayReason;
 
     private PickResult(
         @Nullable Subchannel subchannel, @Nullable ClientStreamTracer.Factory streamTracerFactory,
         Status status, boolean drop) {
-      this.subchannel = subchannel;
-      this.streamTracerFactory = streamTracerFactory;
-      this.status = checkNotNull(status, "status");
-      this.drop = drop;
-      this.authorityOverride = null;
+      this(subchannel, streamTracerFactory, status, drop, null, null, null);
     }
 
     private PickResult(
         @Nullable Subchannel subchannel, @Nullable ClientStreamTracer.Factory streamTracerFactory,
         Status status, boolean drop, @Nullable String authorityOverride) {
+      this(subchannel, streamTracerFactory, status, drop, authorityOverride, null, null);
+    }
+
+    private PickResult(
+        @Nullable Subchannel subchannel, @Nullable ClientStreamTracer.Factory streamTracerFactory,
+        Status status, boolean drop, @Nullable String authorityOverride,
+        @Nullable String delayType, @Nullable String delayReason) {
       this.subchannel = subchannel;
       this.streamTracerFactory = streamTracerFactory;
       this.status = checkNotNull(status, "status");
       this.drop = drop;
       this.authorityOverride = authorityOverride;
+      this.delayType = delayType;
+      this.delayReason = delayReason;
     }
 
     /**
@@ -677,7 +684,7 @@ public static PickResult withSubchannel(Subchannel subchannel) {
      */
     public PickResult copyWithSubchannel(Subchannel subchannel) {
       return new PickResult(checkNotNull(subchannel, "subchannel"), streamTracerFactory,
-          status, drop, authorityOverride);
+          status, drop, authorityOverride, delayType, delayReason);
     }
 
     /**
@@ -688,7 +695,9 @@ public PickResult copyWithSubchannel(Subchannel subchannel) {
      */
     public PickResult copyWithStreamTracerFactory(
         @Nullable ClientStreamTracer.Factory streamTracerFactory) {
-      return new PickResult(subchannel, streamTracerFactory, status, drop, authorityOverride);
+      return new PickResult(
+          subchannel, streamTracerFactory, status, drop, authorityOverride, delayType,
+          delayReason);
     }
 
     /**
@@ -725,6 +734,31 @@ public static PickResult withNoResult() {
       return NO_RESULT;
     }
 
+    /**
+     * No decision could be made.  The RPC will stay buffered with a specific delay type and reason.
+     *
+     * @param delayType low-cardinality root cause label (e.g., "connecting")
+     * @param delayReason high-cardinality diagnostic string for trace events
+     * @since 1.82.0
+     */
+    public static PickResult withNoResult(String delayType, String delayReason) {
+      Preconditions.checkNotNull(delayType, "delayType");
+      Preconditions.checkNotNull(delayReason, "delayReason");
+      return new PickResult(null, null, Status.OK, false, null, delayType, delayReason);
+    }
+
+    /** Returns the delay type label if any. */
+    @Nullable
+    public String getDelayType() {
+      return delayType;
+    }
+
+    /** Returns the diagnostic delay reason if any. */
+    @Nullable
+    public String getDelayReason() {
+      return delayReason;
+    }
+
     /** Returns the authority override if any. */
     @ExperimentalApi("https://github.com/grpc/grpc-java/issues/11656")
     @Nullable

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

@@ -37,6 +37,7 @@
 import java.util.Collection;
 import java.util.Collections;
 import java.util.LinkedHashSet;
+import java.util.Objects;
 import java.util.concurrent.Executor;
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
@@ -157,7 +158,9 @@ public final ClientStream newStream(
         synchronized (lock) {
           PickerState newerState = pickerState;
           if (state == newerState) {
-            return createPendingStream(args, tracers, pickResult);
+            String delayType = determineQueuingDelayType(pickResult);
+            String delayReason = determineQueuingDelayReason(pickResult);
+            return createPendingStream(args, tracers, pickResult, delayType, delayReason);
           }
           state = newerState;
         }
@@ -173,8 +176,8 @@ public final ClientStream newStream(
    */
   @GuardedBy("lock")
   private PendingStream createPendingStream(PickSubchannelArgs args, ClientStreamTracer[] tracers,
-      PickResult pickResult) {
-    PendingStream pendingStream = new PendingStream(args, tracers);
+      PickResult pickResult, @Nullable String delayType, @Nullable String delayReason) {
+    PendingStream pendingStream = new PendingStream(args, tracers, delayType, delayReason);
     if (args.getCallOptions().isWaitForReady() && pickResult != null && pickResult.hasResult()) {
       pendingStream.lastPickStatus = pickResult.getStatus();
     }
@@ -245,7 +248,7 @@ public final void shutdownNow(Status status) {
     }
     if (savedReportTransportTerminated != null) {
       for (PendingStream stream : savedPendingStreams) {
-        Runnable runnable = stream.setStream(
+        Runnable runnable = stream.setStreamAndEndDelay(
             new FailingClientStream(status, RpcProgress.REFUSED, stream.tracers));
         if (runnable != null) {
           // Drain in-line instead of using an executor as failing stream just throws everything
@@ -303,6 +306,7 @@ final void reprocess(@Nullable SubchannelPicker picker) {
       final ClientTransport transport = GrpcUtil.getTransportFromPickResult(pickResult,
           callOptions.isWaitForReady());
       if (transport != null) {
+        stream.endDelay();
         Executor executor = defaultAppExecutor;
         // createRealStream may be expensive. It will start real streams on the transport. If
         // there are pending requests, they will be serialized too, which may be expensive. Since
@@ -315,7 +319,11 @@ final void reprocess(@Nullable SubchannelPicker picker) {
           executor.execute(runnable);
         }
         toRemove.add(stream);
-      }  // else: stay pending
+      } else { // stay pending
+        String delayType = determineQueuingDelayType(pickResult);
+        String delayReason = determineQueuingDelayReason(pickResult);
+        stream.updateDelay(delayType, delayReason);
+      }
     }
 
     synchronized (lock) {
@@ -356,16 +364,110 @@ public InternalLogId getLogId() {
     return logId;
   }
 
+  private static String determineQueuingDelayType(@Nullable PickResult pickResult) {
+    if (pickResult == null) {
+      return "connecting";
+    }
+    if (pickResult.getSubchannel() != null) {
+      return "subchannel_state_mismatch";
+    }
+    if (!pickResult.getStatus().isOk()) {
+      return "picker_failing_with_wait_for_ready";
+    }
+    if (pickResult.getDelayType() != null) {
+      return pickResult.getDelayType();
+    }
+    return "connecting";
+  }
+
+  private static String determineQueuingDelayReason(@Nullable PickResult pickResult) {
+    if (pickResult == null) {
+      return "client channel: waiting for picker";
+    }
+    if (pickResult.getSubchannel() != null) {
+      return "subchannel returned by LB picker has no connected subchannel";
+    }
+    if (!pickResult.getStatus().isOk()) {
+      return "wait_for_ready RPC failed with status: " + pickResult.getStatus();
+    }
+    if (pickResult.getDelayReason() != null) {
+      return pickResult.getDelayReason();
+    }
+    return "client channel: waiting for picker";
+  }
+
   private class PendingStream extends DelayedStream {
     private final PickSubchannelArgs args;
     private final Context context = Context.current();
     private final ClientStreamTracer[] tracers;
     private volatile Status lastPickStatus;
+    @GuardedBy("this")
+    @Nullable private String activeDelayType;
+    @GuardedBy("this")
+    @Nullable private String activeDelayReason;
 
-    private PendingStream(PickSubchannelArgs args, ClientStreamTracer[] tracers) {
+    private PendingStream(PickSubchannelArgs args, ClientStreamTracer[] tracers,
+        @Nullable String initialType, @Nullable String initialReason) {
       super("connecting_and_lb");
       this.args = args;
       this.tracers = tracers;
+      this.activeDelayType = initialType;
+      this.activeDelayReason = initialReason;
+      if (initialType != null) {
+        for (ClientStreamTracer tracer : tracers) {
+          tracer.recordAttemptDelayStart(initialType, initialReason != null ? initialReason : "");
+        }
+      }
+    }
+
+    /**
+     * Updates active attempt delay telemetry state upon load balancing state transitions.
+     *
+     * <p>If {@code newType} differs from the active delay type, active segment timers and child
+     * spans are ended and a new segment is initiated. If only {@code newReason} changes, a
+     * structured transition event is appended to the active span without span re-creation.
+     */
+    synchronized void updateDelay(@Nullable String newType, @Nullable String newReason) {
+      if (!Objects.equals(activeDelayType, newType)) {
+        // Delay type changed (e.g., from RLS lookup to connecting). End the previous delay.
+        if (activeDelayType != null) {
+          for (ClientStreamTracer tracer : tracers) {
+            tracer.recordAttemptDelayEnd();
+          }
+        }
+        activeDelayType = newType;
+        activeDelayReason = null;
+        if (newType != null) {
+          for (ClientStreamTracer tracer : tracers) {
+            tracer.recordAttemptDelayStart(newType, newReason != null ? newReason : "");
+          }
+        }
+      }
+      if (newType != null && newReason != null && !Objects.equals(activeDelayReason, newReason)) {
+        // Delay type is unchanged, but the reason changed (e.g., priority failover).
+        activeDelayReason = newReason;
+        for (ClientStreamTracer tracer : tracers) {
+          tracer.recordAttemptDelayReasonChanged(newReason);
+        }
+      }
+    }
+
+    /**
+     * Ends active attempt delay segment telemetry upon stream creation or stream cancellation.
+     */
+    synchronized void endDelay() {
+      if (activeDelayType != null) {
+        for (ClientStreamTracer tracer : tracers) {
+          tracer.recordAttemptDelayEnd();
+        }
+        activeDelayType = null;
+        activeDelayReason = null;
+      }
+    }
+
+    Runnable setStreamAndEndDelay(ClientStream stream) {
+      endDelay();
+      return setStream(stream);
     }
 
     /** Runnable may be null. */
@@ -386,11 +488,12 @@ private Runnable createRealStream(ClientTransport transport, String authorityOve
         // been called on the delayed stream.
         realStream.setAuthority(authorityOverride);
       }
-      return setStream(realStream);
+      return setStreamAndEndDelay(realStream);
     }
 
     @Override
     public void cancel(Status reason) {
+      endDelay();
       super.cancel(reason);
       synchronized (lock) {
         if (reportTransportTerminated != null) {

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

@@ -39,6 +39,21 @@ public void createPendingStream() {
     delegate().createPendingStream();
   }
 
+  @Override
+  public void recordAttemptDelayStart(String delayType, String delayReason) {
+    delegate().recordAttemptDelayStart(delayType, delayReason);
+  }
+
+  @Override
+  public void recordAttemptDelayReasonChanged(String delayReason) {
+    delegate().recordAttemptDelayReasonChanged(delayReason);
+  }
+
+  @Override
+  public void recordAttemptDelayEnd() {
+    delegate().recordAttemptDelayEnd();
+  }
+
   @Override
   public void outboundHeaders() {
     delegate().outboundHeaders();

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) {
@@ -340,10 +343,13 @@ void processSubchannelState(SubchannelData subchannelData, ConnectivityStateInfo
         // the current address of a valid index exists.
         if ((!enableHappyEyeballs && !addressIndex.isValid())
             || (addressIndex.isValid() && !subchannels.containsKey(
-            addressIndex.getCurrentAddress()))) {
+                addressIndex.getCurrentAddress()))) {
           addressIndex.seekTo(getAddress(subchannelData.subchannel));
         }
-        updateBalancingState(CONNECTING, new FixedResultPicker(PickResult.withNoResult()));
+        updateBalancingState(
+            CONNECTING,
+            new FixedResultPicker(
+                PickResult.withNoResult("connecting", "pick_first: attempting to connect")));
         break;
 
       case READY:
@@ -441,8 +447,9 @@ private void updateHealthCheckedState(SubchannelData subchannelData) {
       updateBalancingState(TRANSIENT_FAILURE, new FixedResultPicker(PickResult.withError(
           subchannelData.healthStateInfo.getStatus())));
     } else if (concludedState != TRANSIENT_FAILURE) {
-      updateBalancingState(subchannelData.getHealthState(),
-          new FixedResultPicker(PickResult.withNoResult()));
+      updateBalancingState(subchannelData.getHealthState(), new FixedResultPicker(
+          PickResult.withNoResult("connecting",
+              "health check state: " + subchannelData.getHealthState())));
     }
   }
 
@@ -668,7 +675,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");
     }
   }