querier: Add validation for querier address flags

CHANGELOG.md

@@ -28,6 +28,7 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 - [#4856](https://github.com/thanos-io/thanos/pull/4856) Mixin: Add Query Frontend Grafana dashboard.
 - [#4874](https://github.com/thanos-io/thanos/pull/4874) Query: Add `--endpoint-strict` flag to statically configure Thanos API server endpoints. It is similar to `--store-strict` but supports passing any Thanos gRPC APIs: StoreAPI, MetadataAPI, RulesAPI, TargetsAPI and ExemplarsAPI.
 - [#4868](https://github.com/thanos-io/thanos/pull/4868) Rule: Support ruleGroup limit introduced by Prometheus v2.31.0.
+- [#4897](https://github.com/thanos-io/thanos/pull/4897) Querier: Add validation for querier address flags.
 
 ### Fixed
 

cmd/thanos/query.go

@@ -104,25 +104,25 @@ func registerQuery(app *extkingpin.App) {
 	selectorLabels := cmd.Flag("selector-label", "Query selector labels that will be exposed in info endpoint (repeated).").
 		PlaceHolder("<name>=\"<value>\"").Strings()
 
-	endpoints := cmd.Flag("endpoint", "Addresses of statically configured Thanos API servers (repeatable). The scheme may be prefixed with 'dns+' or 'dnssrv+' to detect Thanos API servers through respective DNS lookups.").
-		PlaceHolder("<endpoint>").Strings()
+	endpoints := extkingpin.Addrs(cmd.Flag("endpoint", "Addresses of statically configured Thanos API servers (repeatable). The scheme may be prefixed with 'dns+' or 'dnssrv+' to detect Thanos API servers through respective DNS lookups.").
+		PlaceHolder("<endpoint>"))
 
-	stores := cmd.Flag("store", "Deprecation Warning - This flag is deprecated and replaced with `endpoint`. Addresses of statically configured store API servers (repeatable). The scheme may be prefixed with 'dns+' or 'dnssrv+' to detect store API servers through respective DNS lookups.").
-		PlaceHolder("<store>").Strings()
+	stores := extkingpin.Addrs(cmd.Flag("store", "Deprecation Warning - This flag is deprecated and replaced with `endpoint`. Addresses of statically configured store API servers (repeatable). The scheme may be prefixed with 'dns+' or 'dnssrv+' to detect store API servers through respective DNS lookups.").
+		PlaceHolder("<store>"))
 
 	// TODO(bwplotka): Hidden because we plan to extract discovery to separate API: https://github.com/thanos-io/thanos/issues/2600.
-	ruleEndpoints := cmd.Flag("rule", "Deprecation Warning - This flag is deprecated and replaced with `endpoint`. Experimental: Addresses of statically configured rules API servers (repeatable). The scheme may be prefixed with 'dns+' or 'dnssrv+' to detect rule API servers through respective DNS lookups.").
-		Hidden().PlaceHolder("<rule>").Strings()
+	ruleEndpoints := extkingpin.Addrs(cmd.Flag("rule", "Deprecation Warning - This flag is deprecated and replaced with `endpoint`. Experimental: Addresses of statically configured rules API servers (repeatable). The scheme may be prefixed with 'dns+' or 'dnssrv+' to detect rule API servers through respective DNS lookups.").
+		Hidden().PlaceHolder("<rule>"))
 
-	metadataEndpoints := cmd.Flag("metadata", "Deprecation Warning - This flag is deprecated and replaced with `endpoint`. Experimental: Addresses of statically configured metadata API servers (repeatable). The scheme may be prefixed with 'dns+' or 'dnssrv+' to detect metadata API servers through respective DNS lookups.").
-		Hidden().PlaceHolder("<metadata>").Strings()
+	metadataEndpoints := extkingpin.Addrs(cmd.Flag("metadata", "Deprecation Warning - This flag is deprecated and replaced with `endpoint`. Experimental: Addresses of statically configured metadata API servers (repeatable). The scheme may be prefixed with 'dns+' or 'dnssrv+' to detect metadata API servers through respective DNS lookups.").
+		Hidden().PlaceHolder("<metadata>"))
 
-	exemplarEndpoints := cmd.Flag("exemplar", "Deprecation Warning - This flag is deprecated and replaced with `endpoint`. Experimental: Addresses of statically configured exemplars API servers (repeatable). The scheme may be prefixed with 'dns+' or 'dnssrv+' to detect exemplars API servers through respective DNS lookups.").
-		Hidden().PlaceHolder("<exemplar>").Strings()
+	exemplarEndpoints := extkingpin.Addrs(cmd.Flag("exemplar", "Deprecation Warning - This flag is deprecated and replaced with `endpoint`. Experimental: Addresses of statically configured exemplars API servers (repeatable). The scheme may be prefixed with 'dns+' or 'dnssrv+' to detect exemplars API servers through respective DNS lookups.").
+		Hidden().PlaceHolder("<exemplar>"))
 
 	// TODO(atunik): Hidden because we plan to extract discovery to separate API: https://github.com/thanos-io/thanos/issues/2600.
-	targetEndpoints := cmd.Flag("target", "Deprecation Warning - This flag is deprecated and replaced with `endpoint`. Experimental: Addresses of statically configured target API servers (repeatable). The scheme may be prefixed with 'dns+' or 'dnssrv+' to detect target API servers through respective DNS lookups.").
-		Hidden().PlaceHolder("<target>").Strings()
+	targetEndpoints := extkingpin.Addrs(cmd.Flag("target", "Deprecation Warning - This flag is deprecated and replaced with `endpoint`. Experimental: Addresses of statically configured target API servers (repeatable). The scheme may be prefixed with 'dns+' or 'dnssrv+' to detect target API servers through respective DNS lookups.").
+		Hidden().PlaceHolder("<target>"))
 
 	strictStores := cmd.Flag("store-strict", "Deprecation Warning - This flag is deprecated and replaced with `endpoint-strict`. Addresses of only statically configured store API servers that are always used, even if the health check fails. Useful if you have a caching layer on top.").
 		PlaceHolder("<staticstore>").Strings()
@@ -191,22 +191,6 @@ func registerQuery(app *extkingpin.App) {
 			}
 		}
 
-		if dup := firstDuplicate(*stores); dup != "" {
-			return errors.Errorf("Address %s is duplicated for --store flag.", dup)
-		}
-
-		if dup := firstDuplicate(*ruleEndpoints); dup != "" {
-			return errors.Errorf("Address %s is duplicated for --rule flag.", dup)
-		}
-
-		if dup := firstDuplicate(*metadataEndpoints); dup != "" {
-			return errors.Errorf("Address %s is duplicated for --metadata flag.", dup)
-		}
-
-		if dup := firstDuplicate(*exemplarEndpoints); dup != "" {
-			return errors.Errorf("Address %s is duplicated for --exemplar flag.", dup)
-		}
-
 		httpLogOpts, err := logging.ParseHTTPOptions(*reqLogDecision, reqLogConfig)
 		if err != nil {
 			return errors.Wrap(err, "error while parsing config for request logging")
@@ -217,10 +201,6 @@ func registerQuery(app *extkingpin.App) {
 			return errors.Wrap(err, "error while parsing config for request logging")
 		}
 
-		if dup := firstDuplicate(*targetEndpoints); dup != "" {
-			return errors.Errorf("Address %s is duplicated for --target flag.", dup)
-		}
-
 		var fileSD *file.Discovery
 		if len(*fileSDFiles) > 0 {
 			conf := &file.SDConfig{
@@ -733,22 +713,6 @@ func removeDuplicateEndpointSpecs(logger log.Logger, duplicatedStores prometheus
 	return deduplicated
 }
 
-// firstDuplicate returns the first duplicate string in the given string slice
-// or empty string if none was found.
-func firstDuplicate(ss []string) string {
-	set := map[string]struct{}{}
-
-	for _, s := range ss {
-		if _, ok := set[s]; ok {
-			return s
-		}
-
-		set[s] = struct{}{}
-	}
-
-	return ""
-}
-
 // engineFactory creates from 1 to 3 promql.Engines depending on
 // dynamicLookbackDelta and eo.LookbackDelta and returns a function
 // that returns appropriate engine for given maxSourceResolutionMillis.

pkg/extkingpin/flags.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	extflag "github.com/efficientgo/tools/extkingpin"
+	"github.com/pkg/errors"
 	"github.com/prometheus/common/model"
 	"gopkg.in/alecthomas/kingpin.v2"
 )
@@ -20,6 +21,58 @@ func ModelDuration(flags *kingpin.FlagClause) *model.Duration {
 	return value
 }
 
+// Custom parser for IP address flags.
+type addressSlice []string
+
+// addressSlice conforms to flag.Value interface.
+func (a *addressSlice) Set(value string) error {
+	*a = append(*a, value)
+	if err := validateAddrs(*a); err != nil {
+		return err
+	}
+	return nil
+}
+
+func (a *addressSlice) String() string {
+	return strings.Join(*a, ",")
+}
+
+// Ensure flag is repeatable.
+func (a *addressSlice) IsCumulative() bool {
+	return true
+}
+
+func Addrs(flags *kingpin.FlagClause) (target *addressSlice) {
+	target = &addressSlice{}
+	flags.SetValue((*addressSlice)(target))
+	return
+}
+
+// validateAddrs checks an address slice for duplicates and empty or invalid elements.
+func validateAddrs(addrs addressSlice) error {
+	set := map[string]struct{}{}
+
+	for _, addr := range addrs {
+		if addr == "" {
+			return errors.New("Address is empty.")
+		}
+
+		qtypeAndName := strings.SplitN(addr, "+", 2)
+		hostAndPort := strings.SplitN(addr, ":", 2)
+		if len(qtypeAndName) != 2 && len(hostAndPort) != 2 {
+			return errors.Errorf("Address %s is not of <host>:<port> format or a valid DNS query.", addr)
+		}
+
+		if _, ok := set[addr]; ok {
+			return errors.Errorf("Address %s is duplicated.", addr)
+		}
+
+		set[addr] = struct{}{}
+	}
+
+	return nil
+}
+
 // RegisterGRPCFlags registers flags commonly used to configure gRPC servers with.
 func RegisterGRPCFlags(cmd FlagClause) (
 	grpcBindAddr *string,

test/e2e/info_api_test.go

@@ -74,7 +74,6 @@ func TestInfo(t *testing.T) {
 			sidecar1.InternalEndpoint("grpc"),
 			sidecar2.InternalEndpoint("grpc"),
 			sidecar3.InternalEndpoint("grpc"),
-			sidecar3.InternalEndpoint("grpc"),
 			store.InternalEndpoint("grpc"),
 		).
 		Build()