Query: Add servername to grpc dial options for DNS stores (Fixes #1507)

CHANGELOG.md

@@ -11,6 +11,8 @@ We use *breaking* word for marking changes that are not backward compatible (rel
 
 ## Unreleased
 
+- [#2407](https://github.com/thanos-io/thanos/pull/2407) Query: Add servername to gRPC dial options for DNS stores (Fixes #1507)
+
 ### Fixed
 
 - [#2288](https://github.com/thanos-io/thanos/pull/2288) Ruler: Fixes issue #2281 bug in ruler with parsing query addr with path prefix

cmd/thanos/query.go

@@ -56,6 +56,7 @@ func registerQuery(m map[string]setupFunc, app *kingpin.Application) {
 	key := cmd.Flag("grpc-client-tls-key", "TLS Key for the client's certificate").Default("").String()
 	caCert := cmd.Flag("grpc-client-tls-ca", "TLS CA Certificates to use to verify gRPC servers").Default("").String()
 	serverName := 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("").String()
+	dnsServerName := cmd.Flag("grpc-client-dns-server-name", "For stores that are DNS, use the dns name as the server name for connection. Only applicable if using the secure flag").Default("false").Bool()
 
 	webRoutePrefix := cmd.Flag("web.route-prefix", "Prefix for API and UI endpoints. This allows thanos UI to be served on a sub-path. This option is analogous to --web.route-prefix of Promethus.").Default("").String()
 	webExternalPrefix := cmd.Flag("web.external-prefix", "Static prefix for all HTML links and redirect URLs in the UI query web interface. Actual endpoints are still served on / or the web.route-prefix. This allows thanos UI to be served behind a reverse proxy that strips a URL sub-path.").Default("").String()
@@ -107,6 +108,10 @@ func registerQuery(m map[string]setupFunc, app *kingpin.Application) {
 	storeResponseTimeout := modelDuration(cmd.Flag("store.response-timeout", "If a Store doesn't send any data in this specified duration then a Store will be ignored and partial data will be returned if it's enabled. 0 disables timeout.").Default("0ms"))
 
 	m[comp.String()] = func(g *run.Group, logger log.Logger, reg *prometheus.Registry, tracer opentracing.Tracer, _ <-chan struct{}, _ bool) error {
+		if *dnsServerName && !*secure {
+			return errors.New("grpc-client-dns-server-name specified without grpc-client-tls-secure")
+		}
+
 		selectorLset, err := parseFlagLabels(*selectorLabels)
 		if err != nil {
 			return errors.Wrap(err, "parse federation labels")
@@ -147,6 +152,7 @@ func registerQuery(m map[string]setupFunc, app *kingpin.Application) {
 			*key,
 			*caCert,
 			*serverName,
+			*dnsServerName,
 			*httpBindAddr,
 			time.Duration(*httpGracePeriod),
 			*webRoutePrefix,
@@ -188,6 +194,7 @@ func runQuery(
 	key string,
 	caCert string,
 	serverName string,
+	dnsServerName bool,
 	httpBindAddr string,
 	httpGracePeriod time.Duration,
 	webRoutePrefix string,
@@ -240,7 +247,11 @@ func runQuery(
 			func() (specs []query.StoreSpec) {
 				// Add DNS resolved addresses.
 				for _, addr := range dnsProvider.Addresses() {
-					specs = append(specs, query.NewGRPCStoreSpec(addr, false))
+					if dnsServerName {
+						specs = append(specs, query.NewGRPCStoreSpecServerName(addr, false, dnsProvider.ServerName(addr)))
+					} else {
+						specs = append(specs, query.NewGRPCStoreSpec(addr, false))
+					}
 				}
 				// Add strict & static nodes.
 				for _, addr := range strictStores {
@@ -253,6 +264,10 @@ func runQuery(
 			},
 			dialOpts,
 			unhealthyStoreTimeout,
+			dnsServerName,
+			cert,
+			key,
+			caCert,
 		)
 		proxy            = store.NewProxyStore(logger, reg, stores.Get, component.Query, selectorLset, storeResponseTimeout)
 		queryableCreator = query.NewQueryableCreator(logger, proxy)

docs/components/query.md

@@ -288,6 +288,10 @@ Flags:
                                  Server name to verify the hostname on the
                                  returned gRPC certificates. See
                                  https://tools.ietf.org/html/rfc4366#section-3.1
+      --grpc-client-dns-server-name
+                                 For stores that are DNS, use the dns name as
+                                 the server name for connection. Only applicable
+                                 if using the secure flag
       --web.route-prefix=""      Prefix for API and UI endpoints. This allows
                                  thanos UI to be served on a sub-path. This
                                  option is analogous to --web.route-prefix of

pkg/discovery/dns/provider.go

@@ -150,3 +150,19 @@ func (p *Provider) Addresses() []string {
 	}
 	return result
 }
+
+// ServerName returns the server name for an address.
+func (p *Provider) ServerName(addr string) string {
+	p.Lock()
+	defer p.Unlock()
+
+	for name, addrs := range p.resolved {
+		for _, a := range addrs {
+			if addr == a {
+				_, n := GetQTypeName(name)
+				return strings.SplitN(n, ":", 2)[0]
+			}
+		}
+	}
+	return ""
+}

pkg/query/storeset.go

@@ -20,7 +20,9 @@ import (
 	"github.com/thanos-io/thanos/pkg/runutil"
 	"github.com/thanos-io/thanos/pkg/store"
 	"github.com/thanos-io/thanos/pkg/store/storepb"
+	"github.com/thanos-io/thanos/pkg/tls"
 	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
 )
 
 const (
@@ -38,6 +40,8 @@ type StoreSpec interface {
 	Metadata(ctx context.Context, client storepb.StoreClient) (labelSets []storepb.LabelSet, mint int64, maxt int64, storeType component.StoreAPI, err error)
 	// StrictStatic returns true if the StoreAPI has been statically defined and it is under a strict mode.
 	StrictStatic() bool
+	// ServerName returns the StoreAPI's server name for the store spec.
+	ServerName() string
 }
 
 type StoreStatus struct {
@@ -53,12 +57,19 @@ type StoreStatus struct {
 type grpcStoreSpec struct {
 	addr         string
 	strictstatic bool
+	serverName   string
+}
+
+// NewGRPCStoreSpecServerName creates store pure gRPC spec with a server name.
+// It uses Info gRPC call to get Metadata.
+func NewGRPCStoreSpecServerName(addr string, strictstatic bool, serverName string) StoreSpec {
+	return &grpcStoreSpec{addr: addr, strictstatic: strictstatic, serverName: serverName}
 }
 
 // NewGRPCStoreSpec creates store pure gRPC spec.
 // It uses Info gRPC call to get Metadata.
 func NewGRPCStoreSpec(addr string, strictstatic bool) StoreSpec {
-	return &grpcStoreSpec{addr: addr, strictstatic: strictstatic}
+	return &grpcStoreSpec{addr: addr, strictstatic: strictstatic, serverName: ""}
 }
 
 // StrictStatic returns true if the StoreAPI has been statically defined and it is under a strict mode.
@@ -71,6 +82,10 @@ func (s *grpcStoreSpec) Addr() string {
 	return s.addr
 }
 
+func (s *grpcStoreSpec) ServerName() string {
+	return s.serverName
+}
+
 // Metadata method for gRPC store API tries to reach host Info method until context timeout. If we are unable to get metadata after
 // that time, we assume that the host is unhealthy and return error.
 func (s *grpcStoreSpec) Metadata(ctx context.Context, client storepb.StoreClient) (labelSets []storepb.LabelSet, mint int64, maxt int64, storeType component.StoreAPI, err error) {
@@ -165,6 +180,10 @@ type StoreSet struct {
 	storeSpecs          func() []StoreSpec
 	dialOpts            []grpc.DialOption
 	gRPCInfoCallTimeout time.Duration
+	dnsServerName       bool
+	cert                string
+	key                 string
+	caCert              string
 
 	updateMtx         sync.Mutex
 	storesMtx         sync.RWMutex
@@ -186,6 +205,10 @@ func NewStoreSet(
 	storeSpecs func() []StoreSpec,
 	dialOpts []grpc.DialOption,
 	unhealthyStoreTimeout time.Duration,
+	dnsServerName bool,
+	cert string,
+	key string,
+	caCert string,
 ) *StoreSet {
 	storesMetric := newStoreSetNodeCollector()
 	if reg != nil {
@@ -208,6 +231,10 @@ func NewStoreSet(
 		stores:                make(map[string]*storeRef),
 		storeStatuses:         make(map[string]*StoreStatus),
 		unhealthyStoreTimeout: unhealthyStoreTimeout,
+		dnsServerName:         dnsServerName,
+		cert:                  cert,
+		key:                   key,
+		caCert:                caCert,
 	}
 	return ss
 }
@@ -420,7 +447,17 @@ func (s *StoreSet) getActiveStores(ctx context.Context, stores map[string]*store
 			st, seenAlready := stores[addr]
 			if !seenAlready {
 				// New store or was unactive and was removed in the past - create new one.
-				conn, err := grpc.DialContext(ctx, addr, s.dialOpts...)
+				dialOpts := s.dialOpts
+				// Update the dialOpts if we are asked to add dnsServerName.
+				if s.dnsServerName {
+					tlsCfg, err := tls.NewClientConfig(s.logger, s.cert, s.key, s.caCert, spec.ServerName())
+					if err != nil {
+						level.Warn(s.logger).Log("msg", "update of store node failed", "err", errors.Wrap(err, "setting TLS"), "address", addr)
+						return
+					}
+					dialOpts = append(s.dialOpts, grpc.WithTransportCredentials(credentials.NewTLS(tlsCfg)))
+				}
+				conn, err := grpc.DialContext(ctx, addr, dialOpts...)
 				if err != nil {
 					s.updateStoreStatus(&storeRef{addr: addr}, err)
 					level.Warn(s.logger).Log("msg", "update of store node failed", "err", errors.Wrap(err, "dialing connection"), "address", addr)

pkg/query/storeset_test.go

@@ -184,7 +184,7 @@ func TestStoreSet_Update(t *testing.T) {
 			specs = append(specs, NewGRPCStoreSpec(addr, false))
 		}
 		return specs
-	}, testGRPCOpts, time.Minute)
+	}, testGRPCOpts, time.Minute, false, "", "", "")
 	storeSet.gRPCInfoCallTimeout = 2 * time.Second
 	defer storeSet.Close()
 
@@ -532,7 +532,7 @@ func TestStoreSet_Update_NoneAvailable(t *testing.T) {
 			specs = append(specs, NewGRPCStoreSpec(addr, false))
 		}
 		return specs
-	}, testGRPCOpts, time.Minute)
+	}, testGRPCOpts, time.Minute, false, "", "", "")
 	storeSet.gRPCInfoCallTimeout = 2 * time.Second
 
 	// Should not matter how many of these we run.
@@ -596,7 +596,7 @@ func TestQuerierStrict(t *testing.T) {
 			NewGRPCStoreSpec(st.StoreAddresses()[0], true),
 			NewGRPCStoreSpec(st.StoreAddresses()[1], false),
 		}
-	}, testGRPCOpts, time.Minute)
+	}, testGRPCOpts, time.Minute, false, "", "", "")
 	defer storeSet.Close()
 	storeSet.gRPCInfoCallTimeout = 1 * time.Second