diff --git a/pkg/services/apiserver/options/storage.go b/pkg/services/apiserver/options/storage.go index e6979e05ee1..97b95a45b26 100644 --- a/pkg/services/apiserver/options/storage.go +++ b/pkg/services/apiserver/options/storage.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "net" + "time" "github.com/spf13/pflag" "go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc" @@ -52,6 +53,7 @@ type StorageOptions struct { GrpcClientAuthenticationTokenExchangeURL string GrpcClientAuthenticationTokenNamespace string GrpcClientAuthenticationAllowInsecure bool + GrpcClientKeepaliveTime time.Duration // Secrets Manager Configuration for InlineSecureValueSupport SecretsManagerGrpcClientEnable bool @@ -90,6 +92,7 @@ func NewStorageOptions() *StorageOptions { Address: "localhost:10000", GrpcClientAuthenticationTokenNamespace: "*", GrpcClientAuthenticationAllowInsecure: false, + GrpcClientKeepaliveTime: 0, BlobThresholdBytes: BlobThresholdDefault, } } @@ -103,6 +106,7 @@ func (o *StorageOptions) AddFlags(fs *pflag.FlagSet) { fs.StringVar(&o.GrpcClientAuthenticationTokenExchangeURL, "grpc-client-authentication-token-exchange-url", o.GrpcClientAuthenticationTokenExchangeURL, "Token exchange url for grpc client authentication") fs.StringVar(&o.GrpcClientAuthenticationTokenNamespace, "grpc-client-authentication-token-namespace", o.GrpcClientAuthenticationTokenNamespace, "Token namespace for grpc client authentication") fs.BoolVar(&o.GrpcClientAuthenticationAllowInsecure, "grpc-client-authentication-allow-insecure", o.GrpcClientAuthenticationAllowInsecure, "Allow insecure grpc client authentication") + fs.DurationVar(&o.GrpcClientKeepaliveTime, "grpc-client-keepalive-time", o.GrpcClientKeepaliveTime, "gRPC client keep-alive ping interval (e.g., 6m).") // Secrets Manager Configuration flags fs.BoolVar(&o.SecretsManagerGrpcClientEnable, "grafana.secrets-manager.grpc-client-enable", false, "Enable gRPC client for secrets manager") diff --git a/pkg/storage/unified/client.go b/pkg/storage/unified/client.go index 03cab8c6df4..be6efdd0b37 100644 --- a/pkg/storage/unified/client.go +++ b/pkg/storage/unified/client.go @@ -14,6 +14,7 @@ import ( "gocloud.dev/blob/fileblob" "google.golang.org/grpc" "google.golang.org/grpc/credentials/insecure" + "google.golang.org/grpc/keepalive" "github.com/grafana/authlib/types" "github.com/grafana/dskit/flagext" @@ -59,12 +60,13 @@ func ProvideUnifiedStorageClient(opts *Options, // See: apiserver.applyAPIServerConfig(cfg, features, o) apiserverCfg := opts.Cfg.SectionWithEnvOverrides("grafana-apiserver") client, err := newClient(options.StorageOptions{ - StorageType: options.StorageType(apiserverCfg.Key("storage_type").MustString(string(options.StorageTypeUnified))), - DataPath: apiserverCfg.Key("storage_path").MustString(filepath.Join(opts.Cfg.DataPath, "grafana-apiserver")), - Address: apiserverCfg.Key("address").MustString(""), - SearchServerAddress: apiserverCfg.Key("search_server_address").MustString(""), - BlobStoreURL: apiserverCfg.Key("blob_url").MustString(""), - BlobThresholdBytes: apiserverCfg.Key("blob_threshold_bytes").MustInt(options.BlobThresholdDefault), + StorageType: options.StorageType(apiserverCfg.Key("storage_type").MustString(string(options.StorageTypeUnified))), + DataPath: apiserverCfg.Key("storage_path").MustString(filepath.Join(opts.Cfg.DataPath, "grafana-apiserver")), + Address: apiserverCfg.Key("address").MustString(""), + SearchServerAddress: apiserverCfg.Key("search_server_address").MustString(""), + BlobStoreURL: apiserverCfg.Key("blob_url").MustString(""), + BlobThresholdBytes: apiserverCfg.Key("blob_threshold_bytes").MustInt(options.BlobThresholdDefault), + GrpcClientKeepaliveTime: apiserverCfg.Key("grpc_client_keepalive_time").MustDuration(0), }, opts.Cfg, opts.Features, opts.DB, opts.Tracer, opts.Reg, opts.Authzc, opts.Docs, storageMetrics, indexMetrics, opts.SecureValues) if err == nil { // Decide whether to disable SQL fallback stats per resource in Mode 5. @@ -145,13 +147,13 @@ func newClient(opts options.StorageOptions, metrics = newClientMetrics(reg) ) - conn, err = newGrpcConn(opts.Address, metrics, features) + conn, err = newGrpcConn(opts.Address, metrics, features, opts.GrpcClientKeepaliveTime) if err != nil { return nil, err } if opts.SearchServerAddress != "" { - indexConn, err = newGrpcConn(opts.SearchServerAddress, metrics, features) + indexConn, err = newGrpcConn(opts.SearchServerAddress, metrics, features, opts.GrpcClientKeepaliveTime) if err != nil { return nil, err @@ -219,7 +221,7 @@ func newClient(opts options.StorageOptions, } } -func newGrpcConn(address string, metrics *clientMetrics, features featuremgmt.FeatureToggles) (grpc.ClientConnInterface, error) { +func newGrpcConn(address string, metrics *clientMetrics, features featuremgmt.FeatureToggles, clientKeepaliveTime time.Duration) (grpc.ClientConnInterface, error) { // Create either a connection pool or a single connection. // The connection pool __can__ be useful when connection to // server side load balancers like kube-proxy. @@ -229,7 +231,7 @@ func newGrpcConn(address string, metrics *clientMetrics, features featuremgmt.Fe maxCapacity: 6, idleTimeout: time.Minute, factory: func() (*grpc.ClientConn, error) { - return grpcConn(address, metrics) + return grpcConn(address, metrics, clientKeepaliveTime) }, }) if err != nil { @@ -239,7 +241,7 @@ func newGrpcConn(address string, metrics *clientMetrics, features featuremgmt.Fe return conn, nil } - conn, err := grpcConn(address, metrics) + conn, err := grpcConn(address, metrics, clientKeepaliveTime) if err != nil { return nil, err } @@ -248,7 +250,7 @@ func newGrpcConn(address string, metrics *clientMetrics, features featuremgmt.Fe } // grpcConn creates a new gRPC connection to the provided address. -func grpcConn(address string, metrics *clientMetrics) (*grpc.ClientConn, error) { +func grpcConn(address string, metrics *clientMetrics, clientKeepaliveTime time.Duration) (*grpc.ClientConn, error) { // Report gRPC status code errors as labels. unary, stream := instrument(metrics.requestDuration, middleware.ReportGRPCStatusOption) @@ -281,14 +283,22 @@ func grpcConn(address string, metrics *clientMetrics) (*grpc.ClientConn, error) // This reduces the number of requests made to the DNS servers. opts = append(opts, grpc.WithDisableServiceConfig()) + if clientKeepaliveTime > 0 { + opts = append(opts, grpc.WithKeepaliveParams(keepalive.ClientParameters{ + Time: clientKeepaliveTime, + Timeout: 10 * time.Second, + PermitWithoutStream: true, + })) + } // Create a connection to the gRPC server return grpc.NewClient(address, opts...) } // GrpcConn is the public constructor that can be used for testing. +// TODO: also use grpc_client_keepalive_time here. func GrpcConn(address string, reg prometheus.Registerer) (*grpc.ClientConn, error) { metrics := newClientMetrics(reg) - return grpcConn(address, metrics) + return grpcConn(address, metrics, 0) } // instrument is the same as grpcclient.Instrument but without the middleware.ClientUserHeaderInterceptor