Added file SD (thanos-io#546)

* Added file SD to query and rule
* Added static flags in rule to specify queriers
* Added e2e tests for both static flag SD and file SD

Author: Ivan Valkov

Gopkg.lock

@@ -12,6 +12,7 @@ import (
 	prom "github.com/prometheus/prometheus/config"
 	"gopkg.in/yaml.v2"
 	appsv1 "k8s.io/api/apps/v1"
+	"k8s.io/api/core/v1"
 	rbacv1 "k8s.io/api/rbac/v1"
 	"k8s.io/apimachinery/pkg/api/resource"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
@@ -207,7 +208,7 @@ func createPrometheus(opts *opts, name string, bucket string) *appsv1.StatefulSe
 		Name:      name,
 		Namespace: promNamespace,
 		Labels: map[string]string{
-			"app": name,
+			"app":                  name,
 			"thanos-gossip-member": "true",
 		},
 	}
@@ -370,7 +371,7 @@ func createThanosQuery(opts *opts) (*v1.Service, *v1.Pod) {
 		Name:      "thanos-query",
 		Namespace: thanosNamespace,
 		Labels: map[string]string{
-			"app": "thanos-query",
+			"app":                  "thanos-query",
 			"thanos-gossip-member": "true",
 		},
 	}

benchmark/cmd/thanosbench/resources.go

@@ -16,6 +16,7 @@ import (
 	"github.com/grpc-ecosystem/go-grpc-middleware"
 	"github.com/grpc-ecosystem/go-grpc-prometheus"
 	"github.com/improbable-eng/thanos/pkg/cluster"
+	"github.com/improbable-eng/thanos/pkg/discovery/cache"
 	"github.com/improbable-eng/thanos/pkg/query"
 	"github.com/improbable-eng/thanos/pkg/query/api"
 	"github.com/improbable-eng/thanos/pkg/runutil"
@@ -28,6 +29,8 @@ import (
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/common/route"
+	"github.com/prometheus/prometheus/discovery/file"
+	"github.com/prometheus/prometheus/discovery/targetgroup"
 	"github.com/prometheus/prometheus/promql"
 	"github.com/prometheus/tsdb/labels"
 	"google.golang.org/grpc"
@@ -65,6 +68,12 @@ func registerQuery(m map[string]setupFunc, app *kingpin.Application, name string
 	stores := cmd.Flag("store", "Addresses of statically configured store API servers (repeatable).").
 		PlaceHolder("<store>").Strings()
 
+	fileSDFiles := cmd.Flag("store.file-sd-config.files", "Path to files that contain addresses of store API servers. The path can be a glob pattern (repeatable).").
+		PlaceHolder("<path>").Strings()
+
+	fileSDInterval := modelDuration(cmd.Flag("store.file-sd-config.interval", "Refresh interval to re-read file SD files. (used as a fallback)").
+		Default("5m"))
+
 	enableAutodownsampling := cmd.Flag("query.auto-downsampling", "Enable automatic adjustment (step / 5) to what source of data should be used in store gateways if no max_source_resolution param is specified. ").
 		Default("false").Bool()
 
@@ -87,6 +96,15 @@ func registerQuery(m map[string]setupFunc, app *kingpin.Application, name string
 			lookupStores[s] = struct{}{}
 		}
 
+		var fileSD *file.Discovery
+		if len(*fileSDFiles) > 0 {
+			conf := &file.SDConfig{
+				Files:           *fileSDFiles,
+				RefreshInterval: *fileSDInterval,
+			}
+			fileSD = file.NewDiscovery(conf, logger)
+		}
+
 		return runQuery(
 			g,
 			logger,
@@ -109,6 +127,7 @@ func registerQuery(m map[string]setupFunc, app *kingpin.Application, name string
 			selectorLset,
 			*stores,
 			*enableAutodownsampling,
+			fileSD,
 		)
 	}
 }
@@ -218,7 +237,14 @@ func runQuery(
 	selectorLset labels.Labels,
 	storeAddrs []string,
 	enableAutodownsampling bool,
+	fileSD *file.Discovery,
 ) error {
+	duplicatedStores := prometheus.NewCounter(prometheus.CounterOpts{
+		Name: "thanos_query_duplicated_store_address",
+		Help: "The number of times a duplicated store addresses is detected from the different configs in query",
+	})
+	reg.MustRegister(duplicatedStores)
+
 	var staticSpecs []query.StoreSpec
 	for _, addr := range storeAddrs {
 		if addr == "" {
@@ -233,13 +259,17 @@ func runQuery(
 		return errors.Wrap(err, "building gRPC client")
 	}
 
+	fileSDCache := cache.New()
+
 	var (
 		stores = query.NewStoreSet(
 			logger,
 			reg,
 			func() (specs []query.StoreSpec) {
+				// Add store specs from static flags.
 				specs = append(staticSpecs)
 
+				// Add store specs from gossip.
 				for id, ps := range peer.PeerStates(cluster.PeerTypesStoreAPIs()...) {
 					if ps.StoreAPIAddr == "" {
 						level.Error(logger).Log("msg", "Gossip found peer that propagates empty address, ignoring.", "lset", fmt.Sprintf("%v", ps.Metadata.Labels))
@@ -248,6 +278,14 @@ func runQuery(
 
 					specs = append(specs, &gossipSpec{id: id, addr: ps.StoreAPIAddr, peer: peer})
 				}
+
+				// Add store specs from file SD.
+				for _, addr := range fileSDCache.Addresses() {
+					specs = append(specs, query.NewGRPCStoreSpec(addr))
+				}
+
+				specs = removeDuplicateStoreSpecs(logger, duplicatedStores, specs)
+
 				return specs
 			},
 			dialOpts,
@@ -271,6 +309,40 @@ func runQuery(
 			stores.Close()
 		})
 	}
+	// Run File Service Discovery and update the store set when the files are modified.
+	if fileSD != nil {
+		var fileSDUpdates chan []*targetgroup.Group
+		ctxRun, cancelRun := context.WithCancel(context.Background())
+
+		fileSDUpdates = make(chan []*targetgroup.Group)
+
+		g.Add(func() error {
+			fileSD.Run(ctxRun, fileSDUpdates)
+			return nil
+		}, func(error) {
+			cancelRun()
+		})
+
+		ctxUpdate, cancelUpdate := context.WithCancel(context.Background())
+		g.Add(func() error {
+			for {
+				select {
+				case update := <-fileSDUpdates:
+					// Discoverers sometimes send nil updates so need to check for it to avoid panics.
+					if update == nil {
+						continue
+					}
+					fileSDCache.Update(update)
+					stores.Update(ctxUpdate)
+				case <-ctxUpdate.Done():
+					return nil
+				}
+			}
+		}, func(error) {
+			cancelUpdate()
+			close(fileSDUpdates)
+		})
+	}
 	{
 		ctx, cancel := context.WithCancel(context.Background())
 		g.Add(func() error {
@@ -340,6 +412,23 @@ func runQuery(
 	return nil
 }
 
+func removeDuplicateStoreSpecs(logger log.Logger, duplicatedStores prometheus.Counter, specs []query.StoreSpec) []query.StoreSpec {
+	set := make(map[string]query.StoreSpec)
+	for _, spec := range specs {
+		addr := spec.Addr()
+		if _, ok := set[addr]; ok {
+			level.Warn(logger).Log("msg", "Duplicate store address is provided - %v", addr)
+			duplicatedStores.Inc()
+		}
+		set[addr] = spec
+	}
+	deduplicated := make([]query.StoreSpec, 0, len(set))
+	for _, value := range set {
+		deduplicated = append(deduplicated, value)
+	}
+	return deduplicated
+}
+
 type gossipSpec struct {
 	id   string
 	addr string