add test

Author: ulascansenturk

stats/opentelemetry/retry_counter_test.go

@@ -0,0 +1,229 @@
+/*
+ * Copyright 2024 gRPC authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package opentelemetry_test
+
+import (
+	"context"
+	"io"
+	"slices"
+	"strconv"
+	"testing"
+	"time"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/sdk/trace/tracetest"
+	oteltrace "go.opentelemetry.io/otel/trace"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/credentials/insecure"
+	"google.golang.org/grpc/internal/stubserver"
+	"google.golang.org/grpc/metadata"
+	"google.golang.org/grpc/resolver"
+	"google.golang.org/grpc/resolver/manual"
+	"google.golang.org/grpc/stats/opentelemetry"
+	"google.golang.org/grpc/status"
+
+	testgrpc "google.golang.org/grpc/interop/grpc_testing"
+	testpb "google.golang.org/grpc/interop/grpc_testing"
+)
+
+// TestRetryCounter_IncrementsPerAttempt verifies that the previous-rpc-attempts
+// counter increments correctly across multiple retry attempts. This test ensures
+// that the counter is tracked per-call (not per-attempt) as required by issue #8700.
+func (s) TestRetryCounter_IncrementsPerAttempt(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, _ *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])
+						}
+						// Fail first 2 attempts, succeed on 3rd
+						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: "Attempt.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])
+						}
+						// Fail first 2 attempts, succeed on 3rd
+						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: "Attempt.grpc.testing.TestService.FullDuplexCall",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			mo, _ := defaultMetricsOptions(t, nil)
+			to, exporter := defaultTraceOptions(t)
+			rb := manual.NewBuilderWithScheme("retry-test")
+			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()
+
+			// Update resolver state
+			rb.UpdateState(resolver.State{Addresses: []resolver.Address{{Addr: ss.Address}}})
+
+			client := testgrpc.NewTestServiceClient(cc)
+			ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+			defer cancel()
+
+			if err := tt.doCall(ctx, client); err != nil {
+				t.Fatalf("%s call failed: %v", tt.name, err)
+			}
+
+			// Wait for spans to be exported
+			// We expect at least 3 attempt spans
+			var attemptSpans []tracetest.SpanStub
+			for ; ctx.Err() == nil; <-time.After(time.Millisecond) {
+				attemptSpans = nil
+				allSpans := exporter.GetSpans()
+				for _, span := range allSpans {
+					if span.Name == tt.spanName && span.SpanKind == oteltrace.SpanKindInternal {
+						attemptSpans = append(attemptSpans, span)
+					}
+				}
+				if len(attemptSpans) >= 3 {
+					break
+				}
+			}
+			if ctx.Err() != nil {
+				t.Fatalf("Timed out waiting for attempt spans: %v", ctx.Err())
+			}
+
+			if got, want := len(attemptSpans), 3; got != want {
+				t.Fatalf("Got %d attempt spans, want %d", got, want)
+			}
+
+			// Sort attempt spans by start time to get them in chronological order
+			slices.SortFunc(attemptSpans, func(a, b tracetest.SpanStub) int {
+				if a.StartTime.Before(b.StartTime) {
+					return -1
+				}
+				return 1
+			})
+
+			// Verify each attempt has the correct previous-rpc-attempts value
+			// Attempt 1 should have 0, Attempt 2 should have 1, Attempt 3 should have 2
+			for i, span := range attemptSpans {
+				var previousAttemptsAttr *attribute.KeyValue
+				for _, attr := range span.Attributes {
+					if attr.Key == "previous-rpc-attempts" {
+						previousAttemptsAttr = &attr
+						break
+					}
+				}
+
+				if previousAttemptsAttr == nil {
+					t.Errorf("Attempt %d (span %q) missing previous-rpc-attempts attribute", i+1, span.Name)
+					continue
+				}
+
+				got := previousAttemptsAttr.Value.AsInt64()
+				want := int64(i)
+				if got != want {
+					t.Errorf("Attempt %d (span %q): previous-rpc-attempts = %d, want %d", i+1, span.Name, got, want)
+				} else {
+					t.Logf("✓ Attempt %d correctly has previous-rpc-attempts = %d", i+1, got)
+				}
+			}
+		})
+	}
+}