Function TestDeleteCompactionBlockAfterFailedReload

tsdb/compact_test.go:1447–1516

TestDeleteCompactionBlockAfterFailedReload ensures that a failed reloadBlocks immediately after a compaction deletes the resulting block to avoid creating blocks with the same time range.

(t *testing.T)

// TestDeleteCompactionBlockAfterFailedReload ensures that a failed reloadBlocks immediately after a compaction
// deletes the resulting block to avoid creating blocks with the same time range.
func TestDeleteCompactionBlockAfterFailedReload(t *testing.T) {
	t.Parallel()
	tests := map[string]func(*DB) int{
		"Test Head Compaction": func(db *DB) int {
			rangeToTriggerCompaction := db.compactor.(*LeveledCompactor).ranges[0]/2*3 - 1
			defaultLabel := labels.FromStrings("foo", "bar")

			// Add some data to the head that is enough to trigger a compaction.
			app := db.Appender(context.Background())
			_, err := app.Append(0, defaultLabel, 1, 0)
			require.NoError(t, err)
			_, err = app.Append(0, defaultLabel, 2, 0)
			require.NoError(t, err)
			_, err = app.Append(0, defaultLabel, 3+rangeToTriggerCompaction, 0)
			require.NoError(t, err)
			require.NoError(t, app.Commit())

			return 0
		},
		"Test Block Compaction": func(db *DB) int {
			blocks := []*BlockMeta{
				{MinTime: 0, MaxTime: 100},
				{MinTime: 100, MaxTime: 150},
				{MinTime: 150, MaxTime: 200},
			}
			for _, m := range blocks {
				createBlock(t, db.Dir(), genSeries(1, 1, m.MinTime, m.MaxTime))
			}
			require.NoError(t, db.reload())
			require.Len(t, db.Blocks(), len(blocks), "unexpected block count after a reloadBlocks")

			return len(blocks)
		},
	}

	for title, bootStrap := range tests {
		t.Run(title, func(t *testing.T) {
			ctx := context.Background()

			db := newTestDB(t, withRngs(1, 100))
			db.DisableCompactions()

			expBlocks := bootStrap(db)

			// Create a block that will trigger the reloadBlocks to fail.
			blockPath := createBlock(t, db.Dir(), genSeries(1, 1, 200, 300))
			lastBlockIndex := path.Join(blockPath, indexFilename)
			actBlocks, err := blockDirs(db.Dir())
			require.NoError(t, err)
			require.Equal(t, expBlocks, len(actBlocks)-1)    // -1 to exclude the corrupted block.
			require.NoError(t, os.RemoveAll(lastBlockIndex)) // Corrupt the block by removing the index file.

			require.Equal(t, 0.0, prom_testutil.ToFloat64(db.metrics.reloadsFailed), "initial 'failed db reloadBlocks' count metrics mismatch")
			require.Equal(t, 0.0, prom_testutil.ToFloat64(db.compactor.(*LeveledCompactor).metrics.Ran), "initial `compactions` count metric mismatch")
			require.Equal(t, 0.0, prom_testutil.ToFloat64(db.metrics.compactionsFailed), "initial `compactions failed` count metric mismatch")

			// Do the compaction and check the metrics.
			// Compaction should succeed, but the reloadBlocks should fail and