Enabled index-header under experimental flag. (#1986)

Enabled it also on all our tests.

Depends on: #1952

Signed-off-by: Bartlomiej Plotka <bwplotka@gmail.com>

cmd/thanos/store.go

@@ -75,6 +75,9 @@ func registerStore(m map[string]setupFunc, app *kingpin.Application) {
 
 	selectorRelabelConf := regSelectorRelabelFlags(cmd)
 
+	enableIndexHeader := cmd.Flag("experimental.enable-index-header", "If true, Store Gateway will recreate index-header instead of index-cache.json for each block. This will replace index-cache.json permanently once it will be out of experimental stage.").
+		Hidden().Default("false").Bool()
+
 	m[component.Store.String()] = func(g *run.Group, logger log.Logger, reg *prometheus.Registry, tracer opentracing.Tracer, debugLogging bool) error {
 		if minTime.PrometheusTimestamp() > maxTime.PrometheusTimestamp() {
 			return errors.Errorf("invalid argument: --min-time '%s' can't be greater than --max-time '%s'",
@@ -109,6 +112,7 @@ func registerStore(m map[string]setupFunc, app *kingpin.Application) {
 			},
 			selectorRelabelConf,
 			*advertiseCompatibilityLabel,
+			*enableIndexHeader,
 		)
 	}
 }
@@ -140,6 +144,7 @@ func runStore(
 	filterConf *store.FilterConfig,
 	selectorRelabelConf *extflag.PathOrContent,
 	advertiseCompatibilityLabel bool,
+	enableIndexHeader bool,
 ) error {
 	// Initiate HTTP listener providing metrics endpoint and readiness/liveness probes.
 	statusProber := prober.New(component, logger, prometheus.WrapRegistererWithPrefix("thanos_", reg))
@@ -214,6 +219,9 @@ func runStore(
 		return errors.Wrap(err, "meta fetcher")
 	}
 
+	if enableIndexHeader {
+		level.Info(logger).Log("msg", "index-header instead of index-cache.json enabled")
+	}
 	bs, err := store.NewBucketStore(
 		logger,
 		reg,
@@ -228,6 +236,7 @@ func runStore(
 		blockSyncConcurrency,
 		filterConf,
 		advertiseCompatibilityLabel,
+		enableIndexHeader,
 	)
 	if err != nil {
 		return errors.Wrap(err, "create object storage store")

pkg/store/bucket.go

@@ -226,6 +226,7 @@ type BucketStore struct {
 	filterConfig             *FilterConfig
 	advLabelSets             []storepb.LabelSet
 	enableCompatibilityLabel bool
+	enableIndexHeader        bool
 }
 
 // NewBucketStore creates a new bucket backed store that implements the store API against
@@ -244,6 +245,7 @@ func NewBucketStore(
 	blockSyncConcurrency int,
 	filterConfig *FilterConfig,
 	enableCompatibilityLabel bool,
+	enableIndexHeader bool,
 ) (*BucketStore, error) {
 	if logger == nil {
 		logger = log.NewNopLogger()
@@ -280,6 +282,7 @@ func NewBucketStore(
 		samplesLimiter:           NewLimiter(maxSampleCount, metrics.queriesDropped),
 		partitioner:              gapBasedPartitioner{maxGapSize: maxGapSize},
 		enableCompatibilityLabel: enableCompatibilityLabel,
+		enableIndexHeader:        enableIndexHeader,
 	}
 	s.metrics = metrics
 
@@ -439,9 +442,17 @@ func (s *BucketStore) addBlock(ctx context.Context, meta *metadata.Meta) (err er
 	lset := labels.FromMap(meta.Thanos.Labels)
 	h := lset.Hash()
 
-	jr, err := indexheader.NewJSONReader(ctx, s.logger, s.bkt, s.dir, meta.ULID)
-	if err != nil {
-		return errors.Wrap(err, "create index header reader")
+	var indexHeaderReader indexheader.Reader
+	if s.enableIndexHeader {
+		indexHeaderReader, err = indexheader.NewBinaryReader(ctx, s.logger, s.bkt, s.dir, meta.ULID)
+		if err != nil {
+			return errors.Wrap(err, "create index header reader")
+		}
+	} else {
+		indexHeaderReader, err = indexheader.NewJSONReader(ctx, s.logger, s.bkt, s.dir, meta.ULID)
+		if err != nil {
+			return errors.Wrap(err, "create index cache reader")
+		}
 	}
 
 	b, err := newBucketBlock(
@@ -452,7 +463,7 @@ func (s *BucketStore) addBlock(ctx context.Context, meta *metadata.Meta) (err er
 		dir,
 		s.indexCache,
 		s.chunkPool,
-		jr,
+		indexHeaderReader,
 		s.partitioner,
 	)
 	if err != nil {

pkg/store/bucket_e2e_test.go

@@ -163,6 +163,7 @@ func prepareStoreWithTestBlocks(t testing.TB, dir string, bkt objstore.Bucket, m
 		20,
 		filterConf,
 		true,
+		true,
 	)
 	testutil.Ok(t, err)
 	s.store = store

pkg/store/bucket_test.go

@@ -447,6 +447,7 @@ func TestBucketStore_Info(t *testing.T) {
 		20,
 		allowAllFilterConf,
 		true,
+		true,
 	)
 	testutil.Ok(t, err)
 
@@ -687,7 +688,9 @@ func testSharding(t *testing.T, reuseDisk string, bkt objstore.Bucket, all ...ul
 				false,
 				20,
 				allowAllFilterConf,
-				true)
+				true,
+				true,
+			)
 			testutil.Ok(t, err)
 
 			testutil.Ok(t, bucketStore.InitialSync(context.Background()))

test/e2e/spinup_test.go

@@ -269,6 +269,7 @@ func storeGateway(http, grpc address, bucketConfig []byte, relabelConfig []byte)
 				// Accelerated sync time for quicker test (3m by default).
 				"--sync-block-duration", "5s",
 				"--selector.relabel-config", string(relabelConfig),
+				"--experimental.enable-index-header",
 			)), nil
 		},
 	}