Function TestDelayedCompaction

tsdb/compact_test.go:1989–2111

(t *testing.T)

}

func TestDelayedCompaction(t *testing.T) {
	delay := 5 * time.Second
	label := labels.FromStrings("foo", "bar")

	appendSamples := func(t *testing.T, db *DB, timestamps ...int64) {
		app := db.Appender(context.Background())
		for _, ts := range timestamps {
			_, err := app.Append(0, label, ts, 0)
			require.NoError(t, err)
		}
		require.NoError(t, app.Commit())
	}

	compactorRanCount := func(db *DB) float64 {
		return prom_testutil.ToFloat64(db.compactor.(*LeveledCompactor).metrics.Ran)
	}

	t.Run("delay not enabled", func(t *testing.T) {
		t.Parallel()
		db := newTestDB(t, withRngs(10))

		compactAndCheck := func() {
			start := time.Now()
			db.Compact(context.Background())
			require.False(t, db.head.compactable())
			require.Less(t, time.Since(start), delay)
		}

		db.DisableCompactions()
		appendSamples(t, db, 0, 11, 21)
		compactAndCheck()
		require.Equal(t, 1.0, compactorRanCount(db))

		db.DisableCompactions()
		appendSamples(t, db, 31, 41)
		compactAndCheck()
		require.Equal(t, 3.0, compactorRanCount(db))
	})

	t.Run("delay enabled", func(t *testing.T) {
		synctest.Test(t, func(t *testing.T) {
			db := newTestDB(t, withOpts(&Options{CompactionDelay: delay}), withRngs(10))

			getDelayStart := func() time.Time {
				db.cmtx.Lock()
				defer db.cmtx.Unlock()
				return db.timeWhenCompactionDelayStarted
			}

			// triggerAutoCompaction sends a compaction trigger and waits for it to be processed.
			triggerAutoCompaction := func() {
				db.compactc <- struct{}{}
				synctest.Wait()
			}

			// First compaction: expect 1 block.
			db.DisableCompactions()
			appendSamples(t, db, 0, 11, 21)