config: simplify grpc options

The endpoint group service config is genuienly useful for all grpc
connections. The difference is in the resolver - for groups it will
resolve on its own and loadbalnace between the backends for non-groups
we use the passthrough resolver which then will loadbalnace between one
backend which should be same behavior as today - in turn we get retries
on UNAVAILABLE which is still useful.
This also unifies the dial option construction which makes it easier to
reason about.
I also took the liberty to clean up some related stuff.

Signed-off-by: Michael Hoffmann <mhoffmann@cloudflare.com>

Author: MichaHoffmann

cmd/thanos/config.go

@@ -70,6 +70,7 @@ type grpcClientConfig struct {
 	cert, key, caCert string
 	serverName        string
 	compression       string
+	serviceConfig     string
 }
 
 func (gc *grpcClientConfig) registerFlag(cmd extkingpin.FlagClause) *grpcClientConfig {
@@ -81,19 +82,34 @@ func (gc *grpcClientConfig) registerFlag(cmd extkingpin.FlagClause) *grpcClientC
 	cmd.Flag("grpc-client-server-name", "Server name to verify the hostname on the returned gRPC certificates. See https://tools.ietf.org/html/rfc4366#section-3.1").Default("").StringVar(&gc.serverName)
 	compressionOptions := strings.Join([]string{snappy.Name, compressionNone}, ", ")
 	cmd.Flag("grpc-compression", "Compression algorithm to use for gRPC requests to other clients. Must be one of: "+compressionOptions).Default(compressionNone).EnumVar(&gc.compression, snappy.Name, compressionNone)
+	cmd.Flag("grpc-service-config", "gRPC service configuration in JSON format. See https://github.com/grpc/grpc/blob/master/doc/service_config.md").Default("").StringVar(&gc.serviceConfig)
+
+	return gc
+}
+
+func (gc *grpcClientConfig) registerReceiverFlag(cmd extkingpin.FlagClause) *grpcClientConfig {
+	cmd.Flag("remote-write.client-tls-secure", "Use TLS when talking to the other receivers.").Default("false").BoolVar(&gc.secure)
+	cmd.Flag("remote-write.client-tls-skip-verify", "Disable TLS certificate verification when talking to the other receivers i.e self signed, signed by fake CA.").Default("false").BoolVar(&gc.skipVerify)
+	cmd.Flag("remote-write.client-tls-cert", "TLS Certificates to use to identify this client to the server.").Default("").StringVar(&gc.cert)
+	cmd.Flag("remote-write.client-tls-key", "TLS Key for the client's certificate.").Default("").StringVar(&gc.key)
+	cmd.Flag("remote-write.client-tls-ca", "TLS CA Certificates to use to verify servers.").Default("").StringVar(&gc.caCert)
+	cmd.Flag("remote-write.client-server-name", "Server name to verify the hostname on the returned TLS certificates. See https://tools.ietf.org/html/rfc4366#section-3.1").Default("").StringVar(&gc.serverName)
+	compressionOptions := strings.Join([]string{snappy.Name, compressionNone}, ", ")
+	cmd.Flag("receive.grpc-compression", "Compression algorithm to use for gRPC requests to other receivers. Must be one of: "+compressionOptions).Default(snappy.Name).EnumVar(&gc.compression, snappy.Name, compressionNone)
+	cmd.Flag("receive.grpc-service-config", "gRPC service configuration file or content in JSON format. See https://github.com/grpc/grpc/blob/master/doc/service_config.md").PlaceHolder("<content>").Default("").StringVar(&gc.serviceConfig)
 
 	return gc
 }
 
 func (gc *grpcClientConfig) dialOptions(logger log.Logger, reg prometheus.Registerer, tracer opentracing.Tracer) ([]grpc.DialOption, error) {
-	dialOpts, err := extgrpc.StoreClientGRPCOpts(logger, reg, tracer, gc.secure, gc.skipVerify, gc.cert, gc.key, gc.caCert, gc.serverName)
-	if err != nil {
-		return nil, errors.Wrapf(err, "building gRPC client")
+	opts := []extgrpc.StoreClientGRPCOption{
+		extgrpc.WithCompression(gc.compression),
+		extgrpc.WithServiceConfig(gc.serviceConfig),
 	}
-	if gc.compression != compressionNone {
-		dialOpts = append(dialOpts, grpc.WithDefaultCallOptions(grpc.UseCompressor(gc.compression)))
+	if gc.secure {
+		opts = append(opts, extgrpc.WithTLS(gc.cert, gc.key, gc.caCert, gc.serverName, gc.skipVerify))
 	}
-	return dialOpts, nil
+	return extgrpc.StoreClientGRPCOpts(logger, reg, tracer, opts...)
 }
 
 type httpConfig struct {

cmd/thanos/endpointset.go

@@ -26,7 +26,6 @@ import (
 	"github.com/thanos-io/thanos/pkg/discovery/cache"
 	"github.com/thanos-io/thanos/pkg/discovery/dns"
 	"github.com/thanos-io/thanos/pkg/errors"
-	"github.com/thanos-io/thanos/pkg/extgrpc"
 	"github.com/thanos-io/thanos/pkg/extkingpin"
 	"github.com/thanos-io/thanos/pkg/extprom"
 	"github.com/thanos-io/thanos/pkg/logutil"
@@ -116,9 +115,6 @@ func validateEndpointConfig(cfg EndpointConfig) error {
 		if dns.IsDynamicNode(ecfg.Address) && ecfg.Strict {
 			return errors.Newf("%s is a dynamically specified endpoint i.e. it uses SD and that is not permitted under strict mode.", ecfg.Address)
 		}
-		if !ecfg.Group && len(ecfg.ServiceConfig) != 0 {
-			return errors.Newf("%s service_config is only valid for endpoint groups.", ecfg.Address)
-		}
 	}
 	return nil
 }
@@ -354,15 +350,19 @@ func setupEndpointSet(
 		// groups and non dynamic endpoints
 		for _, ecfg := range endpointConfig.Endpoints {
 			strict, group, addr := ecfg.Strict, ecfg.Group, ecfg.Address
+			opts := dialOpts
+			if ecfg.ServiceConfig != "" {
+				opts = append(append(make([]grpc.DialOption, 0, len(dialOpts)+1), dialOpts...), grpc.WithDefaultServiceConfig(ecfg.ServiceConfig))
+			}
 			if group {
-				specs = append(specs, query.NewGRPCEndpointSpec(fmt.Sprintf("thanos:///%s", addr), strict, append(dialOpts, extgrpc.EndpointGroupGRPCOpts(ecfg.ServiceConfig)...)...))
+				specs = append(specs, query.NewGRPCEndpointSpec(fmt.Sprintf("thanos:///%s", addr), strict, opts...))
 			} else if !dns.IsDynamicNode(addr) {
-				specs = append(specs, query.NewGRPCEndpointSpec(addr, strict, dialOpts...))
+				specs = append(specs, query.NewGRPCEndpointSpec(fmt.Sprintf("passthrough:///%s", addr), strict, opts...))
 			}
 		}
 		// dynamic endpoints
 		for _, addr := range dnsEndpointProvider.Addresses() {
-			specs = append(specs, query.NewGRPCEndpointSpec(addr, false, dialOpts...))
+			specs = append(specs, query.NewGRPCEndpointSpec(fmt.Sprintf("passthrough:///%s", addr), false, dialOpts...))
 		}
 		return removeDuplicateEndpointSpecs(specs)
 	}, unhealthyTimeout, endpointTimeout, queryTimeout, queryConnMetricLabels...)

cmd/thanos/receive.go

@@ -30,15 +30,12 @@ import (
 	"github.com/thanos-io/objstore"
 	"github.com/thanos-io/objstore/client"
 	objstoretracing "github.com/thanos-io/objstore/tracing/opentracing"
-	"google.golang.org/grpc"
 	"gopkg.in/yaml.v2"
 
 	"github.com/thanos-io/thanos/pkg/block/metadata"
 	"github.com/thanos-io/thanos/pkg/component"
 	"github.com/thanos-io/thanos/pkg/compressutil"
 	"github.com/thanos-io/thanos/pkg/exemplars"
-	"github.com/thanos-io/thanos/pkg/extgrpc"
-	"github.com/thanos-io/thanos/pkg/extgrpc/snappy"
 	"github.com/thanos-io/thanos/pkg/extkingpin"
 	"github.com/thanos-io/thanos/pkg/extprom"
 	"github.com/thanos-io/thanos/pkg/info"
@@ -158,27 +155,10 @@ func runReceive(
 		return err
 	}
 
-	dialOpts, err := extgrpc.StoreClientGRPCOpts(
-		logger,
-		reg,
-		tracer,
-		conf.rwClientSecure,
-		conf.rwClientSkipVerify,
-		conf.rwClientCert,
-		conf.rwClientKey,
-		conf.rwClientServerCA,
-		conf.rwClientServerName,
-	)
+	dialOpts, err := conf.rwClientConfig.dialOptions(logger, reg, tracer)
 	if err != nil {
 		return err
 	}
-	if conf.compression != compressionNone {
-		dialOpts = append(dialOpts, grpc.WithDefaultCallOptions(grpc.UseCompressor(conf.compression)))
-	}
-
-	if conf.grpcServiceConfig != "" {
-		dialOpts = append(dialOpts, grpc.WithDefaultServiceConfig(conf.grpcServiceConfig))
-	}
 
 	var bkt objstore.Bucket
 	confContentYaml, err := conf.objStoreConfig.Content()
@@ -844,12 +824,7 @@ type receiveConfig struct {
 	rwServerCert          string
 	rwServerKey           string
 	rwServerClientCA      string
-	rwClientCert          string
-	rwClientKey           string
-	rwClientSecure        bool
-	rwClientServerCA      string
-	rwClientServerName    string
-	rwClientSkipVerify    bool
+	rwClientConfig        grpcClientConfig
 	rwServerTlsMinVersion string
 
 	dataDir   string
@@ -873,9 +848,7 @@ type receiveConfig struct {
 	forwardTimeout      *model.Duration
 	maxBackoff          *model.Duration
 	maxArtificialDelay  *model.Duration
-	compression         string
 	replicationProtocol string
-	grpcServiceConfig   string
 
 	tsdbMinBlockDuration         *model.Duration
 	tsdbMaxBlockDuration         *model.Duration
@@ -937,17 +910,7 @@ func (rc *receiveConfig) registerFlag(cmd extkingpin.FlagClause) {
 
 	cmd.Flag("remote-write.server-tls-min-version", "TLS version for the gRPC server, leave blank to default to TLS 1.3, allow values: [\"1.0\", \"1.1\", \"1.2\", \"1.3\"]").Default("1.3").StringVar(&rc.rwServerTlsMinVersion)
 
-	cmd.Flag("remote-write.client-tls-cert", "TLS Certificates to use to identify this client to the server.").Default("").StringVar(&rc.rwClientCert)
-
-	cmd.Flag("remote-write.client-tls-key", "TLS Key for the client's certificate.").Default("").StringVar(&rc.rwClientKey)
-
-	cmd.Flag("remote-write.client-tls-secure", "Use TLS when talking to the other receivers.").Default("false").BoolVar(&rc.rwClientSecure)
-
-	cmd.Flag("remote-write.client-tls-skip-verify", "Disable TLS certificate verification when talking to the other receivers i.e self signed, signed by fake CA.").Default("false").BoolVar(&rc.rwClientSkipVerify)
-
-	cmd.Flag("remote-write.client-tls-ca", "TLS CA Certificates to use to verify servers.").Default("").StringVar(&rc.rwClientServerCA)
-
-	cmd.Flag("remote-write.client-server-name", "Server name to verify the hostname on the returned TLS certificates. See https://tools.ietf.org/html/rfc4366#section-3.1").Default("").StringVar(&rc.rwClientServerName)
+	rc.rwClientConfig.registerReceiverFlag(cmd)
 
 	cmd.Flag("tsdb.path", "Data directory of TSDB.").
 		Default("./data").StringVar(&rc.dataDir)
@@ -987,8 +950,6 @@ func (rc *receiveConfig) registerFlag(cmd extkingpin.FlagClause) {
 	cmd.Flag("receive.replica-header", "HTTP header specifying the replica number of a write request.").Default(receive.DefaultReplicaHeader).StringVar(&rc.replicaHeader)
 
 	cmd.Flag("receive.forward.async-workers", "Number of concurrent workers processing forwarding of remote-write requests.").Default("5").UintVar(&rc.asyncForwardWorkerCount)
-	compressionOptions := strings.Join([]string{snappy.Name, compressionNone}, ", ")
-	cmd.Flag("receive.grpc-compression", "Compression algorithm to use for gRPC requests to other receivers. Must be one of: "+compressionOptions).Default(snappy.Name).EnumVar(&rc.compression, snappy.Name, compressionNone)
 
 	cmd.Flag("receive.replication-factor", "How many times to replicate incoming write requests.").Default("1").Uint64Var(&rc.replicationFactor)
 
@@ -999,8 +960,6 @@ func (rc *receiveConfig) registerFlag(cmd extkingpin.FlagClause) {
 
 	cmd.Flag("receive.capnproto-address", "Address for the Cap'n Proto server.").Default(fmt.Sprintf("0.0.0.0:%s", receive.DefaultCapNProtoPort)).StringVar(&rc.replicationAddr)
 
-	cmd.Flag("receive.grpc-service-config", "gRPC service configuration file or content in JSON format. See https://github.com/grpc/grpc/blob/master/doc/service_config.md").PlaceHolder("<content>").Default("").StringVar(&rc.grpcServiceConfig)
-
 	rc.forwardTimeout = extkingpin.ModelDuration(cmd.Flag("receive-forward-timeout", "Timeout for each forward request.").Default("5s").Hidden())
 
 	rc.maxBackoff = extkingpin.ModelDuration(cmd.Flag("receive-forward-max-backoff", "Maximum backoff for each forward fan-out request").Default("5s").Hidden())

cmd/thanos/rule.go

@@ -417,17 +417,7 @@ func runRule(
 	}
 
 	if len(grpcEndpoints) > 0 {
-		dialOpts, err := extgrpc.StoreClientGRPCOpts(
-			logger,
-			reg,
-			tracer,
-			false,
-			false,
-			"",
-			"",
-			"",
-			"",
-		)
+		dialOpts, err := extgrpc.StoreClientGRPCOpts(logger, reg, tracer)
 		if err != nil {
 			return err
 		}

pkg/extgrpc/client.go

@@ -78,33 +78,81 @@ func (c *nonPoolingCodec) Marshal(v any) (mem.BufferSlice, error) {
 	return mem.BufferSlice{mem.NewBuffer(&bufExact, pool)}, nil
 }
 
-// EndpointGroupGRPCOpts creates gRPC dial options for connecting to endpoint groups.
+// DefaultServiceConfig is the default gRPC service config applied to all client connections.
+// It enables round-robin load balancing and retries on transient UNAVAILABLE errors.
 // For details on retry capabilities, see https://github.com/grpc/proposal/blob/master/A6-client-retries.md#retry-policy-capabilities
-func EndpointGroupGRPCOpts(serviceConfig string) []grpc.DialOption {
-	if serviceConfig == "" {
-		serviceConfig = `
-{
+const DefaultServiceConfig = `{
   "loadBalancingPolicy":"round_robin",
   "retryPolicy": {
     "maxAttempts": 3,
     "initialBackoff": "0.1s",
     "backoffMultiplier": 2,
     "retryableStatusCodes": [
-  	  "UNAVAILABLE"
+      "UNAVAILABLE"
     ]
   }
 }`
+
+// StoreClientGRPCOption is a functional option for StoreClientGRPCOpts.
+type StoreClientGRPCOption func(*storeClientGRPCOpts)
+
+type storeClientGRPCOpts struct {
+	serviceConfig string
+	secure        bool
+	skipVerify    bool
+	cert          string
+	key           string
+	caCert        string
+	serverName    string
+	compression   string
+}
+
+// WithTLS enables TLS for the gRPC client connection.
+// Presence of this option implies secure=true.
+// Empty cert/key/caCert/serverName values are valid and result in TLS with system CA pool.
+func WithTLS(cert, key, caCert, serverName string, skipVerify bool) StoreClientGRPCOption {
+	return func(o *storeClientGRPCOpts) {
+		o.secure = true
+		o.cert = cert
+		o.key = key
+		o.caCert = caCert
+		o.serverName = serverName
+		o.skipVerify = skipVerify
 	}
+}
 
-	return []grpc.DialOption{
-		grpc.WithDefaultServiceConfig(serviceConfig),
-		grpc.WithDisableServiceConfig(),
-		grpc.WithKeepaliveParams(keepalive.ClientParameters{Time: 10 * time.Second, Timeout: 5 * time.Second}),
+// WithServiceConfig overrides the default gRPC service config. No-op if empty.
+func WithServiceConfig(serviceConfig string) StoreClientGRPCOption {
+	return func(o *storeClientGRPCOpts) {
+		if serviceConfig != "" {
+			o.serviceConfig = serviceConfig
+		}
+	}
+}
+
+// WithCompression enables gRPC compression with the given algorithm. No-op if empty or "none".
+func WithCompression(compression string) StoreClientGRPCOption {
+	return func(o *storeClientGRPCOpts) {
+		if compression != "" && compression != "none" {
+			o.compression = compression
+		}
 	}
 }
 
 // StoreClientGRPCOpts creates gRPC dial options for connecting to a store client.
-func StoreClientGRPCOpts(logger log.Logger, reg prometheus.Registerer, tracer opentracing.Tracer, secure, skipVerify bool, cert, key, caCert, serverName string) ([]grpc.DialOption, error) {
+// By default, DefaultServiceConfig is applied and connections are insecure.
+// Use WithTLS, WithServiceConfig, and WithCompression to customize.
+func StoreClientGRPCOpts(logger log.Logger, reg prometheus.Registerer, tracer opentracing.Tracer, options ...StoreClientGRPCOption) ([]grpc.DialOption, error) {
+	o := &storeClientGRPCOpts{}
+	for _, opt := range options {
+		opt(o)
+	}
+
+	serviceConfig := o.serviceConfig
+	if serviceConfig == "" {
+		serviceConfig = DefaultServiceConfig
+	}
+
 	grpcMets := grpc_prometheus.NewClientMetrics(
 		grpc_prometheus.WithClientHandlingTimeHistogram(grpc_prometheus.WithHistogramOpts(
 			&prometheus.HistogramOpts{
@@ -131,18 +179,23 @@ func StoreClientGRPCOpts(logger log.Logger, reg prometheus.Registerer, tracer op
 			tracing.StreamClientInterceptor(tracer),
 		),
 		grpc.WithKeepaliveParams(keepalive.ClientParameters{Time: 10 * time.Second, Timeout: 5 * time.Second}),
+		grpc.WithDefaultServiceConfig(serviceConfig),
+		grpc.WithDisableServiceConfig(),
+	}
+	if o.compression != "" {
+		dialOpts = append(dialOpts, grpc.WithDefaultCallOptions(grpc.UseCompressor(o.compression)))
 	}
 	if reg != nil {
 		reg.MustRegister(grpcMets)
 	}
 
-	if !secure {
+	if !o.secure {
 		return append(dialOpts, grpc.WithTransportCredentials(insecure.NewCredentials())), nil
 	}
 
 	level.Info(logger).Log("msg", "enabling client to server TLS")
 
-	tlsCfg, err := tls.NewClientConfig(logger, cert, key, caCert, serverName, skipVerify)
+	tlsCfg, err := tls.NewClientConfig(logger, o.cert, o.key, o.caCert, o.serverName, o.skipVerify)
 	if err != nil {
 		return nil, err
 	}