Avoid large blocks on forced compaction of blocks ingester (#3344)

* Avoid large blocks on forced compaction of blocks ingester

Signed-off-by: Ganesh Vernekar <cs15btech11018@iith.ac.in>

* Add entry in CHANGELOG

Signed-off-by: Ganesh Vernekar <cs15btech11018@iith.ac.in>

* Refactor code and add unit test

Signed-off-by: Ganesh Vernekar <cs15btech11018@iith.ac.in>

* Fix builds

Signed-off-by: Ganesh Vernekar <cs15btech11018@iith.ac.in>

* Fixing nits.

Signed-off-by: Peter Štibraný <peter.stibrany@grafana.com>

* Removed compactBlocksMtx, and use for-loop again in compactHead.

Signed-off-by: Peter Štibraný <peter.stibrany@grafana.com>

* Comment.

Signed-off-by: Peter Štibraný <peter.stibrany@grafana.com>

* Review feedback.

Signed-off-by: Peter Štibraný <peter.stibrany@grafana.com>

Co-authored-by: Peter Štibraný <peter.stibrany@grafana.com>

Author: codesome

CHANGELOG.md

@@ -106,6 +106,7 @@
 * [BUGFIX] Blocks storage: Avoid deletion of blocks in the ingester which are not shipped to the storage yet. #3346
 * [BUGFIX] Fix common prefixes returned by List method of S3 client. #3358
 * [BUGFIX] Honor configured timeout in Azure and GCS object clients. #3285
+* [BUGFIX] Blocks storage: Avoid creating blocks larger than configured block range period on forced compaction and when TSDB is idle. #3344
 * [BUGFIX] Shuffle sharding: fixed max global series per user/metric limit when shuffle sharding and `-distributor.shard-by-all-labels=true` are both enabled in distributor. When using these global limits you should now set `-distributor.sharding-strategy` and `-distributor.zone-awareness-enabled` to ingesters too. #3369
 * [BUGFIX] Slow query logging: when using downstream server request parameters were not logged. #3276
 * [BUGFIX] Fixed tenant detection in the ruler and alertmanager API when running without auth. #3343

pkg/ingester/ingester_v2.go

@@ -48,7 +48,7 @@ type Shipper interface {
 }
 
 type userTSDB struct {
-	*tsdb.DB
+	db             *tsdb.DB
 	userID         string
 	refCache       *cortex_tsdb.RefCache
 	activeSeries   *ActiveSeries
@@ -66,14 +66,61 @@ type userTSDB struct {
 	ingestedRuleSamples *ewmaRate
 }
 
+// Explicitly wrapping the tsdb.DB functions that we use.
+
+func (u *userTSDB) Appender(ctx context.Context) storage.Appender {
+	return u.db.Appender(ctx)
+}
+
+func (u *userTSDB) Querier(ctx context.Context, mint, maxt int64) (storage.Querier, error) {
+	return u.db.Querier(ctx, mint, maxt)
+}
+
+func (u *userTSDB) Head() *tsdb.Head {
+	return u.db.Head()
+}
+
+func (u *userTSDB) Blocks() []*tsdb.Block {
+	return u.db.Blocks()
+}
+
+func (u *userTSDB) Close() error {
+	return u.db.Close()
+}
+
+func (u *userTSDB) Compact() error {
+	return u.db.Compact()
+}
+
+// compactHead compacts the Head block at specified block durations avoiding a single huge block.
+func (u *userTSDB) compactHead(blockDuration int64) error {
+	h := u.Head()
+
+	minTime, maxTime := h.MinTime(), h.MaxTime()
+
+	for (minTime/blockDuration)*blockDuration != (maxTime/blockDuration)*blockDuration {
+		// Data in Head spans across multiple block ranges, so we break it into blocks here.
+		// Block max time is exclusive, so we do a -1 here.
+		blockMaxTime := ((minTime/blockDuration)+1)*blockDuration - 1
+		if err := u.db.CompactHead(tsdb.NewRangeHead(h, minTime, blockMaxTime)); err != nil {
+			return err
+		}
+
+		// Get current min/max times after compaction.
+		minTime, maxTime = h.MinTime(), h.MaxTime()
+	}
+
+	return u.db.CompactHead(tsdb.NewRangeHead(h, minTime, maxTime))
+}
+
 // PreCreation implements SeriesLifecycleCallback interface.
 func (u *userTSDB) PreCreation(metric labels.Labels) error {
 	if u.limiter == nil {
 		return nil
 	}
 
 	// Total series limit.
-	if err := u.limiter.AssertMaxSeriesPerUser(u.userID, int(u.DB.Head().NumSeries())); err != nil {
+	if err := u.limiter.AssertMaxSeriesPerUser(u.userID, int(u.Head().NumSeries())); err != nil {
 		return makeLimitError(perUserSeriesLimit, err)
 	}
 
@@ -113,15 +160,15 @@ func (u *userTSDB) PostDeletion(metrics ...labels.Labels) {
 
 // blocksToDelete filters the input blocks and returns the blocks which are safe to be deleted from the ingester.
 func (u *userTSDB) blocksToDelete(blocks []*tsdb.Block) map[ulid.ULID]struct{} {
-	if u.DB == nil {
+	if u.db == nil {
 		return nil
 	}
-	deletable := tsdb.DefaultBlocksToDelete(u.DB)(blocks)
+	deletable := tsdb.DefaultBlocksToDelete(u.db)(blocks)
 	if u.shipper == nil {
 		return deletable
 	}
 
-	shipperMeta, err := shipper.ReadMetaFile(u.Dir())
+	shipperMeta, err := shipper.ReadMetaFile(u.db.Dir())
 	if err != nil {
 		// If there is any issue with the shipper, we should be conservative and not delete anything.
 		level.Error(util.Logger).Log("msg", "failed to read shipper meta during deletion of blocks", "user", u.userID, "err", err)
@@ -1022,7 +1069,7 @@ func (i *Ingester) createTSDB(userID string) (*userTSDB, error) {
 		return nil, errors.Wrapf(err, "failed to compact TSDB: %s", udir)
 	}
 
-	userDB.DB = db
+	userDB.db = db
 	// We set the limiter here because we don't want to limit
 	// series during WAL replay.
 	userDB.limiter = i.limiter
@@ -1313,12 +1360,12 @@ func (i *Ingester) compactBlocks(ctx context.Context, force bool) {
 		switch {
 		case force:
 			reason = "forced"
-			err = userDB.CompactHead(tsdb.NewRangeHead(h, h.MinTime(), h.MaxTime()))
+			err = userDB.compactHead(i.cfg.BlocksStorageConfig.TSDB.BlockRanges[0].Milliseconds())
 
 		case i.cfg.BlocksStorageConfig.TSDB.HeadCompactionIdleTimeout > 0 && userDB.isIdle(time.Now(), i.cfg.BlocksStorageConfig.TSDB.HeadCompactionIdleTimeout):
 			reason = "idle"
 			level.Info(util.Logger).Log("msg", "TSDB is idle, forcing compaction", "user", userID)
-			err = userDB.CompactHead(tsdb.NewRangeHead(h, h.MinTime(), h.MaxTime()))
+			err = userDB.compactHead(i.cfg.BlocksStorageConfig.TSDB.BlockRanges[0].Milliseconds())
 
 		default:
 			reason = "regular"

pkg/ingester/ingester_v2_test.go

@@ -1806,7 +1806,7 @@ func TestIngester_flushing(t *testing.T) {
 				// Nothing shipped yet.
 				m.AssertNumberOfCalls(t, "Sync", 0)
 
-				i.FlushHandler(httptest.NewRecorder(), httptest.NewRequest("POST", "/shutdown", nil))
+				i.FlushHandler(httptest.NewRecorder(), httptest.NewRequest("POST", "/flush", nil))
 
 				// Flush handler only triggers compactions, but doesn't wait for them to finish. Let's wait for a moment, and then verify.
 				time.Sleep(1 * time.Second)
@@ -1815,6 +1815,59 @@ func TestIngester_flushing(t *testing.T) {
 				m.AssertNumberOfCalls(t, "Sync", 1)
 			},
 		},
+
+		"flushMultipleBlocksWithDataSpanning3Days": {
+			setupIngester: func(cfg *Config) {
+				cfg.BlocksStorageConfig.TSDB.FlushBlocksOnShutdown = false
+			},
+
+			action: func(t *testing.T, i *Ingester, m *shipperMock) {
+				// Pushing 5 samples, spanning over 3 days.
+				// First block
+				pushSingleSampleAtTime(t, i, 23*time.Hour.Milliseconds())
+				pushSingleSampleAtTime(t, i, 24*time.Hour.Milliseconds()-1)
+
+				// Second block
+				pushSingleSampleAtTime(t, i, 24*time.Hour.Milliseconds()+1)
+				pushSingleSampleAtTime(t, i, 25*time.Hour.Milliseconds())
+
+				// Third block, far in the future.
+				pushSingleSampleAtTime(t, i, 50*time.Hour.Milliseconds())
+
+				// Nothing shipped yet.
+				m.AssertNumberOfCalls(t, "Sync", 0)
+
+				i.FlushHandler(httptest.NewRecorder(), httptest.NewRequest("POST", "/flush", nil))
+
+				// Wait for compaction to finish.
+				test.Poll(t, 5*time.Second, true, func() interface{} {
+					db := i.getTSDB(userID)
+					if db == nil {
+						return false
+					}
+
+					h := db.Head()
+					return h.NumSeries() == 0
+				})
+
+				m.AssertNumberOfCalls(t, "Sync", 1)
+
+				userDB := i.getTSDB(userID)
+				require.NotNil(t, userDB)
+
+				blocks := userDB.Blocks()
+				require.Equal(t, 3, len(blocks))
+				require.Equal(t, 23*time.Hour.Milliseconds(), blocks[0].Meta().MinTime)
+				require.Equal(t, 24*time.Hour.Milliseconds(), blocks[0].Meta().MaxTime) // Block maxt is exclusive.
+
+				// Even though we added 24*time.Hour.Milliseconds()+1, the Head compaction
+				// will leave Head's mint to 24*time.Hour.Milliseconds(). Hence the block mint.
+				require.Equal(t, 24*time.Hour.Milliseconds(), blocks[1].Meta().MinTime)
+				require.Equal(t, 26*time.Hour.Milliseconds(), blocks[1].Meta().MaxTime)
+
+				require.Equal(t, 50*time.Hour.Milliseconds()+1, blocks[2].Meta().MaxTime) // Block maxt is exclusive.
+			},
+		},
 	} {
 		t.Run(name, func(t *testing.T) {
 			cfg := defaultIngesterTestConfig()
@@ -2109,6 +2162,13 @@ func pushSingleSample(t *testing.T, i *Ingester) {
 	require.NoError(t, err)
 }
 
+func pushSingleSampleAtTime(t *testing.T, i *Ingester, ts int64) {
+	ctx := user.InjectOrgID(context.Background(), userID)
+	req, _, _ := mockWriteRequest(labels.Labels{{Name: labels.MetricName, Value: "test"}}, 0, ts)
+	_, err := i.v2Push(ctx, req)
+	require.NoError(t, err)
+}
+
 func TestHeadCompactionOnStartup(t *testing.T) {
 	// Create a temporary directory for TSDB
 	tempDir, err := ioutil.TempDir("", "tsdb")
@@ -2255,7 +2315,7 @@ func TestIngesterNotDeleteUnshippedBlocks(t *testing.T) {
 	require.Equal(t, 3, len(oldBlocks))
 
 	// Saying that we have shipped the second block, so only that should get deleted.
-	require.Nil(t, shipper.WriteMetaFile(nil, db.Dir(), &shipper.Meta{
+	require.Nil(t, shipper.WriteMetaFile(nil, db.db.Dir(), &shipper.Meta{
 		Version:  shipper.MetaVersion1,
 		Uploaded: []ulid.ULID{oldBlocks[1].Meta().ULID},
 	}))
@@ -2276,7 +2336,7 @@ func TestIngesterNotDeleteUnshippedBlocks(t *testing.T) {
 	require.NotEqual(t, oldBlocks[1].Meta().ULID, newBlocks[2].Meta().ULID) // The new block won't match previous 2nd block.
 
 	// Shipping 2 more blocks, hence all the blocks from first round.
-	require.Nil(t, shipper.WriteMetaFile(nil, db.Dir(), &shipper.Meta{
+	require.Nil(t, shipper.WriteMetaFile(nil, db.db.Dir(), &shipper.Meta{
 		Version:  shipper.MetaVersion1,
 		Uploaded: []ulid.ULID{oldBlocks[1].Meta().ULID, newBlocks[0].Meta().ULID, newBlocks[1].Meta().ULID},
 	}))