federated targets functionality (thanos-io#1375)

Signed-off-by: Alexander Tunik <[email protected]>

cmd/thanos/query.go

@@ -46,6 +46,7 @@ import (
 	grpcserver "github.com/thanos-io/thanos/pkg/server/grpc"
 	httpserver "github.com/thanos-io/thanos/pkg/server/http"
 	"github.com/thanos-io/thanos/pkg/store"
+	"github.com/thanos-io/thanos/pkg/targets"
 	"github.com/thanos-io/thanos/pkg/tls"
 	"github.com/thanos-io/thanos/pkg/ui"
 )
@@ -103,6 +104,10 @@ func registerQuery(app *extkingpin.App) {
 	metadataEndpoints := cmd.Flag("metadata", "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()
 
+	// TODO(atunik): Hidden because we plan to extract discovery to separate API: https://github.com/thanos-io/thanos/issues/2600.
+	targetEndpoints := cmd.Flag("target", "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()
+
 	strictStores := cmd.Flag("store-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()
 
@@ -169,6 +174,10 @@ 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{
@@ -222,6 +231,7 @@ func registerQuery(app *extkingpin.App) {
 			getFlagsMap(cmd.Flags()),
 			*stores,
 			*ruleEndpoints,
+			*targetEndpoints,
 			*metadataEndpoints,
 			*enableAutodownsampling,
 			*enableQueryPartialResponse,
@@ -278,6 +288,7 @@ func runQuery(
 	flagsMap map[string]string,
 	storeAddrs []string,
 	ruleAddrs []string,
+	targetAddrs []string,
 	metadataAddrs []string,
 	enableAutodownsampling bool,
 	enableQueryPartialResponse bool,
@@ -323,6 +334,12 @@ func runQuery(
 		dns.ResolverType(dnsSDResolver),
 	)
 
+	dnsTargetProvider := dns.NewProvider(
+		logger,
+		extprom.WrapRegistererWithPrefix("thanos_query_target_apis_", reg),
+		dns.ResolverType(dnsSDResolver),
+	)
+
 	dnsMetadataProvider := dns.NewProvider(
 		logger,
 		extprom.WrapRegistererWithPrefix("thanos_query_metadata_apis_", reg),
@@ -355,6 +372,13 @@ func runQuery(
 
 				return specs
 			},
+			func() (specs []query.TargetSpec) {
+				for _, addr := range dnsTargetProvider.Addresses() {
+					specs = append(specs, query.NewGRPCStoreSpec(addr, false))
+				}
+
+				return specs
+			},
 			func() (specs []query.MetadataSpec) {
 				for _, addr := range dnsMetadataProvider.Addresses() {
 					specs = append(specs, query.NewGRPCStoreSpec(addr, false))
@@ -367,6 +391,7 @@ func runQuery(
 		)
 		proxy            = store.NewProxyStore(logger, reg, stores.Get, component.Query, selectorLset, storeResponseTimeout)
 		rulesProxy       = rules.NewProxy(logger, stores.GetRulesClients)
+		targetsProxy     = targets.NewProxy(logger, stores.GetTargetsClients)
 		metadataProxy    = metadata.NewProxy(logger, stores.GetMetadataClients)
 		queryableCreator = query.NewQueryableCreator(
 			logger,
@@ -454,6 +479,9 @@ func runQuery(
 				if err := dnsRuleProvider.Resolve(resolveCtx, ruleAddrs); err != nil {
 					level.Error(logger).Log("msg", "failed to resolve addresses for rulesAPIs", "err", err)
 				}
+				if err := dnsTargetProvider.Resolve(ctx, targetAddrs); err != nil {
+					level.Error(logger).Log("msg", "failed to resolve addresses for targetsAPIs", "err", err)
+				}
 				if err := dnsMetadataProvider.Resolve(resolveCtx, metadataAddrs); err != nil {
 					level.Error(logger).Log("msg", "failed to resolve addresses for metadataAPIs", "err", err)
 				}
@@ -504,6 +532,7 @@ func runQuery(
 			queryableCreator,
 			// NOTE: Will share the same replica label as the query for now.
 			rules.NewGRPCClientWithDedup(rulesProxy, queryReplicaLabels),
+			targets.NewGRPCClientWithDedup(targetsProxy, queryReplicaLabels),
 			metadata.NewGRPCClient(metadataProxy),
 			enableAutodownsampling,
 			enableQueryPartialResponse,
@@ -550,6 +579,7 @@ func runQuery(
 		s := grpcserver.New(logger, reg, tracer, grpcLogOpts, tagOpts, comp, grpcProbe,
 			grpcserver.WithServer(store.RegisterStoreServer(proxy)),
 			grpcserver.WithServer(rules.RegisterRulesServer(rulesProxy)),
+			grpcserver.WithServer(targets.RegisterTargetsServer(targetsProxy)),
 			grpcserver.WithServer(metadata.RegisterMetadataServer(metadataProxy)),
 			grpcserver.WithListen(grpcBindAddr),
 			grpcserver.WithGracePeriod(grpcGracePeriod),

cmd/thanos/sidecar.go

@@ -42,6 +42,7 @@ import (
 	httpserver "github.com/thanos-io/thanos/pkg/server/http"
 	"github.com/thanos-io/thanos/pkg/shipper"
 	"github.com/thanos-io/thanos/pkg/store"
+	"github.com/thanos-io/thanos/pkg/targets"
 	"github.com/thanos-io/thanos/pkg/tls"
 	"github.com/thanos-io/thanos/pkg/tracing"
 )
@@ -229,6 +230,7 @@ func runSidecar(
 		s := grpcserver.New(logger, reg, tracer, grpcLogOpts, tagOpts, comp, grpcProbe,
 			grpcserver.WithServer(store.RegisterStoreServer(promStore)),
 			grpcserver.WithServer(rules.RegisterRulesServer(rules.NewPrometheus(conf.prometheus.url, c, m.Labels))),
+			grpcserver.WithServer(targets.RegisterTargetsServer(targets.NewPrometheus(conf.prometheus.url, c, m.Labels))),
 			grpcserver.WithServer(meta.RegisterMetadataServer(meta.NewPrometheus(conf.prometheus.url, c))),
 			grpcserver.WithListen(conf.grpc.bindAddress),
 			grpcserver.WithGracePeriod(time.Duration(conf.grpc.gracePeriod)),

go.mod

@@ -17,6 +17,7 @@ require (
 	github.com/facette/natsort v0.0.0-20181210072756-2cd4dd1e2dcb
 	github.com/fatih/structtag v1.1.0
 	github.com/felixge/fgprof v0.9.1
+	github.com/fortytw2/leaktest v1.3.0
 	github.com/fsnotify/fsnotify v1.4.9
 	github.com/go-kit/kit v0.10.0
 	github.com/go-openapi/strfmt v0.20.0

go.sum

@@ -318,6 +318,7 @@ github.com/felixge/httpsnoop v1.0.1/go.mod h1:m8KPJKqk1gH5J9DgRY2ASl2lWCfGKXixSw
 github.com/fogleman/gg v1.2.1-0.20190220221249-0403632d5b90/go.mod h1:R/bRT+9gY/C5z7JzPU0zXsXHKM4/ayA+zqcVNZzPa1k=
 github.com/form3tech-oss/jwt-go v3.2.2+incompatible h1:TcekIExNqud5crz4xD2pavyTgWiPvpYe4Xau31I0PRk=
 github.com/form3tech-oss/jwt-go v3.2.2+incompatible/go.mod h1:pbq4aXjuKjdthFRnoDwaVPLA+WlJuPGy+QneDUgJi2k=
+github.com/fortytw2/leaktest v1.3.0 h1:u8491cBMTQ8ft8aeV+adlcytMZylmA5nnwwkRZjI8vw=
 github.com/fortytw2/leaktest v1.3.0/go.mod h1:jDsjWgpAGjm2CA7WthBh/CdZYEPF31XHquHwclZch5g=
 github.com/franela/goblin v0.0.0-20200105215937-c9ffbefa60db/go.mod h1:7dvUGVsVBjqR7JHJk0brhHOZYGmfBYOrK0ZhYMEtBr4=
 github.com/franela/goreq v0.0.0-20171204163338-bcd34c9993f8/go.mod h1:ZhphrRTfi2rbfLwlschooIH4+wKKDR4Pdxhh+TRoA20=

pkg/api/query/v1.go

@@ -50,6 +50,8 @@ import (
 	"github.com/thanos-io/thanos/pkg/rules/rulespb"
 	"github.com/thanos-io/thanos/pkg/runutil"
 	"github.com/thanos-io/thanos/pkg/store/storepb"
+	"github.com/thanos-io/thanos/pkg/targets"
+	"github.com/thanos-io/thanos/pkg/targets/targetspb"
 	"github.com/thanos-io/thanos/pkg/tracing"
 )
 
@@ -72,6 +74,7 @@ type QueryAPI struct {
 	// queryEngine returns appropriate promql.Engine for a query with a given step.
 	queryEngine func(int64) *promql.Engine
 	ruleGroups  rules.UnaryClient
+	targets     targets.UnaryClient
 	metadatas   metadata.UnaryClient
 
 	enableAutodownsampling              bool
@@ -95,6 +98,7 @@ func NewQueryAPI(
 	qe func(int64) *promql.Engine,
 	c query.QueryableCreator,
 	ruleGroups rules.UnaryClient,
+	targets targets.UnaryClient,
 	metadatas metadata.UnaryClient,
 	enableAutodownsampling bool,
 	enableQueryPartialResponse bool,
@@ -115,6 +119,7 @@ func NewQueryAPI(
 		queryableCreate: c,
 		gate:            gate,
 		ruleGroups:      ruleGroups,
+		targets:         targets,
 		metadatas:       metadatas,
 
 		enableAutodownsampling:                 enableAutodownsampling,
@@ -154,6 +159,8 @@ func (qapi *QueryAPI) Register(r *route.Router, tracer opentracing.Tracer, logge
 
 	r.Get("/rules", instr("rules", NewRulesHandler(qapi.ruleGroups, qapi.enableRulePartialResponse)))
 
+	r.Get("/targets", instr("targets", NewTargetsHandler(qapi.targets, true)))
+
 	r.Get("/metadata", instr("metadata", NewMetricMetadataHandler(qapi.metadatas, qapi.enableMetricMetadataPartialResponse)))
 }
 
@@ -652,6 +659,36 @@ func (qapi *QueryAPI) stores(_ *http.Request) (interface{}, []error, *api.ApiErr
 	return statuses, nil, nil
 }
 
+func NewTargetsHandler(client targets.UnaryClient, enablePartialResponse bool) func(*http.Request) (interface{}, []error, *api.ApiError) {
+	ps := storepb.PartialResponseStrategy_ABORT
+	if enablePartialResponse {
+		ps = storepb.PartialResponseStrategy_WARN
+	}
+
+	return func(r *http.Request) (interface{}, []error, *api.ApiError) {
+		stateParam := r.URL.Query().Get("state")
+		state, ok := targetspb.TargetsRequest_State_value[strings.ToUpper(stateParam)]
+		if !ok {
+			if stateParam != "" {
+				return nil, nil, &api.ApiError{Typ: api.ErrorBadData, Err: errors.Errorf("invalid targets parameter state='%v'", stateParam)}
+			}
+			state = int32(targetspb.TargetsRequest_ANY)
+		}
+
+		req := &targetspb.TargetsRequest{
+			State:                   targetspb.TargetsRequest_State(state),
+			PartialResponseStrategy: ps,
+		}
+
+		t, warnings, err := client.Targets(r.Context(), req)
+		if err != nil {
+			return nil, nil, &api.ApiError{Typ: api.ErrorInternal, Err: errors.Wrap(err, "retrieving targets")}
+		}
+
+		return t, warnings, nil
+	}
+}
+
 // NewRulesHandler created handler compatible with HTTP /api/v1/rules https://prometheus.io/docs/prometheus/latest/querying/api/#rules
 // which uses gRPC Unary Rules API.
 func NewRulesHandler(client rules.UnaryClient, enablePartialResponse bool) func(*http.Request) (interface{}, []error, *api.ApiError) {

pkg/promclient/promclient.go

@@ -35,6 +35,7 @@ import (
 	"github.com/thanos-io/thanos/pkg/rules/rulespb"
 	"github.com/thanos-io/thanos/pkg/runutil"
 	"github.com/thanos-io/thanos/pkg/store/storepb"
+	"github.com/thanos-io/thanos/pkg/targets/targetspb"
 	"github.com/thanos-io/thanos/pkg/tracing"
 	"google.golang.org/grpc/codes"
 	yaml "gopkg.in/yaml.v2"
@@ -758,3 +759,19 @@ func (c *Client) MetadataInGRPC(ctx context.Context, base *url.URL, metric strin
 	}
 	return v.Data, c.get2xxResultWithGRPCErrors(ctx, "/metadata HTTP[client]", &u, &v)
 }
+
+func (c *Client) TargetsInGRPC(ctx context.Context, base *url.URL, stateTargets string) (*targetspb.TargetDiscovery, error) {
+	u := *base
+	u.Path = path.Join(u.Path, "/api/v1/targets")
+
+	if stateTargets != "" {
+		q := u.Query()
+		q.Add("state", stateTargets)
+		u.RawQuery = q.Encode()
+	}
+
+	var v struct {
+		Data *targetspb.TargetDiscovery `json:"data"`
+	}
+	return v.Data, c.get2xxResultWithGRPCErrors(ctx, "/targets HTTP[client]", &u, &v)
+}