feat(unified-storage): add some basic retry logic for the grpc client (#101001)

This commit is contained in:
Jean-Philippe Quéméner
2025-02-20 16:17:50 +01:00
committed by GitHub
parent 522e75c750
commit e83f4fca1e
3 changed files with 63 additions and 1 deletions
+1 -1
View File
@@ -367,7 +367,7 @@ require (
github.com/grafana/loki/pkg/push v0.0.0-20231124142027-e52380921608 // indirect
github.com/grafana/regexp v0.0.0-20240518133315-a468a5bfb3bc // indirect
github.com/grafana/sqlds/v4 v4.1.3 // indirect
github.com/grpc-ecosystem/go-grpc-middleware v1.4.0 // indirect
github.com/grpc-ecosystem/go-grpc-middleware v1.4.0 // @grafana/grafana-search-and-storage
github.com/grpc-ecosystem/go-grpc-prometheus v1.2.1-0.20191002090509-6af20e3a5340 // indirect
github.com/hashicorp/consul/api v1.30.0 // indirect
github.com/hashicorp/errwrap v1.1.0 // indirect
+16
View File
@@ -4,6 +4,7 @@ import (
"context"
"fmt"
"path/filepath"
"time"
otgrpc "github.com/opentracing-contrib/go-grpc"
"github.com/opentracing/opentracing-go"
@@ -47,6 +48,7 @@ type Options struct {
type clientMetrics struct {
requestDuration *prometheus.HistogramVec
requestRetries *prometheus.CounterVec
}
// This adds a UnifiedStorage client into the wire dependency tree
@@ -171,11 +173,25 @@ func GrpcConn(address string, reg prometheus.Registerer) (*grpc.ClientConn, erro
Help: "Time spent executing requests to the resource server.",
Buckets: prometheus.ExponentialBuckets(0.008, 4, 7),
}, []string{"operation", "status_code"}),
requestRetries: 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"}),
}
// Report gRPC status code errors as labels.
unary, stream := instrument(metrics.requestDuration, middleware.ReportGRPCStatusOption)
// Add middleware to retry on transient connection issues. Note that
// we do not implement it for streams, as we don't currently use streams.
retryCfg := retryConfig{
Max: 3,
Backoff: time.Second,
BackoffJitter: 0.5,
}
unary = append(unary, unaryRetryInterceptor(retryCfg))
unary = append(unary, unaryRetryInstrument(metrics.requestRetries))
cfg := grpcclient.Config{}
// Set the defaults that are normally set by Config.RegisterFlags.
flagext.DefaultValues(&cfg)
+46
View File
@@ -0,0 +1,46 @@
package unified
import (
"context"
"strconv"
"time"
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"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
)
type retryConfig struct {
Max uint
Backoff time.Duration
BackoffJitter float64
}
// unaryRetryInterceptor creates an interceptor to perform retries for unary methods.
//
// Note: Retry codes are the same as the default codes.
//
// From go-grpc-middleware/interceptors/retry/options.go:
// `ResourceExhausted` means that the user quota, e.g. per-RPC limits, have been reached.
// `Unavailable` means that system is currently unavailable and the client should retry again.
func unaryRetryInterceptor(cfg retryConfig) grpc.UnaryClientInterceptor {
return grpc_retry.UnaryClientInterceptor(
grpc_retry.WithMax(cfg.Max),
grpc_retry.WithBackoff(grpc_retry.BackoffExponentialWithJitter(cfg.Backoff, cfg.BackoffJitter)),
grpc_retry.WithCodes(codes.ResourceExhausted, codes.Unavailable),
)
}
// unaryRetryInstrument creates an interceptor to count and log retry attempts.
func unaryRetryInstrument(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...)
}
}