Function TestDelayedCompactionDoesNotBlockUnrelatedOps

tsdb/compact_test.go:2115–2173

TestDelayedCompactionDoesNotBlockUnrelatedOps makes sure that when delayed compaction is enabled, operations that don't directly derive from the Head compaction are not delayed, here we consider disk blocks compaction.

(t *testing.T)

// TestDelayedCompactionDoesNotBlockUnrelatedOps makes sure that when delayed compaction is enabled,
// operations that don't directly derive from the Head compaction are not delayed, here we consider disk blocks compaction.
func TestDelayedCompactionDoesNotBlockUnrelatedOps(t *testing.T) {
	t.Parallel()
	cases := []struct {
		name            string
		whenCompactable bool
	}{
		{
			"Head is compactable",
			true,
		},
		{
			"Head is not compactable",
			false,
		},
	}

	for _, c := range cases {
		t.Run(c.name, func(t *testing.T) {
			t.Parallel()

			tmpdir := t.TempDir()
			// Some blocks that need compaction are present.
			createBlock(t, tmpdir, genSeries(1, 1, 0, 100))
			createBlock(t, tmpdir, genSeries(1, 1, 100, 200))
			createBlock(t, tmpdir, genSeries(1, 1, 200, 300))

			options := DefaultOptions()
			// This will make the test timeout if compaction really waits for it.
			options.CompactionDelay = time.Hour
			db, err := open(tmpdir, promslog.NewNopLogger(), nil, options, []int64{10, 200}, nil)
			require.NoError(t, err)
			defer func() {
				require.NoError(t, db.Close())
			}()

			db.DisableCompactions()
			require.Len(t, db.Blocks(), 3)

			if c.whenCompactable {
				label := labels.FromStrings("foo", "bar")
				app := db.Appender(context.Background())
				_, err := app.Append(0, label, 301, 0)
				require.NoError(t, err)
				_, err = app.Append(0, label, 317, 0)
				require.NoError(t, err)
				require.NoError(t, app.Commit())
				// The Head is compactable and will still be at the end.
				require.True(t, db.head.compactable())
				defer func() {
					require.True(t, db.head.compactable())
				}()
			}

			// The blocks were compacted.
			db.Compact(context.Background())
			require.Len(t, db.Blocks(), 2)
		})
	}