Rebase master && update storeset test

Signed-off-by: jojohappy <sarahdj0917@gmail.com>

cmd/thanos/compact.go

@@ -156,7 +156,7 @@ func runCompact(
 	maxCompactionLevel int,
 	blockSyncConcurrency int,
 	concurrency int,
-	selectorRelabelConf *pathOrContent,
+	selectorRelabelConf *extflag.PathOrContent,
 ) error {
 	halted := prometheus.NewGauge(prometheus.GaugeOpts{
 		Name: "thanos_compactor_halted",

cmd/thanos/flags.go

@@ -73,23 +73,11 @@ func regCommonTracingFlags(app *kingpin.Application) *extflag.PathOrContent {
 	)
 }
 
-func regSelectorRelabelFlags(cmd *kingpin.CmdClause) *pathOrContent {
-	fileFlagName := "selector.relabel-config-file"
-	contentFlagName := "selector.relabel-config"
-	help := "Path to YAML file that contains seletor relabeling configuration. It follows native Prometheus relabel-config syntax. See format details: https://prometheus.io/docs/prometheus/latest/configuration/configuration/#relabel_config "
-
-	selectorRelabelConfFile := cmd.Flag(fileFlagName, help).PlaceHolder("<seletor.relabel-config-yaml-path>").String()
-
-	help = fmt.Sprintf("Alternative to '%s' flag. Relabeling configuration in YAML. See format details: https://prometheus.io/docs/prometheus/latest/configuration/configuration/#relabel_config", fileFlagName)
-
-	selectorRelabelConf := cmd.Flag(contentFlagName, help).PlaceHolder("<selector.relabel-config-yaml>").String()
-
-	return &pathOrContent{
-		fileFlagName:    fileFlagName,
-		contentFlagName: contentFlagName,
-		required:        false,
-
-		path:    selectorRelabelConfFile,
-		content: selectorRelabelConf,
-	}
+func regSelectorRelabelFlags(cmd *kingpin.CmdClause) *extflag.PathOrContent {
+	return extflag.RegisterPathOrContent(
+		cmd,
+		"selector.relabel-config",
+		"YAML file that contains seletor relabeling configuration. It follows native Prometheus relabel-config syntax. See format details: https://prometheus.io/docs/prometheus/latest/configuration/configuration/#relabel_config ",
+		false,
+	)
 }

cmd/thanos/store.go

@@ -13,6 +13,7 @@ import (
 	"github.com/opentracing/opentracing-go"
 	"github.com/pkg/errors"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/prometheus/prometheus/pkg/relabel"
 	"github.com/thanos-io/thanos/pkg/component"
 	"github.com/thanos-io/thanos/pkg/model"
 	"github.com/thanos-io/thanos/pkg/objstore/client"
@@ -21,6 +22,7 @@ import (
 	"github.com/thanos-io/thanos/pkg/store"
 	storecache "github.com/thanos-io/thanos/pkg/store/cache"
 	"gopkg.in/alecthomas/kingpin.v2"
+	yaml "gopkg.in/yaml.v2"
 )
 
 // registerStore registers a store command.
@@ -116,7 +118,7 @@ func runStore(
 	syncInterval time.Duration,
 	blockSyncConcurrency int,
 	filterConf *store.FilterConfig,
-	selectorRelabelConf *pathOrContent,
+	selectorRelabelConf *extflag.PathOrContent,
 ) error {
 	statusProber := prober.NewProber(component, logger, prometheus.WrapRegistererWithPrefix("thanos_", reg))
 

pkg/compact/compact.go

@@ -38,7 +38,6 @@ const (
 )
 
 var blockTooFreshSentinelError = errors.New("Block too fresh")
-var blockDroppedInRelabelingError = errors.New("Block dropped in Relabeling")
 
 // Syncer syncronizes block metas from a bucket into a local directory.
 // It sorts them into compaction groups based on equal label sets.
@@ -210,9 +209,10 @@ func (c *Syncer) syncMetas(ctx context.Context) error {
 				}
 
 				meta, err := c.downloadMeta(workCtx, id)
-				if err == blockTooFreshSentinelError || err == blockDroppedInRelabelingError {
+				if err == blockTooFreshSentinelError {
 					continue
 				}
+
 				if err != nil {
 					if removedOrIgnored := c.removeIfMetaMalformed(workCtx, id); removedOrIgnored {
 						continue
@@ -221,6 +221,15 @@ func (c *Syncer) syncMetas(ctx context.Context) error {
 					return
 				}
 
+				// Check for block labels by relabeling.
+				// If output is empty, the block will be dropped.
+				lset := promlables.FromMap(meta.Thanos.Labels)
+				processedLabels := relabel.Process(lset, c.relabelConfig...)
+				if processedLabels == nil {
+					level.Warn(c.logger).Log("msg", "dropping block(drop in relabeling)", "block", id)
+					return
+				}
+
 				c.blocksMtx.Lock()
 				c.blocks[id] = meta
 				c.blocksMtx.Unlock()
@@ -280,15 +289,6 @@ func (c *Syncer) downloadMeta(ctx context.Context, id ulid.ULID) (*metadata.Meta
 		return nil, errors.Wrapf(err, "downloading meta.json for %s", id)
 	}
 
-	// Check for block labels by relabeling.
-	// If output is empty, the block will be dropped.
-	lset := promlables.FromMap(meta.Thanos.Labels)
-	processedLabels := relabel.Process(lset, c.relabelConfig...)
-	if processedLabels == nil {
-		level.Debug(c.logger).Log("msg", "dropping block(drop in relabeling)", "block", id)
-		return nil, blockDroppedInRelabelingError
-	}
-
 	// ULIDs contain a millisecond timestamp. We do not consider blocks that have been created too recently to
 	// avoid races when a block is only partially uploaded. This relates to all blocks, excluding:
 	// - repair created blocks

pkg/query/storeset.go

@@ -218,7 +218,10 @@ func (s *StoreSet) Update(ctx context.Context) {
 	// Record the number of occurrences of external label combinations for current store slice.
 	externalLabelOccurrencesInStores := map[string]int{}
 	for _, st := range healthyStores {
-		externalLabelOccurrencesInStores[externalLabelsFromStore(st)]++
+		if st.storeType != nil && (st.storeType.ToProto() == storepb.StoreType_QUERY ||
+			st.storeType.ToProto() == storepb.StoreType_SIDECAR) {
+			externalLabelOccurrencesInStores[externalLabelsFromStore(st)]++
+		}
 	}
 	level.Debug(s.logger).Log("msg", "updating healthy stores", "externalLabelOccurrencesInStores", fmt.Sprintf("%#+v", externalLabelOccurrencesInStores))
 
@@ -252,7 +255,11 @@ func (s *StoreSet) Update(ctx context.Context) {
 		// Note: No external labels means strictly store gateway or ruler and it is fine to have access to multiple instances of them.
 		// Any other component will error out if it will be configured with empty external labels.
 		externalLabels := externalLabelsFromStore(store)
-		if len(store.LabelSets()) > 0 && externalLabelOccurrencesInStores[externalLabels] != 1 {
+		if len(store.LabelSets()) > 0 &&
+			store.storeType != nil &&
+			(store.storeType.ToProto() == storepb.StoreType_QUERY ||
+				store.storeType.ToProto() == storepb.StoreType_SIDECAR) &&
+			externalLabelOccurrencesInStores[externalLabels] != 1 {
 			store.close()
 			s.updateStoreStatus(store, errors.New(droppingStoreMessage))
 			level.Warn(s.logger).Log("msg", droppingStoreMessage, "address", addr, "extLset", externalLabels, "duplicates", externalLabelOccurrencesInStores[externalLabels])
@@ -360,12 +367,6 @@ func externalLabelsFromStore(store *storeRef) string {
 	}
 	sort.Strings(tsdbLabelSetStrings)
 
-	// Append the storeType to the end of list, because the store gateway will be exposed external labels,
-	// we allow the duplicate external labels with different components.
-	if store.storeType != nil {
-		tsdbLabelSetStrings = append(tsdbLabelSetStrings, store.storeType.String())
-	}
-
 	return strings.Join(tsdbLabelSetStrings, ",")
 }
 

pkg/query/storeset_test.go

@@ -14,6 +14,7 @@ import (
 	"github.com/fortytw2/leaktest"
 	"github.com/go-kit/kit/log"
 	"github.com/go-kit/kit/log/level"
+	"github.com/thanos-io/thanos/pkg/component"
 	"github.com/thanos-io/thanos/pkg/store/storepb"
 	"github.com/thanos-io/thanos/pkg/testutil"
 	"google.golang.org/grpc"
@@ -26,6 +27,11 @@ var testGRPCOpts = []grpc.DialOption{
 	grpc.WithInsecure(),
 }
 
+var (
+	emptyStoresExtLabels [][]storepb.Label
+	emptyStoresTypes     []component.StoreAPI
+)
+
 type testStore struct {
 	info storepb.InfoResponse
 }
@@ -54,7 +60,7 @@ type testStores struct {
 	srvs map[string]*grpc.Server
 }
 
-func newTestStores(numStores int, storesExtLabels ...[]storepb.Label) (*testStores, error) {
+func newTestStores(numStores int, storesExtLabels [][]storepb.Label, storesTypes []component.StoreAPI) (*testStores, error) {
 	st := &testStores{
 		srvs: map[string]*grpc.Server{},
 	}
@@ -79,7 +85,15 @@ func newTestStores(numStores int, storesExtLabels ...[]storepb.Label) (*testStor
 			return []storepb.LabelSet{{Labels: storesExtLabels[i]}}
 		}
 
-		srv, addr, err := startStore(lsetFn)
+		storeTypeFn := func() storepb.StoreType {
+			if len(storesTypes) != numStores {
+				return component.Sidecar.ToProto()
+			}
+			st := storesTypes[i]
+			return st.ToProto()
+		}
+
+		srv, addr, err := startStore(lsetFn, storeTypeFn)
 		if err != nil {
 			// Close so far started servers.
 			st.Close()
@@ -92,14 +106,14 @@ func newTestStores(numStores int, storesExtLabels ...[]storepb.Label) (*testStor
 	return st, nil
 }
 
-func startStore(lsetFn func(addr string) []storepb.LabelSet) (*grpc.Server, string, error) {
+func startStore(lsetFn func(addr string) []storepb.LabelSet, storeTypeFn func() storepb.StoreType) (*grpc.Server, string, error) {
 	listener, err := net.Listen("tcp", "127.0.0.1:0")
 	if err != nil {
 		return nil, "", err
 	}
 
 	srv := grpc.NewServer()
-	storepb.RegisterStoreServer(srv, &testStore{info: storepb.InfoResponse{LabelSets: lsetFn(listener.Addr().String())}})
+	storepb.RegisterStoreServer(srv, &testStore{info: storepb.InfoResponse{LabelSets: lsetFn(listener.Addr().String()), StoreType: storeTypeFn()}})
 	go func() {
 		_ = srv.Serve(listener)
 	}()
@@ -144,7 +158,7 @@ func specsFromAddrFunc(addrs []string) func() []StoreSpec {
 func TestStoreSet_AllAvailable_ThenDown(t *testing.T) {
 	defer leaktest.CheckTimeout(t, 10*time.Second)()
 
-	st, err := newTestStores(2)
+	st, err := newTestStores(2, emptyStoresExtLabels, emptyStoresTypes)
 	testutil.Ok(t, err)
 	defer st.Close()
 
@@ -188,7 +202,7 @@ func TestStoreSet_AllAvailable_ThenDown(t *testing.T) {
 func TestStoreSet_StaticStores_OneAvailable(t *testing.T) {
 	defer leaktest.CheckTimeout(t, 10*time.Second)()
 
-	st, err := newTestStores(2)
+	st, err := newTestStores(2, emptyStoresExtLabels, emptyStoresTypes)
 	testutil.Ok(t, err)
 	defer st.Close()
 
@@ -217,7 +231,7 @@ func TestStoreSet_StaticStores_OneAvailable(t *testing.T) {
 func TestStoreSet_StaticStores_NoneAvailable(t *testing.T) {
 	defer leaktest.CheckTimeout(t, 10*time.Second)()
 
-	st, err := newTestStores(2)
+	st, err := newTestStores(2, emptyStoresExtLabels, emptyStoresTypes)
 	testutil.Ok(t, err)
 	defer st.Close()
 
@@ -263,7 +277,16 @@ func TestStoreSet_AllAvailable_BlockExtLsetDuplicates(t *testing.T) {
 		},
 	}
 
-	st, err := newTestStores(6, storeExtLabels...)
+	storeTypes := []component.StoreAPI{
+		component.Query,
+		component.Sidecar,
+		component.Query,
+		component.Store,
+		component.Store,
+		component.Sidecar,
+	}
+
+	st, err := newTestStores(6, storeExtLabels, storeTypes)
 	testutil.Ok(t, err)
 	defer st.Close()
 

pkg/store/bucket.go

@@ -470,7 +470,7 @@ func (s *BucketStore) addBlock(ctx context.Context, id ulid.ULID) (err error) {
 		level.Debug(s.logger).Log("msg", "dropping block(drop in relabeling)", "block", id)
 		return os.RemoveAll(dir)
 	}
-	b.labels = labels.FromMap(processedLabels.Map())
+	b.labels = lset
 	sort.Sort(b.labels)
 
 	set, ok := s.blockSets[h]
@@ -542,25 +542,23 @@ func (s *BucketStore) Info(context.Context, *storepb.InfoRequest) (*storepb.Info
 		MaxTime:   maxt,
 	}
 
-	if len(s.relabelConfig) != 0 {
-		labelSets := make(map[uint64][]storepb.Label, len(s.blocks))
-		for _, bs := range s.blocks {
-			ls := map[string]string{}
-			for _, l := range bs.labels {
-				ls[l.Name] = l.Value
-			}
-
-			res := []storepb.Label{}
-			for k, v := range ls {
-				res = append(res, storepb.Label{Name: k, Value: v})
-			}
-			labelSets[bs.labels.Hash()] = res
+	labelSets := make(map[uint64][]storepb.Label, len(s.blocks))
+	for _, bs := range s.blocks {
+		ls := map[string]string{}
+		for _, l := range bs.labels {
+			ls[l.Name] = l.Value
 		}
 
-		res.LabelSets = make([]storepb.LabelSet, 0, len(labelSets))
-		for _, v := range labelSets {
-			res.LabelSets = append(res.LabelSets, storepb.LabelSet{Labels: v})
+		res := []storepb.Label{}
+		for k, v := range ls {
+			res = append(res, storepb.Label{Name: k, Value: v})
 		}
+		labelSets[bs.labels.Hash()] = res
+	}
+
+	res.LabelSets = make([]storepb.LabelSet, 0, len(labelSets))
+	for _, v := range labelSets {
+		res.LabelSets = append(res.LabelSets, storepb.LabelSet{Labels: v})
 	}
 
 	return res, nil