From 5219ccddb612a8f885e311d12094d9e540490b9b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Mustafa=20Sencer=20=C3=96zcan?= <32759850+mustafasencer@users.noreply.github.com> Date: Tue, 13 Jan 2026 12:42:21 +0100 Subject: [PATCH] fix: improve resilience for unified storage and search service grpc clients (#116122) * fix: reliability * fix: resilience * fix: add connection backoff * fix: reduce backoff --- pkg/server/ring.go | 53 +++++++++++++++++++++ pkg/services/apiserver/options/storage.go | 57 +++++++++++++++++++---- pkg/storage/unified/client.go | 6 ++- pkg/storage/unified/client_retry.go | 15 ++++++ 4 files changed, 121 insertions(+), 10 deletions(-) diff --git a/pkg/server/ring.go b/pkg/server/ring.go index 90026e58391..4dc261d68c5 100644 --- a/pkg/server/ring.go +++ b/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, + }) +} diff --git a/pkg/services/apiserver/options/storage.go b/pkg/services/apiserver/options/storage.go index 28f6e1046ab..057858caa01 100644 --- a/pkg/services/apiserver/options/storage.go +++ b/pkg/services/apiserver/options/storage.go @@ -11,11 +11,16 @@ import ( "github.com/spf13/pflag" "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc" "google.golang.org/grpc" + "google.golang.org/grpc/backoff" + "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/keepalive" genericapiserver "k8s.io/apiserver/pkg/server" "k8s.io/apiserver/pkg/server/options" "k8s.io/client-go/rest" + grpc_retry "github.com/grpc-ecosystem/go-grpc-middleware/retry" + apiserverrest "github.com/grafana/grafana/pkg/apiserver/rest" "github.com/grafana/grafana/pkg/infra/tracing" secret "github.com/grafana/grafana/pkg/registry/apis/secret/contracts" @@ -232,19 +237,16 @@ func (o *StorageOptions) ApplyTo(serverConfig *genericapiserver.RecommendedConfi if o.StorageType != StorageTypeUnifiedGrpc { return nil } - conn, err := grpc.NewClient(o.Address, - grpc.WithStatsHandler(otelgrpc.NewClientHandler()), - grpc.WithTransportCredentials(insecure.NewCredentials()), - ) + + grpcOpts := o.buildGrpcDialOptions() + + conn, err := grpc.NewClient(o.Address, grpcOpts...) if err != nil { return err } var indexConn *grpc.ClientConn if o.SearchServerAddress != "" { - indexConn, err = grpc.NewClient(o.SearchServerAddress, - grpc.WithStatsHandler(otelgrpc.NewClientHandler()), - grpc.WithTransportCredentials(insecure.NewCredentials()), - ) + indexConn, err = grpc.NewClient(o.SearchServerAddress, grpcOpts...) if err != nil { return err } @@ -293,3 +295,42 @@ func (o *StorageOptions) ApplyTo(serverConfig *genericapiserver.RecommendedConfi serverConfig.RESTOptionsGetter = getter return nil } + +// buildGrpcDialOptions creates gRPC dial options with resilience mechanisms: +// - Round-robin load balancing with client-side health checking +// - Retry interceptor for transient connection issues +// - Keepalive for long-lived connections +func (o *StorageOptions) buildGrpcDialOptions() []grpc.DialOption { + // Retry interceptor for transient connection issues (codes.Unavailable includes connection refused) + retryInterceptor := grpc_retry.UnaryClientInterceptor( + grpc_retry.WithMax(3), + grpc_retry.WithBackoff(grpc_retry.BackoffExponentialWithJitter(time.Second, 0.5)), + grpc_retry.WithCodes(codes.ResourceExhausted, codes.Unavailable), + ) + + opts := []grpc.DialOption{ + grpc.WithStatsHandler(otelgrpc.NewClientHandler()), + grpc.WithTransportCredentials(insecure.NewCredentials()), + grpc.WithChainUnaryInterceptor(retryInterceptor), + grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"round_robin"}`), + 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, + }), + } + + if o.GrpcClientKeepaliveTime > 0 { + opts = append(opts, grpc.WithKeepaliveParams(keepalive.ClientParameters{ + Time: o.GrpcClientKeepaliveTime, + Timeout: 10 * time.Second, + PermitWithoutStream: true, + })) + } + + return opts +} diff --git a/pkg/storage/unified/client.go b/pkg/storage/unified/client.go index 82336b3a5d1..e5c396907e3 100644 --- a/pkg/storage/unified/client.go +++ b/pkg/storage/unified/client.go @@ -271,7 +271,7 @@ func grpcConn(address string, metrics *clientMetrics, clientKeepaliveTime time.D retryCfg := retryConfig{ Max: 3, Backoff: time.Second, - BackoffJitter: 0.5, + BackoffJitter: 0.1, } unary = append(unary, unaryRetryInterceptor(retryCfg)) unary = append(unary, unaryRetryInstrument(metrics.requestRetries)) @@ -288,13 +288,15 @@ func grpcConn(address string, metrics *clientMetrics, clientKeepaliveTime time.D opts = append(opts, grpc.WithStatsHandler(otelgrpc.NewClientHandler())) opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials())) - // Use round_robin to balances requests more evenly over the available Storage server. + // Use round_robin to balance requests more evenly over the available Storage server. opts = append(opts, grpc.WithDefaultServiceConfig(`{"loadBalancingPolicy":"round_robin"}`)) // Disable looking up service config from TXT DNS records. // This reduces the number of requests made to the DNS servers. opts = append(opts, grpc.WithDisableServiceConfig()) + opts = append(opts, connectionBackoffOptions()) + if clientKeepaliveTime > 0 { opts = append(opts, grpc.WithKeepaliveParams(keepalive.ClientParameters{ Time: clientKeepaliveTime, diff --git a/pkg/storage/unified/client_retry.go b/pkg/storage/unified/client_retry.go index 47b899e4198..3df79807673 100644 --- a/pkg/storage/unified/client_retry.go +++ b/pkg/storage/unified/client_retry.go @@ -9,6 +9,7 @@ import ( "github.com/grpc-ecosystem/go-grpc-middleware/util/metautils" "github.com/prometheus/client_golang/prometheus" "google.golang.org/grpc" + "google.golang.org/grpc/backoff" "google.golang.org/grpc/codes" ) @@ -44,3 +45,17 @@ func unaryRetryInstrument(metric *prometheus.CounterVec) grpc.UnaryClientInterce 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, + }) +}