stats/opentelemetry: record retry attempts from clientStream

stats/opentelemetry/client_metrics.go

@@ -79,6 +79,7 @@ func getOrCreateCallInfo(ctx context.Context, cc *grpc.ClientConn, method string
 			target: cc.CanonicalTarget(),
 			method: determineMethod(method, opts...),
 		}
+		ci.previousRPCAttempts = new(atomic.Uint32)
 		ctx = setCallInfo(ctx, ci)
 	}
 	return ctx, ci

stats/opentelemetry/client_tracing.go

@@ -21,6 +21,7 @@ import (
 	"log"
 	"strings"
 
+	"go.opentelemetry.io/otel/attribute"
 	otelcodes "go.opentelemetry.io/otel/codes"
 	"go.opentelemetry.io/otel/trace"
 	"google.golang.org/grpc"
@@ -83,7 +84,10 @@ func (h *clientTracingHandler) finishTrace(err error, ts trace.Span) {
 // 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 *clientTracingHandler) traceTagRPC(ctx context.Context, ai *attemptInfo, nameResolutionDelayed bool) (context.Context, *attemptInfo) {
+//
+// Note: The passed attemptInfo pointer (ai) is mutated in-place. Fields such as
+// ai.traceSpan are updated directly. No new attemptInfo is returned.
+func (h *clientTracingHandler) traceTagRPC(ctx context.Context, ai *attemptInfo, nameResolutionDelayed bool) context.Context {
 	// 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.
@@ -98,7 +102,7 @@ func (h *clientTracingHandler) traceTagRPC(ctx context.Context, ai *attemptInfo,
 	carrier := otelinternaltracing.NewOutgoingCarrier(ctx)
 	h.options.TraceOptions.TextMapPropagator.Inject(ctx, carrier)
 	ai.traceSpan = span
-	return carrier.Context(), ai
+	return carrier.Context()
 }
 
 // createCallTraceSpan creates a call span to put in the provided context using
@@ -121,7 +125,13 @@ func (h *clientTracingHandler) HandleConn(context.Context, stats.ConnStats) {}
 // TagRPC implements per RPC attempt context management for traces.
 func (h *clientTracingHandler) TagRPC(ctx context.Context, info *stats.RPCTagInfo) context.Context {
 	ctx, ai := getOrCreateRPCAttemptInfo(ctx)
-	ctx, ai = h.traceTagRPC(ctx, ai, info.NameResolutionDelay)
+	ci := getCallInfo(ctx)
+	if ci == nil {
+		logger.Error("context passed into client side stats handler (TagRPC) has no call info")
+		return ctx
+	}
+	ai.previousRPCAttempts = ci.previousRPCAttempts
+	ctx = h.traceTagRPC(ctx, ai, info.NameResolutionDelay)
 	return setRPCInfo(ctx, &rpcInfo{ai: ai})
 }
 
@@ -132,5 +142,22 @@ func (h *clientTracingHandler) HandleRPC(ctx context.Context, rs stats.RPCStats)
 		logger.Error("ctx passed into client side tracing handler trace event handling has no client attempt data present")
 		return
 	}
+
+	// Client-specific Begin attributes.
+	var previousRPCAttempts int64
+	if ri.ai.previousRPCAttempts != nil {
+		previousRPCAttempts = int64(ri.ai.previousRPCAttempts.Load())
+	}
+	if begin, ok := rs.(*stats.Begin); ok {
+		ri.ai.traceSpan.SetAttributes(
+			attribute.Bool("Client", begin.Client),
+			attribute.Bool("FailFast", begin.FailFast),
+			attribute.Int64("previous-rpc-attempts", previousRPCAttempts),
+			attribute.Bool("transparent-retry", begin.IsTransparentRetryAttempt),
+		)
+		if !begin.IsTransparentRetryAttempt && ri.ai.previousRPCAttempts != nil {
+			ri.ai.previousRPCAttempts.Add(1)
+		}
+	}
 	populateSpan(rs, ri.ai)
 }