stats/opentelemetry: add trace event for name resolution delay (#8074)

Author: vinothkumarr227

clientconn.go

@@ -689,22 +689,31 @@ func (cc *ClientConn) Connect() {
 	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 has succeeded or return false along with error if resolution has
+// 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
 	}
+	internal.NewStreamWaitingForResolver()
 	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
 	}
 }
 

internal/internal.go

@@ -266,6 +266,13 @@ var (
 	TimeAfterFunc = func(d time.Duration, f func()) Timer {
 		return time.AfterFunc(d, f)
 	}
+
+	// NewStreamWaitingForResolver is a test hook that is triggered when a
+	// new stream blocks while waiting for name resolution. This can be
+	// used in tests to synchronize resolver updates and avoid race conditions.
+	// When set, the function will be called before the stream enters
+	// the blocking state.
+	NewStreamWaitingForResolver = func() {}
 )
 
 // HealthChecker defines the signature of the client-side LB channel health

stats/handlers.go

@@ -38,6 +38,15 @@ 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 the RPC needed to wait for the
+	// initial name resolver update before it could begin. This should only
+	// happen if the channel is IDLE when the RPC is started.  Note that
+	// all retry or hedging attempts for an RPC that experienced a delay
+	// will have it set.
+	//
+	// This field is only valid on the client side; it is always false on
+	// the server side.
+	NameResolutionDelay bool
 }
 
 // Handler defines the interface for the related stats handling (e.g., RPCs, connections).

stats/opentelemetry/client_metrics.go

@@ -191,7 +191,7 @@ func (h *clientStatsHandler) TagRPC(ctx context.Context, info *stats.RPCTagInfo)
 		method:    removeLeadingSlash(info.FullMethodName),
 	}
 	if h.options.isTracingEnabled() {
-		ctx, ai = h.traceTagRPC(ctx, ai)
+		ctx, ai = h.traceTagRPC(ctx, ai, info.NameResolutionDelay)
 	}
 	return setRPCInfo(ctx, &rpcInfo{
 		ai: ai,

stats/opentelemetry/client_tracing.go

@@ -25,14 +25,25 @@ import (
 	otelinternaltracing "google.golang.org/grpc/stats/opentelemetry/internal/tracing"
 )
 
-const tracerName = "grpc-go"
+const (
+	delayedResolutionEventName = "Delayed name resolution complete"
+	tracerName                 = "grpc-go"
+)
 
 // traceTagRPC populates provided context with a new span using the
 // TextMapPropagator supplied in trace options and internal itracing.carrier.
 // It creates a new outgoing carrier which serializes information about this
 // span into gRPC Metadata, if TextMapPropagator is provided in the trace
 // options. if TextMapPropagator is not provided, it returns the context as is.
-func (h *clientStatsHandler) traceTagRPC(ctx context.Context, ai *attemptInfo) (context.Context, *attemptInfo) {
+func (h *clientStatsHandler) traceTagRPC(ctx context.Context, ai *attemptInfo, nameResolutionDelayed bool) (context.Context, *attemptInfo) {
+	// Add a "Delayed name resolution complete" event to the call span
+	// if there was name resolution delay. In case of multiple retry attempts,
+	// ensure that event is added only once.
+	callSpan := trace.SpanFromContext(ctx)
+	ci := getCallInfo(ctx)
+	if nameResolutionDelayed && !ci.nameResolutionEventAdded.Swap(true) && callSpan.SpanContext().IsValid() {
+		callSpan.AddEvent(delayedResolutionEventName)
+	}
 	mn := "Attempt." + strings.Replace(ai.method, "/", ".", -1)
 	tracer := h.options.TraceOptions.TracerProvider.Tracer(tracerName, trace.WithInstrumentationVersion(grpc.Version))
 	ctx, span := tracer.Start(ctx, mn)

stats/opentelemetry/e2e_test.go

@@ -21,6 +21,7 @@ import (
 	"fmt"
 	"io"
 	"slices"
+	"strconv"
 	"testing"
 	"time"
 
@@ -48,9 +49,11 @@ import (
 	"go.opentelemetry.io/otel/sdk/trace"
 	"go.opentelemetry.io/otel/sdk/trace/tracetest"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/credentials/insecure"
 	"google.golang.org/grpc/encoding/gzip"
 	experimental "google.golang.org/grpc/experimental/opentelemetry"
+	"google.golang.org/grpc/internal"
 	"google.golang.org/grpc/internal/grpcsync"
 	"google.golang.org/grpc/internal/grpctest"
 	"google.golang.org/grpc/internal/stubserver"
@@ -59,9 +62,13 @@ import (
 	setup "google.golang.org/grpc/internal/testutils/xds/e2e/setup"
 	testgrpc "google.golang.org/grpc/interop/grpc_testing"
 	testpb "google.golang.org/grpc/interop/grpc_testing"
+	"google.golang.org/grpc/metadata"
 	"google.golang.org/grpc/orca"
+	"google.golang.org/grpc/resolver"
+	"google.golang.org/grpc/resolver/manual"
 	"google.golang.org/grpc/stats/opentelemetry"
 	"google.golang.org/grpc/stats/opentelemetry/internal/testutils"
+	"google.golang.org/grpc/status"
 )
 
 var defaultTestTimeout = 5 * time.Second
@@ -1560,3 +1567,168 @@ func (s) TestRPCSpanErrorStatus(t *testing.T) {
 		t.Fatalf("got rpc error %s, want %s", spans[0].Status.Description, rpcErrorMsg)
 	}
 }
+
+const delayedResolutionEventName = "Delayed name resolution complete"
+
+// TestTraceSpan_WithRetriesAndNameResolutionDelay verifies that
+// "Delayed name resolution complete" event is recorded in the call trace span
+// only once if any of the retry attempt encountered a delay in name resolution
+func (s) TestTraceSpan_WithRetriesAndNameResolutionDelay(t *testing.T) {
+	tests := []struct {
+		name      string
+		setupStub func() *stubserver.StubServer
+		doCall    func(context.Context, testgrpc.TestServiceClient) error
+		spanName  string
+	}{
+		{
+			name: "unary",
+			setupStub: func() *stubserver.StubServer {
+				return &stubserver.StubServer{
+					UnaryCallF: func(ctx context.Context, in *testpb.SimpleRequest) (*testpb.SimpleResponse, error) {
+						md, _ := metadata.FromIncomingContext(ctx)
+						headerAttempts := 0
+						if h := md["grpc-previous-rpc-attempts"]; len(h) > 0 {
+							headerAttempts, _ = strconv.Atoi(h[0])
+						}
+						if headerAttempts < 2 {
+							return nil, status.Errorf(codes.Unavailable, "retry (%d)", headerAttempts)
+						}
+						return &testpb.SimpleResponse{}, nil
+					},
+				}
+			},
+			doCall: func(ctx context.Context, client testgrpc.TestServiceClient) error {
+				_, err := client.UnaryCall(ctx, &testpb.SimpleRequest{})
+				return err
+			},
+			spanName: "grpc.testing.TestService.UnaryCall",
+		},
+		{
+			name: "streaming",
+			setupStub: func() *stubserver.StubServer {
+				return &stubserver.StubServer{
+					FullDuplexCallF: func(stream testgrpc.TestService_FullDuplexCallServer) error {
+						md, _ := metadata.FromIncomingContext(stream.Context())
+						headerAttempts := 0
+						if h := md["grpc-previous-rpc-attempts"]; len(h) > 0 {
+							headerAttempts, _ = strconv.Atoi(h[0])
+						}
+						if headerAttempts < 2 {
+							return status.Errorf(codes.Unavailable, "retry (%d)", headerAttempts)
+						}
+						for {
+							_, err := stream.Recv()
+							if err == io.EOF {
+								return nil
+							}
+							if err != nil {
+								return err
+							}
+						}
+					},
+				}
+			},
+			doCall: func(ctx context.Context, client testgrpc.TestServiceClient) error {
+				stream, err := client.FullDuplexCall(ctx)
+				if err != nil {
+					return err
+				}
+				if err := stream.Send(&testpb.StreamingOutputCallRequest{}); err != nil {
+					return err
+				}
+				if err := stream.CloseSend(); err != nil {
+					return err
+				}
+				_, err = stream.Recv()
+				if err != nil && err != io.EOF {
+					return err
+				}
+				return nil
+			},
+			spanName: "grpc.testing.TestService.FullDuplexCall",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			resolutionWait := grpcsync.NewEvent()
+			prevHook := internal.NewStreamWaitingForResolver
+			internal.NewStreamWaitingForResolver = func() { resolutionWait.Fire() }
+			defer func() { internal.NewStreamWaitingForResolver = prevHook }()
+
+			mo, _ := defaultMetricsOptions(t, nil)
+			to, exporter := defaultTraceOptions(t)
+			rb := manual.NewBuilderWithScheme("delayed")
+			ss := tt.setupStub()
+			opts := opentelemetry.Options{MetricsOptions: *mo, TraceOptions: *to}
+			if err := ss.Start([]grpc.ServerOption{opentelemetry.ServerOption(opts)}); err != nil {
+				t.Fatal(err)
+			}
+			defer ss.Stop()
+
+			retryPolicy := `{
+				"methodConfig": [{
+					"name": [{"service": "grpc.testing.TestService"}],
+					"retryPolicy": {
+						"maxAttempts": 3,
+						"initialBackoff": "0.05s",
+						"maxBackoff": "0.2s",
+						"backoffMultiplier": 1.0,
+						"retryableStatusCodes": ["UNAVAILABLE"]
+					}
+				}]
+			}`
+			cc, err := grpc.NewClient(
+				rb.Scheme()+":///test.server",
+				grpc.WithTransportCredentials(insecure.NewCredentials()),
+				grpc.WithResolvers(rb),
+				opentelemetry.DialOption(opts),
+				grpc.WithDefaultServiceConfig(retryPolicy),
+			)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer cc.Close()
+
+			client := testgrpc.NewTestServiceClient(cc)
+			ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+			defer cancel()
+
+			go func() {
+				<-resolutionWait.Done()
+				rb.UpdateState(resolver.State{Addresses: []resolver.Address{{Addr: ss.Address}}})
+			}()
+			if err := tt.doCall(ctx, client); err != nil {
+				t.Fatalf("%s call failed: %v", tt.name, err)
+			}
+
+			wantSpanInfo := traceSpanInfo{
+				name:     tt.spanName,
+				spanKind: oteltrace.SpanKindClient.String(),
+				events:   []trace.Event{{Name: delayedResolutionEventName}},
+			}
+			spans, err := waitForTraceSpans(ctx, exporter, []traceSpanInfo{wantSpanInfo})
+			if err != nil {
+				t.Fatal(err)
+			}
+			verifyTrace(t, spans, wantSpanInfo)
+		})
+	}
+}
+
+func verifyTrace(t *testing.T, spans tracetest.SpanStubs, want traceSpanInfo) {
+	match := false
+	for _, span := range spans {
+		if span.Name == want.name && span.SpanKind.String() == want.spanKind {
+			match = true
+			if diff := cmp.Diff(want.events, span.Events, cmpopts.IgnoreFields(trace.Event{}, "Time")); diff != "" {
+				t.Errorf("Span event mismatch for %q (kind: %s) (-want +got):\n%s",
+					want.name, want.spanKind, diff)
+			}
+			break
+		}
+	}
+	if !match {
+		t.Errorf("Expected span not found: %q (kind: %s)", want.name, want.spanKind)
+	}
+}

stats/opentelemetry/opentelemetry.go

@@ -25,6 +25,7 @@ package opentelemetry
 import (
 	"context"
 	"strings"
+	"sync/atomic"
 	"time"
 
 	otelattribute "go.opentelemetry.io/otel/attribute"
@@ -148,6 +149,10 @@ type callInfo struct {
 	target string
 
 	method string
+
+	// nameResolutionEventAdded is set when the resolver delay trace event
+	// is added. Prevents duplicate events, since it is reported per-attempt.
+	nameResolutionEventAdded atomic.Bool
 }
 
 type callInfoKey struct{}