Unify with common gRPC client options.

Signed-off-by: Tom Wilkie <[email protected]>

Author: tomwilkie

pkg/chunk/gcp/bigtable_index_client.go

@@ -65,7 +65,9 @@ type storageClientV1 struct {
 // NewStorageClientV1 returns a new v1 StorageClient.
 func NewStorageClientV1(ctx context.Context, cfg Config, schemaCfg chunk.SchemaConfig) (chunk.IndexClient, error) {
 	opts := instrumentation()
-	opts = append(opts, option.WithGRPCDialOption(cfg.GRPCClientConfig.DialOption()))
+	for _, dialOption := range cfg.GRPCClientConfig.DialOptions() {
+		opts = append(opts, option.WithGRPCDialOption(dialOption))
+	}
 
 	client, err := bigtable.NewClient(ctx, cfg.Project, cfg.Instance, opts...)
 	if err != nil {

pkg/ingester/client/client.go

@@ -50,9 +50,8 @@ func MakeIngesterClient(addr string, cfg Config) (HealthAndIngesterClient, error
 			middleware.StreamClientUserHeaderInterceptor,
 			cortex_middleware.PrometheusGRPCStreamInstrumentation(ingesterClientRequestDuration),
 		)),
-		cfg.GRPCClientConfig.DialOption(),
-		grpc.WithBalancer(grpc.RoundRobin(grpcclient.NewPoolResolver(10))),
 	}
+	opts = append(opts, cfg.GRPCClientConfig.DialOptions()...)
 	conn, err := grpc.Dial(addr, opts...)
 	if err != nil {
 		return nil, err

pkg/querier/frontend/worker.go

@@ -212,14 +212,14 @@ func (w *worker) process(ctx context.Context, c Frontend_ProcessClient) error {
 }
 
 func (w *worker) connect(address string) (FrontendClient, error) {
-	conn, err := grpc.Dial(
-		address,
+	opts := []grpc.DialOption{
 		grpc.WithInsecure(),
 		grpc.WithUnaryInterceptor(grpc_middleware.ChainUnaryClient(
 			middleware.ClientUserHeaderInterceptor,
 		)),
-		w.cfg.GRPCClientConfig.DialOption(),
-	)
+	}
+	opts = append(opts, w.cfg.GRPCClientConfig.DialOptions()...)
+	conn, err := grpc.Dial(address, opts...)
 	if err != nil {
 		return nil, err
 	}

pkg/util/grpcclient/grpcclient.go

@@ -11,13 +11,15 @@ type Config struct {
 	MaxRecvMsgSize     int  `yaml:"max_recv_msg_size"`
 	MaxSendMsgSize     int  `yaml:"max_send_msg_size"`
 	UseGzipCompression bool `yaml:"use_gzip_compression"`
+	ConnectionPoolSize int  `yaml:"connect_pool_size"`
 }
 
 // RegisterFlags registers flags.
 func (cfg *Config) RegisterFlags(prefix string, f *flag.FlagSet) {
 	f.IntVar(&cfg.MaxRecvMsgSize, prefix+".grpc-max-recv-msg-size", 100<<20, "gRPC client max receive message size (bytes).")
 	f.IntVar(&cfg.MaxSendMsgSize, prefix+".grpc-max-send-msg-size", 16<<20, "gRPC client max send message size (bytes).")
 	f.BoolVar(&cfg.UseGzipCompression, prefix+".grpc-use-gzip-compression", false, "Use compression when sending messages.")
+	f.IntVar(&cfg.ConnectionPoolSize, prefix+".grpc-connection-pool-size", 1, "Number of connections to keep active per endpoint.")
 }
 
 // CallOptions returns the config in terms of CallOptions.
@@ -31,7 +33,10 @@ func (cfg *Config) CallOptions() []grpc.CallOption {
 	return opts
 }
 
-// DialOption returns the config as a grpc.DialOptions.
-func (cfg *Config) DialOption() grpc.DialOption {
-	return grpc.WithDefaultCallOptions(cfg.CallOptions()...)
+// DialOptions returns the config as a grpc.DialOptions.
+func (cfg *Config) DialOptions() []grpc.DialOption {
+	return []grpc.DialOption{
+		grpc.WithDefaultCallOptions(cfg.CallOptions()...),
+		grpc.WithBalancer(grpc.RoundRobin(NewPoolResolver(cfg.ConnectionPoolSize))),
+	}
 }