stats/opentelemetry: add trace event for name resolution delay

clientconn.go

@@ -675,22 +675,26 @@
 	cc.mu.Unlock()
 }
 
-// waitForResolvedAddrs blocks until the resolver has provided addresses or the
-// context expires.  Returns nil unless the context expires first; otherwise
-// returns a status error based on the context.
-func (cc *ClientConn) waitForResolvedAddrs(ctx context.Context) error {
+// waitForResolvedAddrs blocks until the resolver provides addresses or the
+// context expires, whichever happens first. Error is nil unless the context
+// expires first; otherwise returns a status error based on the context. The
+// returned boolean indicates whether it did block or not. If the resolution
+// has already happened once before, it returns false without blocking.
+// Otherwise, it wait for the resolution and return true if resolution
+// succeeded or false along with error if resolution failed.
+func (cc *ClientConn) waitForResolvedAddrs(ctx context.Context) (bool, error) {
 	// This is on the RPC path, so we use a fast path to avoid the
 	// more-expensive "select" below after the resolver has returned once.
 	if cc.firstResolveEvent.HasFired() {
-		return nil
+		return false, nil
 	}
 	select {
 	case <-cc.firstResolveEvent.Done():
-		return nil
+		return true, nil
 	case <-ctx.Done():
-		return status.FromContextError(ctx.Err()).Err()
+		return false, status.FromContextError(ctx.Err()).Err()
 	case <-cc.ctx.Done():
-		return ErrClientConnClosing
+		return false, ErrClientConnClosing
 	}
 }
 

stats/handlers.go

@@ -38,6 +38,9 @@ type RPCTagInfo struct {
 	// FailFast indicates if this RPC is failfast.
 	// This field is only valid on client side, it's always false on server side.
 	FailFast bool
+	// NameResolutionDelay indicates if there was a delay in the name resolution.
+	// This field is only valid on client side, it's always false on server side.
+	NameResolutionDelay bool
 }
 
 // Handler defines the interface for the related stats handling (e.g., RPCs, connections).

stats/opentelemetry/client_metrics.go

@@ -190,6 +190,7 @@ func (h *clientStatsHandler) TagRPC(ctx context.Context, info *stats.RPCTagInfo)
 		xdsLabels: labels.TelemetryLabels,
 		method:    removeLeadingSlash(info.FullMethodName),
 	}
+	ai.nameResolutionDelayed = info.NameResolutionDelay
 	if h.options.isTracingEnabled() {
 		ctx, ai = h.traceTagRPC(ctx, ai)
 	}

stats/opentelemetry/opentelemetry.go

@@ -211,6 +211,9 @@ type attemptInfo struct {
 	countSentMsg        uint32
 	countRecvMsg        uint32
 	previousRPCAttempts uint32
+	// NameResolutionDelay indicates if there was a delay in the name resolution.
+	// This field is only valid on client side, it's always false on server side.
+	nameResolutionDelayed bool
 }
 
 type clientMetrics struct {

stats/opentelemetry/trace.go

@@ -52,6 +52,9 @@
 		)
 		// increment previous rpc attempts applicable for next attempt
 		atomic.AddUint32(&ai.previousRPCAttempts, 1)
+		if ai.nameResolutionDelayed {
+			span.AddEvent("Delayed name resolution complete")
+		}
 	case *stats.PickerUpdated:
 		span.AddEvent("Delayed LB pick complete")
 	case *stats.InPayload:

stream.go

@@ -212,14 +212,15 @@ func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 	}
 	// Provide an opportunity for the first RPC to see the first service config
 	// provided by the resolver.
-	if err := cc.waitForResolvedAddrs(ctx); err != nil {
+	nameResolutionDelayed, err := cc.waitForResolvedAddrs(ctx)
+	if err != nil {
 		return nil, err
 	}
 
 	var mc serviceconfig.MethodConfig
 	var onCommit func()
 	newStream := func(ctx context.Context, done func()) (iresolver.ClientStream, error) {
-		return newClientStreamWithParams(ctx, desc, cc, method, mc, onCommit, done, opts...)
+		return newClientStreamWithParams(ctx, desc, cc, method, mc, onCommit, done, nameResolutionDelayed, opts...)
 	}
 
 	rpcInfo := iresolver.RPCInfo{Context: ctx, Method: method}
@@ -257,7 +258,7 @@ func newClientStream(ctx context.Context, desc *StreamDesc, cc *ClientConn, meth
 	return newStream(ctx, func() {})
 }
 
-func newClientStreamWithParams(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, mc serviceconfig.MethodConfig, onCommit, doneFunc func(), opts ...CallOption) (_ iresolver.ClientStream, err error) {
+func newClientStreamWithParams(ctx context.Context, desc *StreamDesc, cc *ClientConn, method string, mc serviceconfig.MethodConfig, onCommit, doneFunc func(), nameResolutionDelayed bool, opts ...CallOption) (_ iresolver.ClientStream, err error) {
 	callInfo := defaultCallInfo()
 	if mc.WaitForReady != nil {
 		callInfo.failFast = !*mc.WaitForReady
@@ -321,19 +322,20 @@ func newClientStreamWithParams(ctx context.Context, desc *StreamDesc, cc *Client
 	}
 
 	cs := &clientStream{
-		callHdr:      callHdr,
-		ctx:          ctx,
-		methodConfig: &mc,
-		opts:         opts,
-		callInfo:     callInfo,
-		cc:           cc,
-		desc:         desc,
-		codec:        callInfo.codec,
-		compressorV0: compressorV0,
-		compressorV1: compressorV1,
-		cancel:       cancel,
-		firstAttempt: true,
-		onCommit:     onCommit,
+		callHdr:             callHdr,
+		ctx:                 ctx,
+		methodConfig:        &mc,
+		opts:                opts,
+		callInfo:            callInfo,
+		cc:                  cc,
+		desc:                desc,
+		codec:               callInfo.codec,
+		compressorV0:        compressorV0,
+		compressorV1:        compressorV1,
+		cancel:              cancel,
+		firstAttempt:        true,
+		onCommit:            onCommit,
+		NameResolutionDelay: nameResolutionDelayed,
 	}
 	if !cc.dopts.disableRetry {
 		cs.retryThrottler = cc.retryThrottler.Load().(*retryThrottler)
@@ -417,7 +419,7 @@ func (cs *clientStream) newAttemptLocked(isTransparent bool) (*csAttempt, error)
 	var beginTime time.Time
 	shs := cs.cc.dopts.copts.StatsHandlers
 	for _, sh := range shs {
-		ctx = sh.TagRPC(ctx, &stats.RPCTagInfo{FullMethodName: method, FailFast: cs.callInfo.failFast})
+		ctx = sh.TagRPC(ctx, &stats.RPCTagInfo{FullMethodName: method, FailFast: cs.callInfo.failFast, NameResolutionDelay: cs.NameResolutionDelay})
 		beginTime = time.Now()
 		begin := &stats.Begin{
 			Client:                    true,
@@ -573,6 +575,9 @@ type clientStream struct {
 	onCommit         func()
 	replayBuffer     []replayOp // operations to replay on retry
 	replayBufferSize int        // current size of replayBuffer
+	// NameResolutionDelay indicates if there was a delay in the name resolution.
+	// This field is only valid on client side, it's always false on server side.
+	NameResolutionDelay bool
 }
 
 type replayOp struct {

test/clientconn_test.go

@@ -29,9 +29,12 @@ import (
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/internal/channelz"
+	"google.golang.org/grpc/internal/stubserver"
 	testgrpc "google.golang.org/grpc/interop/grpc_testing"
 	testpb "google.golang.org/grpc/interop/grpc_testing"
+	"google.golang.org/grpc/resolver"
 	"google.golang.org/grpc/resolver/manual"
+	"google.golang.org/grpc/stats"
 	"google.golang.org/grpc/status"
 )
 
@@ -83,3 +86,132 @@ func (s) TestClientConnClose_WithPendingRPC(t *testing.T) {
 		t.Fatal(err)
 	}
 }
+
+type testStatsHandler struct {
+	nameResolutionDelayed bool
+}
+
+// TagRPC is called when an RPC is initiated and allows adding metadata to the
+// context. It checks if the RPC experienced a name resolution delay and
+// updates the handler's state.
+func (h *testStatsHandler) TagRPC(ctx context.Context, rpcInfo *stats.RPCTagInfo) context.Context {
+	h.nameResolutionDelayed = rpcInfo.NameResolutionDelay
+	return ctx
+}
+
+// This method is required to satisfy the stats.Handler interface.
+func (h *testStatsHandler) HandleRPC(_ context.Context, _ stats.RPCStats) {}
+
+// TagConn exists to satisfy stats.Handler.
+func (h *testStatsHandler) TagConn(ctx context.Context, _ *stats.ConnTagInfo) context.Context {
+	return ctx
+}
+
+// HandleConn exists to satisfy stats.Handler.
+func (h *testStatsHandler) HandleConn(_ context.Context, _ stats.ConnStats) {}
+
+// startStubServer initializes a stub gRPC server and returns its address and cleanup function.
+func startStubServer(t *testing.T) (*stubserver.StubServer, func()) {
+	t.Helper()
+	stub := &stubserver.StubServer{
+		EmptyCallF: func(_ context.Context, _ *testpb.Empty) (*testpb.Empty, error) {
+			t.Log("EmptyCall received and processed")
+			return &testpb.Empty{}, nil
+		},
+	}
+	if err := stub.Start(nil); err != nil {
+		t.Fatalf("Failed to start StubServer: %v", err)
+	}
+	return stub, func() { stub.Stop() }
+}
+
+// createTestClient sets up a gRPC client connection with a manual resolver.
+func createTestClient(t *testing.T, scheme string, statsHandler *testStatsHandler) (*grpc.ClientConn, *manual.Resolver) {
+	t.Helper()
+	rb := manual.NewBuilderWithScheme(scheme)
+	cc, err := grpc.NewClient(
+		scheme+":///test.server",
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithResolvers(rb),
+		grpc.WithStatsHandler(statsHandler),
+	)
+	if err != nil {
+		t.Fatalf("grpc.NewClient() failed: %v", err)
+	}
+	return cc, rb
+}
+
+// TestClientConnRPC_WithoutNameResolutionDelay verify that if the resolution
+// has already happened once before at the time of making RPC, the name
+// resolution flag is not set indicating there was no delay in name resolution.
+func (s) TestClientConnRPC_WithoutNameResolutionDelay(t *testing.T) {
+	stub, cleanup := startStubServer(t)
+	defer cleanup()
+
+	statsHandler := &testStatsHandler{}
+	rb := manual.NewBuilderWithScheme("instant")
+	rb.InitialState(resolver.State{Addresses: []resolver.Address{{Addr: stub.Address}}})
+	cc, err := grpc.NewClient(rb.Scheme()+":///test.server",
+		grpc.WithTransportCredentials(insecure.NewCredentials()),
+		grpc.WithResolvers(rb),
+		grpc.WithStatsHandler(statsHandler),
+	)
+	if err != nil {
+		t.Fatalf("grpc.NewClient() failed: %v", err)
+	}
+	defer cc.Close()
+
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+	client := testgrpc.NewTestServiceClient(cc)
+	// Verify that the RPC succeeds.
+	if _, err := client.EmptyCall(ctx, &testpb.Empty{}); err != nil {
+		t.Fatalf("First RPC failed unexpectedly: %v", err)
+	}
+	// Verify that name resolution did not happen.
+	if statsHandler.nameResolutionDelayed {
+		t.Fatalf("Expected nameResolutionDelayed to be false, but got %v", statsHandler.nameResolutionDelayed)
+	}
+}
+
+// TestStatsHandlerDetectsResolutionDelay verifies that if this is the
+// first time resolution is happening at the time of making RPC,
+// nameResolutionDelayed flag was set indicating there was a delay in name
+// resolution waiting for resolver to return addresses.
+func (s) TestClientConnRPC_WithNameResolutionDelay(t *testing.T) {
+	stub, cleanup := startStubServer(t)
+	defer cleanup()
+
+	statsHandler := &testStatsHandler{}
+	cc, rb := createTestClient(t, "delayed", statsHandler)
+	defer cc.Close()
+
+	client := testgrpc.NewTestServiceClient(cc)
+	rpcError := make(chan error, 1)
+	ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+	defer cancel()
+
+	go func() {
+		t.Log("RPC waiting for resolved addresses")
+		_, err := client.EmptyCall(ctx, &testpb.Empty{})
+		rpcError <- err
+	}()
+
+	timer := time.AfterFunc(100*time.Millisecond, func() {
+		rb.UpdateState(resolver.State{Addresses: []resolver.Address{{Addr: stub.Address}}})
+		t.Log("Resolver updated to return addresses.")
+	})
+	defer timer.Stop()
+
+	select {
+	case err := <-rpcError:
+		if err != nil {
+			t.Fatalf("RPC failed after resolution: %v", err)
+		}
+		if !statsHandler.nameResolutionDelayed {
+			t.Fatalf("Expected statsHandler.nameResolutionDelayed to be true, but got %v", statsHandler.nameResolutionDelayed)
+		}
+	case <-ctx.Done():
+		t.Fatal("Test setup timed out unexpectedly.")
+	}
+}