Merge remote-tracking branch 'origin' into axelav/dash-validator-app-mvp

pkg/server/ring.go

@@ -3,6 +3,7 @@ package server
 import (
 	"context"
 	"fmt"
+	"strconv"
 	"time"
 
 	"github.com/grafana/dskit/flagext"
@@ -15,11 +16,15 @@ import (
 	"github.com/grafana/grafana/pkg/services/featuremgmt"
 	"github.com/grafana/grafana/pkg/setting"
 	"github.com/grafana/grafana/pkg/storage/unified/resource"
+	grpc_retry "github.com/grpc-ecosystem/go-grpc-middleware/retry"
+	"github.com/grpc-ecosystem/go-grpc-middleware/util/metautils"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promauto"
 	"go.opentelemetry.io/otel"
 	"go.opentelemetry.io/otel/trace"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/backoff"
+	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/health/grpc_health_v1"
 )
 
@@ -111,14 +116,25 @@ func newClientPool(clientCfg grpcclient.Config, log log.Logger, reg prometheus.R
 		Help:    "Time spent executing requests to resource server.",
 		Buckets: prometheus.ExponentialBuckets(0.008, 4, 7),
 	}, []string{"operation", "status_code"})
+	factoryRequestRetries := promauto.With(reg).NewCounterVec(prometheus.CounterOpts{
+		Name: "resource_server_client_request_retries_total",
+		Help: "Total number of retries for requests to the resource server.",
+	}, []string{"operation"})
 
 	factory := ringclient.PoolInstFunc(func(inst ring.InstanceDesc) (ringclient.PoolClient, error) {
 		unaryInterceptors, streamInterceptors := grpcclient.Instrument(factoryRequestDuration)
+
+		// Add retry interceptors for transient connection issues
+		unaryInterceptors = append(unaryInterceptors, ringClientRetryInterceptor())
+		unaryInterceptors = append(unaryInterceptors, ringClientRetryInstrument(factoryRequestRetries))
+
 		opts, err := clientCfg.DialOption(unaryInterceptors, streamInterceptors, nil)
 		if err != nil {
 			return nil, err
 		}
 
+		opts = append(opts, connectionBackoffOptions())
+
 		conn, err := grpc.NewClient(inst.Addr, opts...)
 		if err != nil {
 			return nil, fmt.Errorf("failed to dial resource server %s %s: %s", inst.Id, inst.Addr, err)
@@ -135,3 +151,40 @@ func newClientPool(clientCfg grpcclient.Config, log log.Logger, reg prometheus.R
 
 	return ringclient.NewPool(resource.RingName, poolCfg, nil, factory, clientsCount, log)
 }
+
+// ringClientRetryInterceptor creates an interceptor to perform retries for unary methods.
+// It retries on ResourceExhausted and Unavailable codes, which are typical for
+// transient connection issues and rate limiting.
+func ringClientRetryInterceptor() grpc.UnaryClientInterceptor {
+	return grpc_retry.UnaryClientInterceptor(
+		grpc_retry.WithMax(3),
+		grpc_retry.WithBackoff(grpc_retry.BackoffExponentialWithJitter(time.Second, 0.1)),
+		grpc_retry.WithCodes(codes.ResourceExhausted, codes.Unavailable),
+	)
+}
+
+// ringClientRetryInstrument creates an interceptor to count retry attempts for metrics.
+func ringClientRetryInstrument(metric *prometheus.CounterVec) grpc.UnaryClientInterceptor {
+	return func(ctx context.Context, method string, req, resp interface{}, cc *grpc.ClientConn, invoker grpc.UnaryInvoker, opts ...grpc.CallOption) error {
+		// We can tell if a call is a retry by checking the retry attempt metadata.
+		attempt, err := strconv.Atoi(metautils.ExtractOutgoing(ctx).Get(grpc_retry.AttemptMetadataKey))
+		if err == nil && attempt > 0 {
+			metric.WithLabelValues(method).Inc()
+		}
+		return invoker(ctx, method, req, resp, cc, opts...)
+	}
+}
+
+// connectionBackoffOptions configures connection backoff parameters for faster recovery from
+// transient connection failures (e.g., during pod restarts).
+func connectionBackoffOptions() grpc.DialOption {
+	return grpc.WithConnectParams(grpc.ConnectParams{
+		Backoff: backoff.Config{
+			BaseDelay:  100 * time.Millisecond,
+			Multiplier: 1.6,
+			Jitter:     0.2,
+			MaxDelay:   10 * time.Second,
+		},
+		MinConnectTimeout: 5 * time.Second,
+	})
+}