Tools: Add flag --selector.relabel-config-file / `--selector.relabe…

…l-config` / `--max-time` / `--min-time` to filter served blocks. (#4453)

.bingo/prometheus.mod

@@ -9,6 +9,7 @@ replace (
 	github.com/cockroachdb/cmux => github.com/cockroachdb/cmux v0.0.0-20170110192607-30d10be49292
 	github.com/cockroachdb/cockroach => github.com/cockroachdb/cockroach v0.0.0-20170608034007-84bc9597164f
 	github.com/googleapis/gnostic => github.com/googleapis/gnostic v0.2.3-0.20180520015035-48a0ecefe2e4
+	github.com/gophercloud/gophercloud => github.com/gophercloud/gophercloud v0.18.0
 	github.com/miekg/dns => github.com/miekg/dns v1.0.4
 	github.com/prometheus/client_golang => github.com/prometheus/client_golang v0.9.0-pre1.0.20180607123607-faf4ec335fe0
 	github.com/prometheus/common => github.com/prometheus/common v0.0.0-20180518154759-7600349dcfe1

CHANGELOG.md

@@ -11,6 +11,7 @@ We use *breaking :warning:* to mark changes that are not backward compatible (re
 ## Unreleased
 
 ### Added
+- [#4453](https://github.com/thanos-io/thanos/pull/4453) Tools: Add flag `--selector.relabel-config-file` / `--selector.relabel-config` / `--max-time` / `--min-time` to filter served blocks.
 
 ### Fixed
 

cmd/thanos/tools_bucket.go

@@ -53,6 +53,7 @@ import (
 	"github.com/thanos-io/thanos/pkg/replicate"
 	"github.com/thanos-io/thanos/pkg/runutil"
 	httpserver "github.com/thanos-io/thanos/pkg/server/http"
+	"github.com/thanos-io/thanos/pkg/store"
 	"github.com/thanos-io/thanos/pkg/ui"
 	"github.com/thanos-io/thanos/pkg/verifier"
 	"golang.org/x/text/language"
@@ -334,6 +335,12 @@ func registerBucketWeb(app extkingpin.AppClause, objStoreConfig *extflag.PathOrC
 	interval := cmd.Flag("refresh", "Refresh interval to download metadata from remote storage").Default("30m").Duration()
 	timeout := cmd.Flag("timeout", "Timeout to download metadata from remote storage").Default("5m").Duration()
 	label := cmd.Flag("label", "Prometheus label to use as timeline title").String()
+	filterConf := &store.FilterConfig{}
+	cmd.Flag("min-time", "Start of time range limit to serve. Thanos tool bucket web will serve only blocks, which happened later than this value. Option can be a constant time in RFC3339 format or time duration relative to current time, such as -1d or 2h45m. Valid duration units are ms, s, m, h, d, w, y.").
+		Default("0000-01-01T00:00:00Z").SetValue(&filterConf.MinTime)
+	cmd.Flag("max-time", "End of time range limit to serve. Thanos tool bucket web will serve only blocks, which happened earlier than this value. Option can be a constant time in RFC3339 format or time duration relative to current time, such as -1d or 2h45m. Valid duration units are ms, s, m, h, d, w, y.").
+		Default("9999-12-31T23:59:59Z").SetValue(&filterConf.MaxTime)
+	selectorRelabelConf := *extkingpin.RegisterSelectorRelabelFlags(cmd)
 
 	cmd.Setup(func(g *run.Group, logger log.Logger, reg *prometheus.Registry, tracer opentracing.Tracer, _ <-chan struct{}, _ bool) error {
 		comp := component.Bucket
@@ -414,8 +421,22 @@ func registerBucketWeb(app extkingpin.AppClause, objStoreConfig *extflag.PathOrC
 			return errors.Wrap(err, "bucket client")
 		}
 
+		relabelContentYaml, err := selectorRelabelConf.Content()
+		if err != nil {
+			return errors.Wrap(err, "get content of relabel configuration")
+		}
+
+		relabelConfig, err := block.ParseRelabelConfig(relabelContentYaml, block.SelectorSupportedRelabelActions)
+		if err != nil {
+			return err
+		}
 		// TODO(bwplotka): Allow Bucket UI to visualize the state of block as well.
-		fetcher, err := block.NewMetaFetcher(logger, block.FetcherConcurrency, bkt, "", extprom.WrapRegistererWithPrefix(extpromPrefix, reg), nil, nil)
+		fetcher, err := block.NewMetaFetcher(logger, block.FetcherConcurrency, bkt, "", extprom.WrapRegistererWithPrefix(extpromPrefix, reg),
+			[]block.MetadataFilter{
+				block.NewTimePartitionMetaFilter(filterConf.MinTime, filterConf.MaxTime),
+				block.NewLabelShardedMetaFilter(relabelConfig),
+				block.NewDeduplicateFilter(),
+			}, nil)
 		if err != nil {
 			return err
 		}

docs/components/tools.md

@@ -228,6 +228,21 @@ Flags:
       --log.format=logfmt       Log format to use. Possible options: logfmt or
                                 json.
       --log.level=info          Log filtering level.
+      --max-time=9999-12-31T23:59:59Z  
+                                End of time range limit to serve. Thanos tool
+                                bucket web will serve only blocks, which
+                                happened earlier than this value. Option can be
+                                a constant time in RFC3339 format or time
+                                duration relative to current time, such as -1d
+                                or 2h45m. Valid duration units are ms, s, m, h,
+                                d, w, y.
+      --min-time=0000-01-01T00:00:00Z  
+                                Start of time range limit to serve. Thanos tool
+                                bucket web will serve only blocks, which
+                                happened later than this value. Option can be a
+                                constant time in RFC3339 format or time duration
+                                relative to current time, such as -1d or 2h45m.
+                                Valid duration units are ms, s, m, h, d, w, y.
       --objstore.config=<content>  
                                 Alternative to 'objstore.config-file' flag
                                 (mutually exclusive). Content of YAML file that
@@ -240,6 +255,20 @@ Flags:
                                 https://thanos.io/tip/thanos/storage.md/#configuration
       --refresh=30m             Refresh interval to download metadata from
                                 remote storage
+      --selector.relabel-config=<content>  
+                                Alternative to 'selector.relabel-config-file'
+                                flag (mutually exclusive). Content of YAML file
+                                that contains relabeling configuration that
+                                allows selecting blocks. It follows native
+                                Prometheus relabel-config syntax. See format
+                                details:
+                                https://prometheus.io/docs/prometheus/latest/configuration/configuration/#relabel_config
+      --selector.relabel-config-file=<file-path>  
+                                Path to YAML file that contains relabeling
+                                configuration that allows selecting blocks. It
+                                follows native Prometheus relabel-config syntax.
+                                See format details:
+                                https://prometheus.io/docs/prometheus/latest/configuration/configuration/#relabel_config
       --timeout=5m              Timeout to download metadata from remote storage
       --tracing.config=<content>  
                                 Alternative to 'tracing.config-file' flag

test/e2e/e2ethanos/services.go

@@ -662,7 +662,14 @@ func NewMemcached(name string) *e2e.ConcreteService {
 	return memcached
 }
 
-func NewToolsBucketWeb(name string, bucketConfig client.BucketConfig, routePrefix, externalPrefix string) (*Service, error) {
+func NewToolsBucketWeb(
+	name string,
+	bucketConfig client.BucketConfig,
+	routePrefix,
+	externalPrefix string,
+	minTime string,
+	maxTime string,
+	relabelConfig string) (*Service, error) {
 	bktConfigBytes, err := yaml.Marshal(bucketConfig)
 	if err != nil {
 		return nil, errors.Wrapf(err, "generate tools bucket web config file: %v", bucketConfig)
@@ -682,6 +689,18 @@ func NewToolsBucketWeb(name string, bucketConfig client.BucketConfig, routePrefi
 		args = append(args, "--web.external-prefix="+externalPrefix)
 	}
 
+	if minTime != "" {
+		args = append(args, "--min-time="+minTime)
+	}
+
+	if maxTime != "" {
+		args = append(args, "--max-time="+maxTime)
+	}
+
+	if relabelConfig != "" {
+		args = append(args, "--selector.relabel-config="+relabelConfig)
+	}
+
 	args = append([]string{"bucket", "web"}, args...)
 
 	toolsBucketWeb := NewService(

test/e2e/tools_bucket_web_test.go

@@ -4,12 +4,25 @@
 package e2e_test
 
 import (
+	"context"
+	"encoding/json"
+	"io/ioutil"
+	"net/http"
 	"net/http/httptest"
+	"os"
+	"path"
+	"path/filepath"
 	"testing"
+	"time"
 
 	"github.com/cortexproject/cortex/integration/e2e"
 	e2edb "github.com/cortexproject/cortex/integration/e2e/db"
+	"github.com/go-kit/kit/log"
+	"github.com/prometheus/prometheus/pkg/labels"
+	"github.com/prometheus/prometheus/pkg/timestamp"
 
+	v1 "github.com/thanos-io/thanos/pkg/api/blocks"
+	"github.com/thanos-io/thanos/pkg/objstore"
 	"github.com/thanos-io/thanos/pkg/objstore/client"
 	"github.com/thanos-io/thanos/pkg/objstore/s3"
 	"github.com/thanos-io/thanos/pkg/testutil"
@@ -43,6 +56,9 @@ func TestToolsBucketWebExternalPrefixWithoutReverseProxy(t *testing.T) {
 		svcConfig,
 		"",
 		externalPrefix,
+		"",
+		"",
+		"",
 	)
 	testutil.Ok(t, err)
 	testutil.Ok(t, s.StartAndWaitReady(b))
@@ -78,6 +94,9 @@ func TestToolsBucketWebExternalPrefix(t *testing.T) {
 		svcConfig,
 		"",
 		externalPrefix,
+		"",
+		"",
+		"",
 	)
 	testutil.Ok(t, err)
 	testutil.Ok(t, s.StartAndWaitReady(b))
@@ -119,6 +138,9 @@ func TestToolsBucketWebExternalPrefixAndRoutePrefix(t *testing.T) {
 		svcConfig,
 		routePrefix,
 		externalPrefix,
+		"",
+		"",
+		"",
 	)
 	testutil.Ok(t, err)
 	testutil.Ok(t, s.StartAndWaitReady(b))
@@ -130,3 +152,99 @@ func TestToolsBucketWebExternalPrefixAndRoutePrefix(t *testing.T) {
 
 	checkNetworkRequests(t, toolsBucketWebProxy.URL+"/"+externalPrefix+"/blocks")
 }
+
+func TestToolsBucketWebWithTimeAndRelabelFilter(t *testing.T) {
+	t.Parallel()
+	// Create network.
+	s, err := e2e.NewScenario("e2e_test_tools_bucket_web_time_and_relabel_filter")
+	testutil.Ok(t, err)
+	t.Cleanup(e2ethanos.CleanScenario(t, s))
+	// Create Minio.
+	const bucket = "toolsBucketWeb_test"
+	m := e2edb.NewMinio(8080, bucket)
+	testutil.Ok(t, s.StartAndWaitReady(m))
+	// Create bucket.
+	logger := log.NewLogfmtLogger(os.Stdout)
+	bkt, err := s3.NewBucketWithConfig(logger, s3.Config{
+		Bucket:    bucket,
+		AccessKey: e2edb.MinioAccessKey,
+		SecretKey: e2edb.MinioSecretKey,
+		Endpoint:  m.HTTPEndpoint(),
+		Insecure:  true,
+	}, "tools")
+	testutil.Ok(t, err)
+	// Create share dir for upload.
+	dir := filepath.Join(s.SharedDir(), "tmp")
+	testutil.Ok(t, os.MkdirAll(dir, os.ModePerm))
+	// Upload blocks.
+	now, err := time.Parse(time.RFC3339, "2021-07-24T08:00:00Z")
+	testutil.Ok(t, err)
+	blocks := []blockDesc{
+		{
+			series:  []labels.Labels{labels.FromStrings("a", "1", "b", "2")},
+			extLset: labels.FromStrings("tenant_id", "b", "replica", "1"),
+			mint:    timestamp.FromTime(now),
+			maxt:    timestamp.FromTime(now.Add(2 * time.Hour)),
+		},
+		{
+			series:  []labels.Labels{labels.FromStrings("a", "1", "b", "2")},
+			extLset: labels.FromStrings("tenant_id", "a", "replica", "1"),
+			mint:    timestamp.FromTime(now),
+			maxt:    timestamp.FromTime(now.Add(2 * time.Hour)),
+		},
+		{
+			series:  []labels.Labels{labels.FromStrings("a", "1", "b", "2")},
+			extLset: labels.FromStrings("tenant_id", "b", "replica", "1"),
+			mint:    timestamp.FromTime(now.Add(2 * time.Hour)),
+			maxt:    timestamp.FromTime(now.Add(4 * time.Hour)),
+		},
+	}
+	for _, b := range blocks {
+		id, err := b.Create(context.Background(), dir, 0, b.hashFunc)
+		testutil.Ok(t, err)
+		testutil.Ok(t, objstore.UploadDir(context.Background(), logger, bkt, path.Join(dir, id.String()), id.String()))
+	}
+	// Start thanos tool bucket web.
+	svcConfig := client.BucketConfig{
+		Type: client.S3,
+		Config: s3.Config{
+			Bucket:    bucket,
+			AccessKey: e2edb.MinioAccessKey,
+			SecretKey: e2edb.MinioSecretKey,
+			Endpoint:  m.NetworkHTTPEndpoint(),
+			Insecure:  true,
+		},
+	}
+	b, err := e2ethanos.NewToolsBucketWeb(
+		"1",
+		svcConfig,
+		"",
+		"",
+		now.Format(time.RFC3339),
+		now.Add(1*time.Hour).Format(time.RFC3339),
+		`
+- action: keep
+  regex: "b"
+  source_labels: ["tenant_id"]`,
+	)
+	testutil.Ok(t, err)
+	testutil.Ok(t, s.StartAndWaitReady(b))
+	// Request blocks api.
+	resp, err := http.DefaultClient.Get("http://" + b.HTTPEndpoint() + "/api/v1/blocks")
+	testutil.Ok(t, err)
+	testutil.Equals(t, http.StatusOK, resp.StatusCode)
+	defer resp.Body.Close()
+	body, err := ioutil.ReadAll(resp.Body)
+	testutil.Ok(t, err)
+	var data struct {
+		Status string
+		Data   *v1.BlocksInfo
+	}
+	testutil.Ok(t, json.Unmarshal(body, &data))
+	testutil.Equals(t, "success", data.Status)
+	// Filtered by time and relabel, result only one blocks.
+	testutil.Equals(t, 1, len(data.Data.Blocks))
+	testutil.Equals(t, data.Data.Blocks[0].MaxTime, blocks[0].maxt)
+	testutil.Equals(t, data.Data.Blocks[0].MinTime, blocks[0].mint)
+	testutil.Equals(t, data.Data.Blocks[0].Thanos.Labels, blocks[0].extLset.Map())
+}