api, census: add new pendingStreamCreated on clientStreamTracer and new tracer annotation (#10014)

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

@@ -39,6 +39,18 @@ public abstract class ClientStreamTracer extends StreamTracer {
   public void streamCreated(@Grpc.TransportAttr Attributes transportAttrs, Metadata headers) {
   }
 
+  /**
+   * Name resolution is completed and the connection starts getting established. This method is only
+   * invoked on the streams that encounter such delay.
+   *
+   * </p>gRPC buffers the client call if the remote address and configurations, e.g. timeouts and
+   * retry policy, are not ready. Asynchronously gRPC internally does the name resolution to get
+   * this information. The streams that are processed immediately on ready transports by the time
+   * the RPC comes do not go through the pending process, thus this callback will not be invoked.
+   */
+  public void createPendingStream() {
+  }
+
   /**
    * Headers has been sent to the socket.
    */

census/src/main/java/io/grpc/census/CensusTracingModule.java

@@ -299,6 +299,7 @@ final class CensusTracingModule {
     final Metadata.Key<SpanContext> tracingHeader;
     final boolean isSampledToLocalTracing;
     volatile int seqNo;
+    boolean isPendingStream;
 
     ClientTracer(
         Span span, Span parentSpan, Metadata.Key<SpanContext> tracingHeader,
@@ -315,6 +316,14 @@ final class CensusTracingModule {
         headers.discardAll(tracingHeader);
         headers.put(tracingHeader, span.getContext());
       }
+      if (isPendingStream) {
+        span.addAnnotation("Delayed LB pick complete");
+      }
+    }
+
+    @Override
+    public void createPendingStream() {
+      isPendingStream = true;
     }
 
     @Override

census/src/test/java/io/grpc/census/CensusModulesTest.java

@@ -746,6 +746,7 @@ public class CensusModulesTest {
         censusTracing.newClientCallTracer(spyClientSpan, method);
     Metadata headers = new Metadata();
     ClientStreamTracer clientStreamTracer = callTracer.newClientStreamTracer(STREAM_INFO, headers);
+    clientStreamTracer.createPendingStream();
     clientStreamTracer.streamCreated(Attributes.EMPTY, headers);
     verify(tracer).spanBuilderWithExplicitParent(
         eq("Attempt.package1.service2.method3"), eq(spyClientSpan));
@@ -767,6 +768,7 @@ public class CensusModulesTest {
         .putAttribute("previous-rpc-attempts", AttributeValue.longAttributeValue(0));
     inOrder.verify(spyAttemptSpan)
         .putAttribute("transparent-retry", AttributeValue.booleanAttributeValue(false));
+    inOrder.verify(spyAttemptSpan).addAnnotation("Delayed LB pick complete");
     inOrder.verify(spyAttemptSpan, times(2)).addMessageEvent(messageEventCaptor.capture());
     List<MessageEvent> events = messageEventCaptor.getAllValues();
     assertEquals(

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

@@ -183,6 +183,9 @@ final class DelayedClientTransport implements ManagedClientTransport {
     if (getPendingStreamsCount() == 1) {
       syncContext.executeLater(reportTransportInUse);
     }
+    for (ClientStreamTracer streamTracer : tracers) {
+      streamTracer.createPendingStream();
+    }
     return pendingStream;
   }
 

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

@@ -34,6 +34,11 @@ public abstract class ForwardingClientStreamTracer extends ClientStreamTracer {
     delegate().streamCreated(transportAttrs, headers);
   }
 
+  @Override
+  public void createPendingStream() {
+    delegate().createPendingStream();
+  }
+
   @Override
   public void outboundHeaders() {
     delegate().outboundHeaders();

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

@@ -33,6 +33,11 @@ public abstract class ForwardingClientStreamTracer extends ClientStreamTracer {
     delegate().streamCreated(transportAttrs, headers);
   }
 
+  @Override
+  public void createPendingStream() {
+    delegate().createPendingStream();
+  }
+
   @Override
   public void outboundHeaders() {
     delegate().outboundHeaders();