upstream/prometheus
1
0
Fork 0
mirror of https://github.com/prometheus/prometheus.git synced 2026-07-16 12:33:18 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions 5
prometheus/tsdb/compact_test.go
Arve Knudsen 9caf2ca7b1
tsdb: optimize head chunk access paths (#18300) 
* tsdb: optimize head chunk access paths

Replace O(n²) linked-list traversals with O(n) or O(1) alternatives in
head chunk operations.

* tsdb: enable head-chunk cache during compaction

---------

Signed-off-by: Arve Knudsen <arve.knudsen@gmail.com>
Co-authored-by: George Krajcsovits <krajorama@users.noreply.github.com>
Co-authored-by: György Krajcsovits <gyorgy.krajcsovits@grafana.com>
2026-07-15 16:03:35 +02:00

2593 lines
83 KiB
Go
Raw Permalink Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

// Copyright The Prometheus Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package tsdb
import (
"context"
"errors"
"fmt"
"log/slog"
"math"
"math/rand"
"os"
"path"
"path/filepath"
"strconv"
"sync"
"testing"
"testing/synctest"
"time"
"github.com/oklog/ulid/v2"
"github.com/prometheus/client_golang/prometheus"
prom_testutil "github.com/prometheus/client_golang/prometheus/testutil"
"github.com/prometheus/common/promslog"
"github.com/stretchr/testify/require"
"github.com/prometheus/prometheus/model/histogram"
"github.com/prometheus/prometheus/model/labels"
"github.com/prometheus/prometheus/storage"
"github.com/prometheus/prometheus/tsdb/chunkenc"
"github.com/prometheus/prometheus/tsdb/chunks"
"github.com/prometheus/prometheus/tsdb/fileutil"
"github.com/prometheus/prometheus/tsdb/index"
"github.com/prometheus/prometheus/tsdb/tombstones"
"github.com/prometheus/prometheus/tsdb/tsdbutil"
"github.com/prometheus/prometheus/util/compression"
)
func TestSplitByRange(t *testing.T) {
cases := []struct {
trange int64
ranges [][2]int64
output [][][2]int64
}{
{
trange: 60,
ranges: [][2]int64{{0, 10}},
output: [][][2]int64{
{{0, 10}},
},
},
{
trange: 60,
ranges: [][2]int64{{0, 60}},
output: [][][2]int64{
{{0, 60}},
},
},
{
trange: 60,
ranges: [][2]int64{{0, 10}, {9, 15}, {30, 60}},
output: [][][2]int64{
{{0, 10}, {9, 15}, {30, 60}},
},
},
{
trange: 60,
ranges: [][2]int64{{70, 90}, {125, 130}, {130, 180}, {1000, 1001}},
output: [][][2]int64{
{{70, 90}},
{{125, 130}, {130, 180}},
{{1000, 1001}},
},
},
// Mis-aligned or too-large blocks are ignored.
{
trange: 60,
ranges: [][2]int64{{50, 70}, {70, 80}},
output: [][][2]int64{
{{70, 80}},
},
},
{
trange: 72,
ranges: [][2]int64{{0, 144}, {144, 216}, {216, 288}},
output: [][][2]int64{
{{144, 216}},
{{216, 288}},
},
},
// Various awkward edge cases easy to hit with negative numbers.
{
trange: 60,
ranges: [][2]int64{{-10, -5}},
output: [][][2]int64{
{{-10, -5}},
},
},
{
trange: 60,
ranges: [][2]int64{{-60, -50}, {-10, -5}},
output: [][][2]int64{
{{-60, -50}, {-10, -5}},
},
},
{
trange: 60,
ranges: [][2]int64{{-60, -50}, {-10, -5}, {0, 15}},
output: [][][2]int64{
{{-60, -50}, {-10, -5}},
{{0, 15}},
},
},
}
for _, c := range cases {
// Transform input range tuples into dirMetas.
blocks := make([]dirMeta, 0, len(c.ranges))
for _, r := range c.ranges {
blocks = append(blocks, dirMeta{
meta: &BlockMeta{
MinTime: r[0],
MaxTime: r[1],
},
})
}
// Transform output range tuples into dirMetas.
exp := make([][]dirMeta, len(c.output))
for i, group := range c.output {
for _, r := range group {
exp[i] = append(exp[i], dirMeta{
meta: &BlockMeta{MinTime: r[0], MaxTime: r[1]},
})
}
}
require.Equal(t, exp, splitByRange(blocks, c.trange))
}
}
// See https://github.com/prometheus/prometheus/issues/3064
func TestNoPanicFor0Tombstones(t *testing.T) {
metas := []dirMeta{
{
dir: "1",
meta: &BlockMeta{
MinTime: 0,
MaxTime: 100,
},
},
{
dir: "2",
meta: &BlockMeta{
MinTime: 101,
MaxTime: 200,
},
},
}
c, err := NewLeveledCompactor(context.Background(), nil, nil, []int64{50}, nil, nil)
require.NoError(t, err)
c.plan(metas)
}
// staleMetaRange is like metaRange but tags the block with the from-stale-series hint.
func staleMetaRange(name string, mint, maxt int64) dirMeta {
dm := metaRange(name, mint, maxt, nil)
dm.meta.Compaction.SetStaleSeries()
return dm
}
// TestPlanDoesNotMergeStaleAndNonStaleBlocks verifies that the planner never
// groups a from-stale-series block with non-stale blocks. Mixing them would
// drop the hint on the merged block, which would then wrongly count towards
// inOrderBlocksMaxTime and advance the WAL-replay cutoff past WAL-only data.
// See https://github.com/prometheus/prometheus/issues/18379.
func TestPlanDoesNotMergeStaleAndNonStaleBlocks(t *testing.T) {
compactor, err := NewLeveledCompactor(context.Background(), nil, promslog.NewNopLogger(), []int64{
20,
60,
180,
}, nil, nil)
require.NoError(t, err)
t.Run("stale and non-stale in the same range are not merged", func(t *testing.T) {
// Two non-stale blocks and one stale block all fall into the same 60-wide
// range [0,60). Without segregation the planner would return all three;
// with segregation it must only return the two non-stale blocks.
metas := []dirMeta{
metaRange("1", 0, 20, nil),
staleMetaRange("stale", 0, 20),
metaRange("2", 20, 40, nil),
metaRange("3", 40, 60, nil),
// A fresh block so the last (most recent) block is excluded as usual.
metaRange("fresh", 60, 80, nil),
}
res, err := compactor.plan(metas)
require.NoError(t, err)
require.NotContains(t, res, "stale", "stale block must not be merged with non-stale blocks")
require.Equal(t, []string{"1", "2", "3"}, res)
})
t.Run("stale blocks are merged together", func(t *testing.T) {
// Three stale blocks fill the 60-wide range [0,60) and a fourth stale block
// makes that range no longer the most recent, so the three are compacted
// together: stale data still gets compacted, just never mixed with non-stale.
metas := []dirMeta{
staleMetaRange("s1", 0, 20),
staleMetaRange("s2", 20, 40),
staleMetaRange("s3", 40, 60),
staleMetaRange("s4", 60, 80),
}
res, err := compactor.plan(metas)
require.NoError(t, err)
require.Equal(t, []string{"s1", "s2", "s3"}, res)
})
t.Run("stale blocks are planned when there is nothing to do for non-stale", func(t *testing.T) {
// A single non-stale block (nothing to merge) plus a full stale range.
// The non-stale class yields nothing, so the stale class must be planned.
metas := []dirMeta{
metaRange("1", 0, 20, nil),
staleMetaRange("s1", 0, 20),
staleMetaRange("s2", 20, 40),
staleMetaRange("s3", 40, 60),
staleMetaRange("s4", 60, 80),
}
res, err := compactor.plan(metas)
require.NoError(t, err)
require.Equal(t, []string{"s1", "s2", "s3"}, res)
})
}
func TestLeveledCompactor(t *testing.T) {
// Tests for the private plan() method.
t.Run("plan", func(t *testing.T) {
// This mimics our default ExponentialBlockRanges with min block size equals to 20.
compactor, err := NewLeveledCompactor(context.Background(), nil, nil, []int64{
20,
60,
180,
540,
1620,
}, nil, nil)
require.NoError(t, err)
cases := map[string]struct {
metas []dirMeta
expected []string
}{
"Outside Range": {
metas: []dirMeta{
metaRange("1", 0, 20, nil),
},
expected: nil,
},
"We should wait for four blocks of size 20 to appear before compacting.": {
metas: []dirMeta{
metaRange("1", 0, 20, nil),
metaRange("2", 20, 40, nil),
},
expected: nil,
},
`We should wait for a next block of size 20 to appear before compacting
the existing ones. We have three, but we ignore the fresh one from WAl`: {
metas: []dirMeta{
metaRange("1", 0, 20, nil),
metaRange("2", 20, 40, nil),
metaRange("3", 40, 60, nil),
},
expected: nil,
},
"Block to fill the entire parent range appeared – should be compacted": {
metas: []dirMeta{
metaRange("1", 0, 20, nil),
metaRange("2", 20, 40, nil),
metaRange("3", 40, 60, nil),
metaRange("4", 60, 80, nil),
},
expected: []string{"1", "2", "3"},
},
`Block for the next parent range appeared with gap with size 20. Nothing will happen in the first one
anymore but we ignore fresh one still, so no compaction`: {
metas: []dirMeta{
metaRange("1", 0, 20, nil),
metaRange("2", 20, 40, nil),
metaRange("3", 60, 80, nil),
},
expected: nil,
},
`Block for the next parent range appeared, and we have a gap with size 20 between second and third block.
We will not get this missed gap anymore and we should compact just these two.`: {
metas: []dirMeta{
metaRange("1", 0, 20, nil),
metaRange("2", 20, 40, nil),
metaRange("3", 60, 80, nil),
metaRange("4", 80, 100, nil),
},
expected: []string{"1", "2"},
},
"We have 20, 20, 20, 60, 60 range blocks. '5' is marked as fresh one": {
metas: []dirMeta{
metaRange("1", 0, 20, nil),
metaRange("2", 20, 40, nil),
metaRange("3", 40, 60, nil),
metaRange("4", 60, 120, nil),
metaRange("5", 120, 180, nil),
},
expected: []string{"1", "2", "3"},
},
"We have 20, 60, 20, 60, 240 range blocks. We can compact 20 + 60 + 60": {
metas: []dirMeta{
metaRange("2", 20, 40, nil),
metaRange("4", 60, 120, nil),
metaRange("5", 960, 980, nil), // Fresh one.
metaRange("6", 120, 180, nil),
metaRange("7", 720, 960, nil),
},
expected: []string{"2", "4", "6"},
},
"Do not select large blocks that have many tombstones when there is no fresh block": {
metas: []dirMeta{
metaRange("1", 0, 540, &BlockStats{
NumSeries: 10,
NumTombstones: 3,
}),
},
expected: nil,
},
"Select large blocks that have many tombstones when fresh appears": {
metas: []dirMeta{
metaRange("1", 0, 540, &BlockStats{
NumSeries: 10,
NumTombstones: 3,
}),
metaRange("2", 540, 560, nil),
},
expected: []string{"1"},
},
"For small blocks, do not compact tombstones, even when fresh appears.": {
metas: []dirMeta{
metaRange("1", 0, 60, &BlockStats{
NumSeries: 10,
NumTombstones: 3,
}),
metaRange("2", 60, 80, nil),
},
expected: nil,
},
`Regression test: we were stuck in a compact loop where we always recompacted
the same block when tombstones and series counts were zero`: {
metas: []dirMeta{
metaRange("1", 0, 540, &BlockStats{
NumSeries: 0,
NumTombstones: 0,
}),
metaRange("2", 540, 560, nil),
},
expected: nil,
},
`Regression test: we were wrongly assuming that new block is fresh from WAL when its ULID is newest.
We need to actually look on max time instead.
With previous, wrong approach "8" block was ignored, so we were wrongly compacting 5 and 7 and introducing
block overlaps`: {
metas: []dirMeta{
metaRange("5", 0, 360, nil),
metaRange("6", 540, 560, nil), // Fresh one.
metaRange("7", 360, 420, nil),
metaRange("8", 420, 540, nil),
},
expected: []string{"7", "8"},
},
// |--------------|
// |----------------|
// |--------------|
"Overlapping blocks 1": {
metas: []dirMeta{
metaRange("1", 0, 20, nil),
metaRange("2", 19, 40, nil),
metaRange("3", 40, 60, nil),
},
expected: []string{"1", "2"},
},
// |--------------|
// |--------------|
// |--------------|
"Overlapping blocks 2": {
metas: []dirMeta{
metaRange("1", 0, 20, nil),
metaRange("2", 20, 40, nil),
metaRange("3", 30, 50, nil),
},
expected: []string{"2", "3"},
},
// |--------------|
// |---------------------|
// |--------------|
"Overlapping blocks 3": {
metas: []dirMeta{
metaRange("1", 0, 20, nil),
metaRange("2", 10, 40, nil),
metaRange("3", 30, 50, nil),
},
expected: []string{"1", "2", "3"},
},
// |--------------|
// |--------------------------------|
// |--------------|
// |--------------|
"Overlapping blocks 4": {
metas: []dirMeta{
metaRange("5", 0, 360, nil),
metaRange("6", 340, 560, nil),
metaRange("7", 360, 420, nil),
metaRange("8", 420, 540, nil),
},
expected: []string{"5", "6", "7", "8"},
},
// |--------------|
// |--------------|
// |--------------|
// |--------------|
"Overlapping blocks 5": {
metas: []dirMeta{
metaRange("1", 0, 10, nil),
metaRange("2", 9, 20, nil),
metaRange("3", 30, 40, nil),
metaRange("4", 39, 50, nil),
},
expected: []string{"1", "2"},
},
}
for title, c := range cases {
if !t.Run(title, func(t *testing.T) {
res, err := compactor.plan(c.metas)
require.NoError(t, err)
require.Equal(t, c.expected, res)
}) {
return
}
}
})
// Tests for the public Plan() method.
t.Run("Plan", func(t *testing.T) {
// Verify that when a BlockExcludeFilter excludes a block in the middle of
// the list, subsequent blocks are not processed.
t.Run("BlockExcludeFilter stops iteration", func(t *testing.T) {
dir := t.TempDir()
// Create 4 blocks with sequential ULIDs.
block1ULID := ulid.MustNew(1, nil)
block2ULID := ulid.MustNew(2, nil)
block3ULID := ulid.MustNew(3, nil)
block4ULID := ulid.MustNew(4, nil)
for i, uid := range []ulid.ULID{block1ULID, block2ULID, block3ULID, block4ULID} {
blockDir := filepath.Join(dir, uid.String())
require.NoError(t, os.MkdirAll(blockDir, 0o777))
meta := &BlockMeta{
ULID: uid,
MinTime: int64(i * 10),
MaxTime: int64((i + 1) * 10),
}
meta.Compaction.Level = 1
_, err := writeMetaFile(promslog.NewNopLogger(), blockDir, meta)
require.NoError(t, err)
}
// Track which blocks were evaluated by the exclude function.
var evaluatedBlocks []ulid.ULID
excludeFunc := func(meta *BlockMeta) bool {
evaluatedBlocks = append(evaluatedBlocks, meta.ULID)
return meta.ULID == block2ULID
}
c, err := NewLeveledCompactorWithOptions(
context.Background(),
nil,
promslog.NewNopLogger(),
[]int64{20},
chunkenc.NewPool(),
LeveledCompactorOptions{
BlockExcludeFilter: excludeFunc,
EnableOverlappingCompaction: true,
},
)
require.NoError(t, err)
// Plan should evaluate all blocks.
_, err = c.Plan(dir)
require.NoError(t, err)
require.Len(t, evaluatedBlocks, 2, "Expected only 2 blocks to be evaluated")
require.Contains(t, evaluatedBlocks, block1ULID)
require.Contains(t, evaluatedBlocks, block2ULID)
})
})
}
func TestRangeWithFailedCompactionWontGetSelected(t *testing.T) {
compactor, err := NewLeveledCompactor(context.Background(), nil, nil, []int64{
20,
60,
240,
720,
2160,
}, nil, nil)
require.NoError(t, err)
cases := []struct {
metas []dirMeta
}{
{
metas: []dirMeta{
metaRange("1", 0, 20, nil),
metaRange("2", 20, 40, nil),
metaRange("3", 40, 60, nil),
metaRange("4", 60, 80, nil),
},
},
{
metas: []dirMeta{
metaRange("1", 0, 20, nil),
metaRange("2", 20, 40, nil),
metaRange("3", 60, 80, nil),
metaRange("4", 80, 100, nil),
},
},
{
metas: []dirMeta{
metaRange("1", 0, 20, nil),
metaRange("2", 20, 40, nil),
metaRange("3", 40, 60, nil),
metaRange("4", 60, 120, nil),
metaRange("5", 120, 180, nil),
metaRange("6", 180, 200, nil),
},
},
}
for _, c := range cases {
c.metas[1].meta.Compaction.Failed = true
res, err := compactor.plan(c.metas)
require.NoError(t, err)
require.Equal(t, []string(nil), res)
}
}
func TestCompactionFailWillCleanUpTempDir(t *testing.T) {
compactor, err := NewLeveledCompactor(context.Background(), nil, promslog.NewNopLogger(), []int64{
20,
60,
240,
720,
2160,
}, nil, nil)
require.NoError(t, err)
tmpdir := t.TempDir()
require.Error(t, compactor.write(tmpdir, &BlockMeta{}, DefaultBlockPopulator{}, erringBReader{}))
_, err = os.Stat(filepath.Join(tmpdir, BlockMeta{}.ULID.String()) + tmpForCreationBlockDirSuffix)
require.True(t, os.IsNotExist(err), "directory is not cleaned up")
}
func metaRange(name string, mint, maxt int64, stats *BlockStats) dirMeta {
meta := &BlockMeta{MinTime: mint, MaxTime: maxt}
if stats != nil {
meta.Stats = *stats
}
return dirMeta{
dir: name,
meta: meta,
}
}
type erringBReader struct{}
func (erringBReader) Index() (IndexReader, error) { return nil, errors.New("index") }
func (erringBReader) Chunks() (ChunkReader, error) { return nil, errors.New("chunks") }
func (erringBReader) Tombstones() (tombstones.Reader, error) { return nil, errors.New("tombstones") }
func (erringBReader) Meta() BlockMeta { return BlockMeta{} }
func (erringBReader) Size() int64 { return 0 }
type nopChunkWriter struct{}
func (nopChunkWriter) WriteChunks(...chunks.Meta) error { return nil }
func (nopChunkWriter) Close() error { return nil }
func samplesForRange(minTime, maxTime int64, maxSamplesPerChunk int) (ret [][]sample) {
var curr []sample
for i := minTime; i <= maxTime; i++ {
curr = append(curr, sample{t: i})
if len(curr) >= maxSamplesPerChunk {
ret = append(ret, curr)
curr = []sample{}
}
}
if len(curr) > 0 {
ret = append(ret, curr)
}
return ret
}
func TestCompaction_populateBlock(t *testing.T) {
for _, tc := range []struct {
title string
inputSeriesSamples [][]seriesSamples
compactMinTime int64
compactMaxTime int64 // When not defined the test runner sets a default of math.MaxInt64.
irPostingsFunc IndexReaderPostingsFunc
expSeriesSamples []seriesSamples
expErr error
}{
{
title: "Populate block from empty input should return error.",
inputSeriesSamples: [][]seriesSamples{},
expErr: errors.New("cannot populate block from no readers"),
},
{
// Populate from single block without chunks. We expect these kind of series being ignored.
inputSeriesSamples: [][]seriesSamples{
{{lset: map[string]string{"a": "b"}}},
},
},
{
title: "Populate from single block. We expect the same samples at the output.",
inputSeriesSamples: [][]seriesSamples{
{
{
lset: map[string]string{"a": "b"},
chunks: [][]sample{{{t: 0}, {t: 10}}, {{t: 11}, {t: 20}}},
},
},
},
expSeriesSamples: []seriesSamples{
{
lset: map[string]string{"a": "b"},
chunks: [][]sample{{{t: 0}, {t: 10}}, {{t: 11}, {t: 20}}},
},
},
},
{
title: "Populate from two blocks.",
inputSeriesSamples: [][]seriesSamples{
{
{
lset: map[string]string{"a": "b"},
chunks: [][]sample{{{t: 0}, {t: 10}}, {{t: 11}, {t: 20}}},
},
{
lset: map[string]string{"a": "c"},
chunks: [][]sample{{{t: 1}, {t: 9}}, {{t: 10}, {t: 19}}},
},
{
// no-chunk series should be dropped.
lset: map[string]string{"a": "empty"},
},
},
{
{
lset: map[string]string{"a": "b"},
chunks: [][]sample{{{t: 21}, {t: 30}}},
},
{
lset: map[string]string{"a": "c"},
chunks: [][]sample{{{t: 40}, {t: 45}}},
},
},
},
expSeriesSamples: []seriesSamples{
{
lset: map[string]string{"a": "b"},
chunks: [][]sample{{{t: 0}, {t: 10}}, {{t: 11}, {t: 20}}, {{t: 21}, {t: 30}}},
},
{
lset: map[string]string{"a": "c"},
chunks: [][]sample{{{t: 1}, {t: 9}}, {{t: 10}, {t: 19}}, {{t: 40}, {t: 45}}},
},
},
},
{
title: "Populate from two blocks; chunks with negative time.",
inputSeriesSamples: [][]seriesSamples{
{
{
lset: map[string]string{"a": "b"},
chunks: [][]sample{{{t: 0}, {t: 10}}, {{t: 11}, {t: 20}}},
},
{
lset: map[string]string{"a": "c"},
chunks: [][]sample{{{t: -11}, {t: -9}}, {{t: 10}, {t: 19}}},
},
{
// no-chunk series should be dropped.
lset: map[string]string{"a": "empty"},
},
},
{
{
lset: map[string]string{"a": "b"},
chunks: [][]sample{{{t: 21}, {t: 30}}},
},
{
lset: map[string]string{"a": "c"},
chunks: [][]sample{{{t: 40}, {t: 45}}},
},
},
},
compactMinTime: -11,
expSeriesSamples: []seriesSamples{
{
lset: map[string]string{"a": "b"},
chunks: [][]sample{{{t: 0}, {t: 10}}, {{t: 11}, {t: 20}}, {{t: 21}, {t: 30}}},
},
{
lset: map[string]string{"a": "c"},
chunks: [][]sample{{{t: -11}, {t: -9}}, {{t: 10}, {t: 19}}, {{t: 40}, {t: 45}}},
},
},
},
{
title: "Populate from two blocks showing that order is maintained.",
inputSeriesSamples: [][]seriesSamples{
{
{
lset: map[string]string{"a": "b"},
chunks: [][]sample{{{t: 0}, {t: 10}}, {{t: 11}, {t: 20}}},
},
{
lset: map[string]string{"a": "c"},
chunks: [][]sample{{{t: 1}, {t: 9}}, {{t: 10}, {t: 19}}},
},
},
{
{
lset: map[string]string{"a": "b"},
chunks: [][]sample{{{t: 21}, {t: 30}}},
},
{
lset: map[string]string{"a": "c"},
chunks: [][]sample{{{t: 40}, {t: 45}}},
},
},
},
expSeriesSamples: []seriesSamples{
{
lset: map[string]string{"a": "b"},
chunks: [][]sample{{{t: 0}, {t: 10}}, {{t: 11}, {t: 20}}, {{t: 21}, {t: 30}}},
},
{
lset: map[string]string{"a": "c"},
chunks: [][]sample{{{t: 1}, {t: 9}}, {{t: 10}, {t: 19}}, {{t: 40}, {t: 45}}},
},
},
},
{
title: "Populate from two blocks showing that order of series is sorted.",
inputSeriesSamples: [][]seriesSamples{
{
{
lset: map[string]string{"a": "4"},
chunks: [][]sample{{{t: 5}, {t: 7}}},
},
{
lset: map[string]string{"a": "3"},
chunks: [][]sample{{{t: 5}, {t: 6}}},
},
{
lset: map[string]string{"a": "same"},
chunks: [][]sample{{{t: 1}, {t: 4}}},
},
},
{
{
lset: map[string]string{"a": "2"},
chunks: [][]sample{{{t: 1}, {t: 3}}},
},
{
lset: map[string]string{"a": "1"},
chunks: [][]sample{{{t: 1}, {t: 2}}},
},
{
lset: map[string]string{"a": "same"},
chunks: [][]sample{{{t: 5}, {t: 8}}},
},
},
},
expSeriesSamples: []seriesSamples{
{
lset: map[string]string{"a": "1"},
chunks: [][]sample{{{t: 1}, {t: 2}}},
},
{
lset: map[string]string{"a": "2"},
chunks: [][]sample{{{t: 1}, {t: 3}}},
},
{
lset: map[string]string{"a": "3"},
chunks: [][]sample{{{t: 5}, {t: 6}}},
},
{
lset: map[string]string{"a": "4"},
chunks: [][]sample{{{t: 5}, {t: 7}}},
},
{
lset: map[string]string{"a": "same"},
chunks: [][]sample{{{t: 1}, {t: 4}}, {{t: 5}, {t: 8}}},
},
},
},
{
title: "Populate from two blocks 1:1 duplicated chunks; with negative timestamps.",
inputSeriesSamples: [][]seriesSamples{
{
{
lset: map[string]string{"a": "1"},
chunks: [][]sample{{{t: 1}, {t: 2}}, {{t: 3}, {t: 4}}},
},
{
lset: map[string]string{"a": "2"},
chunks: [][]sample{{{t: -3}, {t: -2}}, {{t: 1}, {t: 3}, {t: 4}}, {{t: 5}, {t: 6}}},
},
},
{
{
lset: map[string]string{"a": "1"},
chunks: [][]sample{{{t: 3}, {t: 4}}},
},
{
lset: map[string]string{"a": "2"},
chunks: [][]sample{{{t: 1}, {t: 3}, {t: 4}}, {{t: 7}, {t: 8}}},
},
},
},
compactMinTime: -3,
expSeriesSamples: []seriesSamples{
{
lset: map[string]string{"a": "1"},
chunks: [][]sample{{{t: 1}, {t: 2}}, {{t: 3}, {t: 4}}},
},
{
lset: map[string]string{"a": "2"},
chunks: [][]sample{{{t: -3}, {t: -2}}, {{t: 1}, {t: 3}, {t: 4}}, {{t: 5}, {t: 6}}, {{t: 7}, {t: 8}}},
},
},
},
{
// This should not happened because head block is making sure the chunks are not crossing block boundaries.
// We used to return error, but now chunk is trimmed.
title: "Populate from single block containing chunk outside of compact meta time range.",
inputSeriesSamples: [][]seriesSamples{
{
{
lset: map[string]string{"a": "b"},
chunks: [][]sample{{{t: 1}, {t: 2}}, {{t: 10}, {t: 30}}},
},
},
},
compactMinTime: 0,
compactMaxTime: 20,
expSeriesSamples: []seriesSamples{
{
lset: map[string]string{"a": "b"},
chunks: [][]sample{{{t: 1}, {t: 2}}, {{t: 10}}},
},
},
},
{
// Introduced by https://github.com/prometheus/tsdb/issues/347. We used to return error, but now chunk is trimmed.
title: "Populate from single block containing extra chunk",
inputSeriesSamples: [][]seriesSamples{
{
{
lset: map[string]string{"a": "issue347"},
chunks: [][]sample{{{t: 1}, {t: 2}}, {{t: 10}, {t: 20}}},
},
},
},
compactMinTime: 0,
compactMaxTime: 10,
expSeriesSamples: []seriesSamples{
{
lset: map[string]string{"a": "issue347"},
chunks: [][]sample{{{t: 1}, {t: 2}}},
},
},
},
{
// Deduplication expected.
// Introduced by pull/370 and pull/539.
title: "Populate from two blocks containing duplicated chunk.",
inputSeriesSamples: [][]seriesSamples{
{
{
lset: map[string]string{"a": "b"},
chunks: [][]sample{{{t: 1}, {t: 2}}, {{t: 10}, {t: 20}}},
},
},
{
{
lset: map[string]string{"a": "b"},
chunks: [][]sample{{{t: 10}, {t: 20}}},
},
},
},
expSeriesSamples: []seriesSamples{
{
lset: map[string]string{"a": "b"},
chunks: [][]sample{{{t: 1}, {t: 2}}, {{t: 10}, {t: 20}}},
},
},
},
{
// Introduced by https://github.com/prometheus/tsdb/pull/539.
title: "Populate from three overlapping blocks.",
inputSeriesSamples: [][]seriesSamples{
{
{
lset: map[string]string{"a": "overlap-all"},
chunks: [][]sample{{{t: 19}, {t: 30}}},
},
{
lset: map[string]string{"a": "overlap-beginning"},
chunks: [][]sample{{{t: 0}, {t: 5}}},
},
{
lset: map[string]string{"a": "overlap-ending"},
chunks: [][]sample{{{t: 21}, {t: 30}}},
},
},
{
{
lset: map[string]string{"a": "overlap-all"},
chunks: [][]sample{{{t: 0}, {t: 10}, {t: 11}, {t: 20}}},
},
{
lset: map[string]string{"a": "overlap-beginning"},
chunks: [][]sample{{{t: 0}, {t: 10}, {t: 12}, {t: 20}}},
},
{
lset: map[string]string{"a": "overlap-ending"},
chunks: [][]sample{{{t: 0}, {t: 10}, {t: 13}, {t: 20}}},
},
},
{
{
lset: map[string]string{"a": "overlap-all"},
chunks: [][]sample{{{t: 27}, {t: 35}}},
},
{
lset: map[string]string{"a": "overlap-ending"},
chunks: [][]sample{{{t: 27}, {t: 35}}},
},
},
},
expSeriesSamples: []seriesSamples{
{
lset: map[string]string{"a": "overlap-all"},
chunks: [][]sample{{{t: 0}, {t: 10}, {t: 11}, {t: 19}, {t: 20}, {t: 27}, {t: 30}, {t: 35}}},
},
{
lset: map[string]string{"a": "overlap-beginning"},
chunks: [][]sample{{{t: 0}, {t: 5}, {t: 10}, {t: 12}, {t: 20}}},
},
{
lset: map[string]string{"a": "overlap-ending"},
chunks: [][]sample{{{t: 0}, {t: 10}, {t: 13}, {t: 20}}, {{t: 21}, {t: 27}, {t: 30}, {t: 35}}},
},
},
},
{
title: "Populate from three partially overlapping blocks with few full chunks.",
inputSeriesSamples: [][]seriesSamples{
{
{
lset: map[string]string{"a": "1", "b": "1"},
chunks: samplesForRange(0, 659, 120), // 5 chunks and half.
},
{
lset: map[string]string{"a": "1", "b": "2"},
chunks: samplesForRange(0, 659, 120),
},
},
{
{
lset: map[string]string{"a": "1", "b": "2"},
chunks: samplesForRange(480, 1199, 120), // two chunks overlapping with previous, two non overlapping and two overlapping with next block.
},
{
lset: map[string]string{"a": "1", "b": "3"},
chunks: samplesForRange(480, 1199, 120),
},
},
{
{
lset: map[string]string{"a": "1", "b": "2"},
chunks: samplesForRange(960, 1499, 120), // 5 chunks and half.
},
{
lset: map[string]string{"a": "1", "b": "4"},
chunks: samplesForRange(960, 1499, 120),
},
},
},
expSeriesSamples: []seriesSamples{
{
lset: map[string]string{"a": "1", "b": "1"},
chunks: samplesForRange(0, 659, 120),
},
{
lset: map[string]string{"a": "1", "b": "2"},
chunks: samplesForRange(0, 1499, 120),
},
{
lset: map[string]string{"a": "1", "b": "3"},
chunks: samplesForRange(480, 1199, 120),
},
{
lset: map[string]string{"a": "1", "b": "4"},
chunks: samplesForRange(960, 1499, 120),
},
},
},
{
title: "Populate from three partially overlapping blocks with chunks that are expected to merge into single big chunks.",
inputSeriesSamples: [][]seriesSamples{
{
{
lset: map[string]string{"a": "1", "b": "2"},
chunks: [][]sample{{{t: 0}, {t: 6902464}}, {{t: 6961968}, {t: 7080976}}},
},
},
{
{
lset: map[string]string{"a": "1", "b": "2"},
chunks: [][]sample{{{t: 3600000}, {t: 13953696}}, {{t: 14042952}, {t: 14221464}}},
},
},
{
{
lset: map[string]string{"a": "1", "b": "2"},
chunks: [][]sample{{{t: 10800000}, {t: 14251232}}, {{t: 14280984}, {t: 14340488}}},
},
},
},
expSeriesSamples: []seriesSamples{
{
lset: map[string]string{"a": "1", "b": "2"},
chunks: [][]sample{{{t: 0}, {t: 3600000}, {t: 6902464}, {t: 6961968}, {t: 7080976}, {t: 10800000}, {t: 13953696}, {t: 14042952}, {t: 14221464}, {t: 14251232}}, {{t: 14280984}, {t: 14340488}}},
},
},
},
{
// Regression test for populateWithDelChunkSeriesIterator failing to set minTime on chunks.
title: "Populate from mixed type series and expect sample inside the interval only.",
compactMinTime: 1,
compactMaxTime: 11,
inputSeriesSamples: [][]seriesSamples{
{
{
lset: map[string]string{"a": "1"},
chunks: [][]sample{
{{t: 0, h: tsdbutil.GenerateTestHistogram(0)}, {t: 1, h: tsdbutil.GenerateTestHistogram(1)}},
{{t: 10, f: 1}, {t: 11, f: 2}},
},
},
},
},
expSeriesSamples: []seriesSamples{
{
lset: map[string]string{"a": "1"},
chunks: [][]sample{
{{t: 1, h: tsdbutil.GenerateTestHistogram(1)}},
{{t: 10, f: 1}},
},
},
},
},
{
title: "Populate from single block with index reader postings function selecting different series. Expect empty block.",
inputSeriesSamples: [][]seriesSamples{
{
{
lset: map[string]string{"a": "b"},
chunks: [][]sample{{{t: 0}, {t: 10}}, {{t: 11}, {t: 20}}},
},
},
},
irPostingsFunc: func(ctx context.Context, reader IndexReader) index.Postings {
p, err := reader.Postings(ctx, "a", "c")
if err != nil {
return index.EmptyPostings()
}
return reader.SortedPostings(p)
},
},
{
title: "Populate from single block with index reader postings function selecting one series. Expect partial block.",
inputSeriesSamples: [][]seriesSamples{
{
{
lset: map[string]string{"a": "b"},
chunks: [][]sample{{{t: 0}, {t: 10}}, {{t: 11}, {t: 20}}},
},
{
lset: map[string]string{"a": "c"},
chunks: [][]sample{{{t: 0}, {t: 10}}, {{t: 11}, {t: 20}}},
},
{
lset: map[string]string{"a": "d"},
chunks: [][]sample{{{t: 0}, {t: 10}}, {{t: 11}, {t: 20}}},
},
},
},
irPostingsFunc: func(ctx context.Context, reader IndexReader) index.Postings {
p, err := reader.Postings(ctx, "a", "c", "d")
if err != nil {
return index.EmptyPostings()
}
return reader.SortedPostings(p)
},
expSeriesSamples: []seriesSamples{
{
lset: map[string]string{"a": "c"},
chunks: [][]sample{{{t: 0}, {t: 10}}, {{t: 11}, {t: 20}}},
},
{
lset: map[string]string{"a": "d"},
chunks: [][]sample{{{t: 0}, {t: 10}}, {{t: 11}, {t: 20}}},
},
},
},
} {
t.Run(tc.title, func(t *testing.T) {
blocks := make([]BlockReader, 0, len(tc.inputSeriesSamples))
for _, b := range tc.inputSeriesSamples {
ir, cr, mint, maxt := createIdxChkReaders(t, b)
blocks = append(blocks, &mockBReader{ir: ir, cr: cr, mint: mint, maxt: maxt})
}
c, err := NewLeveledCompactor(context.Background(), nil, nil, []int64{0}, nil, nil)
require.NoError(t, err)
meta := &BlockMeta{
MinTime: tc.compactMinTime,
MaxTime: tc.compactMaxTime,
}
if meta.MaxTime == 0 {
meta.MaxTime = math.MaxInt64
}
iw := &mockIndexWriter{}
blockPopulator := DefaultBlockPopulator{}
irPostingsFunc := AllSortedPostings
if tc.irPostingsFunc != nil {
irPostingsFunc = tc.irPostingsFunc
}
err = blockPopulator.PopulateBlock(c.ctx, c.metrics, c.logger, c.chunkPool, c.mergeFunc, blocks, meta, iw, nopChunkWriter{}, irPostingsFunc)
if tc.expErr != nil {
require.EqualError(t, err, tc.expErr.Error())
return
}
require.NoError(t, err)
// Check if response is expected and chunk is valid.
var raw []seriesSamples
for _, s := range iw.seriesChunks {
ss := seriesSamples{lset: s.l.Map()}
var iter chunkenc.Iterator
for _, chk := range s.chunks {
var (
samples = make([]sample, 0, chk.Chunk.NumSamples())
iter = chk.Chunk.Iterator(iter)
firstTs int64 = math.MaxInt64
s sample
)
for vt := iter.Next(); vt != chunkenc.ValNone; vt = iter.Next() {
switch vt {
case chunkenc.ValFloat:
s.t, s.f = iter.At()
samples = append(samples, s)
case chunkenc.ValHistogram:
s.t, s.h = iter.AtHistogram(nil)
samples = append(samples, s)
case chunkenc.ValFloatHistogram:
s.t, s.fh = iter.AtFloatHistogram(nil)
samples = append(samples, s)
default:
require.Fail(t, "unexpected value type")
}
if firstTs == math.MaxInt64 {
firstTs = s.t
}
}
// Check if chunk has correct min, max times.
require.Equal(t, firstTs, chk.MinTime, "chunk Meta %v does not match the first encoded sample timestamp: %v", chk, firstTs)
require.Equal(t, s.t, chk.MaxTime, "chunk Meta %v does not match the last encoded sample timestamp %v", chk, s.t)
require.NoError(t, iter.Err())
ss.chunks = append(ss.chunks, samples)
}
raw = append(raw, ss)
}
require.Equal(t, tc.expSeriesSamples, raw)
// Check if stats are calculated properly.
s := BlockStats{NumSeries: uint64(len(tc.expSeriesSamples))}
for _, series := range tc.expSeriesSamples {
s.NumChunks += uint64(len(series.chunks))
for _, chk := range series.chunks {
s.NumSamples += uint64(len(chk))
for _, smpl := range chk {
if smpl.h != nil || smpl.fh != nil {
s.NumHistogramSamples++
} else {
s.NumFloatSamples++
}
}
}
}
require.Equal(t, s, meta.Stats)
})
}
}
func BenchmarkCompaction(b *testing.B) {
cases := []struct {
ranges [][2]int64
compactionType string
}{
{
ranges: [][2]int64{{0, 100}, {200, 300}, {400, 500}, {600, 700}},
compactionType: "normal",
},
{
ranges: [][2]int64{{0, 1000}, {2000, 3000}, {4000, 5000}, {6000, 7000}},
compactionType: "normal",
},
{
ranges: [][2]int64{{0, 2000}, {3000, 5000}, {6000, 8000}, {9000, 11000}},
compactionType: "normal",
},
{
ranges: [][2]int64{{0, 5000}, {6000, 11000}, {12000, 17000}, {18000, 23000}},
compactionType: "normal",
},
// 40% overlaps.
{
ranges: [][2]int64{{0, 100}, {60, 160}, {120, 220}, {180, 280}},
compactionType: "vertical",
},
{
ranges: [][2]int64{{0, 1000}, {600, 1600}, {1200, 2200}, {1800, 2800}},
compactionType: "vertical",
},
{
ranges: [][2]int64{{0, 2000}, {1200, 3200}, {2400, 4400}, {3600, 5600}},
compactionType: "vertical",
},
{
ranges: [][2]int64{{0, 5000}, {3000, 8000}, {6000, 11000}, {9000, 14000}},
compactionType: "vertical",
},
}
nSeries := 10000
for _, c := range cases {
nBlocks := len(c.ranges)
b.Run(fmt.Sprintf("type=%s,blocks=%d,series=%d,samplesPerSeriesPerBlock=%d", c.compactionType, nBlocks, nSeries, c.ranges[0][1]-c.ranges[0][0]+1), func(b *testing.B) {
dir := b.TempDir()
blockDirs := make([]string, 0, len(c.ranges))
var blocks []*Block
for _, r := range c.ranges {
block, err := OpenBlock(nil, createBlock(b, dir, genSeries(nSeries, 10, r[0], r[1])), nil, nil)
require.NoError(b, err)
blocks = append(blocks, block)
defer func() {
require.NoError(b, block.Close())
}()
blockDirs = append(blockDirs, block.Dir())
}
c, err := NewLeveledCompactor(context.Background(), nil, promslog.NewNopLogger(), []int64{0}, nil, nil)
require.NoError(b, err)
b.ResetTimer()
b.ReportAllocs()
for b.Loop() {
_, err = c.Compact(dir, blockDirs, blocks)
require.NoError(b, err)
}
})
}
}
func BenchmarkCompactionFromHead(b *testing.B) {
dir := b.TempDir()
totalSeries := 100000
for labelNames := 1; labelNames < totalSeries; labelNames *= 10 {
labelValues := totalSeries / labelNames
b.Run(fmt.Sprintf("labelnames=%d,labelvalues=%d", labelNames, labelValues), func(b *testing.B) {
chunkDir := b.TempDir()
opts := DefaultHeadOptions()
opts.ChunkRange = 1000
opts.ChunkDirRoot = chunkDir
h, err := NewHead(nil, nil, nil, nil, opts, nil)
require.NoError(b, err)
for ln := 0; ln < labelNames; ln++ {
app := h.Appender(context.Background())
for lv := range labelValues {
app.Append(0, labels.FromStrings(strconv.Itoa(ln), fmt.Sprintf("%d%s%d", lv, postingsBenchSuffix, ln)), 0, 0)
}
require.NoError(b, app.Commit())
}
b.ResetTimer()
b.ReportAllocs()
for i := 0; b.Loop(); i++ {
createBlockFromHead(b, filepath.Join(dir, fmt.Sprintf("%d-%d", i, labelNames)), h)
}
h.Close()
})
}
// Sub-benchmark with multiple head chunks per series, as can happen with
// native histograms that trigger a counter reset on every sample. This
// stresses the linked-list traversal in s.chunk() and exercises the
// head-chunk cache.
for _, tc := range []struct{ series, chunks int }{
{100000, 1}, // Baseline: cache skipped (prev==nil).
{100000, 10}, // Realistic: cache active, but improvement negligible vs total compaction cost.
{10, 100}, // Moderate pathological case.
{10, 1000}, // Heavy pathological case.
} {
nSeries, nChunks := tc.series, tc.chunks
b.Run(fmt.Sprintf("many head chunks/series=%d,chunks=%d", nSeries, nChunks), func(b *testing.B) {
dir := b.TempDir()
chunkDir := b.TempDir()
opts := DefaultHeadOptions()
opts.ChunkRange = int64(nChunks+1) * 1000 // Wide enough so nothing gets mmapped.
opts.ChunkDirRoot = chunkDir
h, err := NewHead(nil, nil, nil, nil, opts, nil)
require.NoError(b, err)
app := h.Appender(b.Context())
for i := range nSeries {
lbls := labels.FromStrings("__name__", "bench", "series", strconv.Itoa(i))
for j := range nChunks {
// Counter reset on every histogram forces a new head chunk per sample.
hist := tsdbutil.GenerateTestHistogramWithHint(j, histogram.CounterReset)
_, err := app.AppendHistogram(0, lbls, int64(j)*1000, hist, nil)
require.NoError(b, err)
}
}
require.NoError(b, app.Commit())
// Verify we actually have the expected number of head chunks.
for i := range nSeries {
lbls := labels.FromStrings("__name__", "bench", "series", strconv.Itoa(i))
s := h.series.getByHash(lbls.Hash(), lbls)
require.NotNil(b, s, "series %d not found", i)
require.Equal(b, uint32(nChunks), s.headChunkCount.Load(), "series %d: unexpected head chunk count", i)
}
b.ResetTimer()
b.ReportAllocs()
for i := 0; b.Loop(); i++ {
createBlockFromHead(b, filepath.Join(dir, strconv.Itoa(i)), h)
}
h.Close()
})
}
}
func TestCompactionFromHead(t *testing.T) {
t.Run("multiple head chunks", func(t *testing.T) {
// Verify compaction from a Head with multiple head chunks per series
// produces correct blocks. This exercises the chunkCacheEnabler path
// in PopulateBlock, which enables the head-chunk cache during
// compaction for O(1) chunk lookups.
opts := DefaultHeadOptions()
opts.ChunkRange = 100
opts.ChunkDirRoot = t.TempDir()
h, err := NewHead(nil, nil, nil, nil, opts, nil)
require.NoError(t, err)
t.Cleanup(func() { require.NoError(t, h.Close()) })
// Append enough samples to create multiple head chunks.
// With ChunkRange=100 and DefaultSamplesPerChunk=120, each chunk holds
// ~20 samples (range/step = 100/5). 200 samples → ~10 head chunks.
lbls := labels.FromStrings("__name__", "test")
var expSamples []sample
app := h.Appender(t.Context())
for i := int64(0); i < 1000; i += 5 {
_, err := app.Append(0, lbls, i, float64(i))
require.NoError(t, err)
expSamples = append(expSamples, sample{t: i, f: float64(i)})
}
require.NoError(t, app.Commit())
// Verify we have multiple head chunks.
s := h.series.getByID(1)
require.NotNil(t, s)
s.Lock()
headChunks := int(s.headChunkCount.Load())
s.Unlock()
require.Greater(t, headChunks, 1, "need multiple head chunks for this test")
// Compact from head.
blockDir := createBlockFromHead(t, t.TempDir(), h)
// Re-read the block and verify all samples survived.
block, err := OpenBlock(promslog.NewNopLogger(), blockDir, nil, nil)
require.NoError(t, err)
t.Cleanup(func() { require.NoError(t, block.Close()) })
q, err := NewBlockQuerier(block, math.MinInt64, math.MaxInt64)
require.NoError(t, err)
defer q.Close()
ss := q.Select(t.Context(), false, nil, labels.MustNewMatcher(labels.MatchEqual, "__name__", "test"))
require.True(t, ss.Next(), "expected a series")
var actSamples []sample
it := ss.At().Iterator(nil)
for it.Next() == chunkenc.ValFloat {
ts, v := it.At()
actSamples = append(actSamples, sample{t: ts, f: v})
}
require.NoError(t, it.Err())
require.False(t, ss.Next(), "expected only one series")
require.NoError(t, ss.Err())
require.Equal(t, expSamples, actSamples, "compacted block should contain all samples")
})
}
func BenchmarkCompactionFromOOOHead(b *testing.B) {
dir := b.TempDir()
totalSeries := 100000
totalSamples := 100
for labelNames := 1; labelNames < totalSeries; labelNames *= 10 {
labelValues := totalSeries / labelNames
b.Run(fmt.Sprintf("labelnames=%d,labelvalues=%d", labelNames, labelValues), func(b *testing.B) {
chunkDir := b.TempDir()
opts := DefaultHeadOptions()
opts.ChunkRange = 1000
opts.ChunkDirRoot = chunkDir
opts.OutOfOrderTimeWindow.Store(int64(totalSamples))
h, err := NewHead(nil, nil, nil, nil, opts, nil)
require.NoError(b, err)
for ln := 0; ln < labelNames; ln++ {
app := h.Appender(context.Background())
for lv := range labelValues {
lbls := labels.FromStrings(strconv.Itoa(ln), fmt.Sprintf("%d%s%d", lv, postingsBenchSuffix, ln))
_, err = app.Append(0, lbls, int64(totalSamples), 0)
require.NoError(b, err)
for ts := range totalSamples {
_, err = app.Append(0, lbls, int64(ts), float64(ts))
require.NoError(b, err)
}
}
require.NoError(b, app.Commit())
}
b.ResetTimer()
b.ReportAllocs()
for i := 0; b.Loop(); i++ {
oooHead, err := NewOOOCompactionHead(context.TODO(), h)
require.NoError(b, err)
createBlockFromOOOHead(b, filepath.Join(dir, fmt.Sprintf("%d-%d", i, labelNames)), oooHead)
}
h.Close()
})
}
}
// setupDBForSelectedSeriesBenchmark opens a fresh DB whose head contains
// totalSeries series, each with samplesPerSeries in-order samples confined to
// a single 1s head chunk range. It returns the DB and the head series
// refs in append order.
//
// Auto-compaction is disabled so the benchmark observes exactly the state
// produced by this helper. The caller owns the returned DB and must close it.
//
// The helper is shared by BenchmarkCompactSelectedSeries and the upcoming
// filterSeriesAndSortPostings benchmark, so it does not assume which entry
// point will use the setup.
func setupDBForSelectedSeriesBenchmark(tb testing.TB, totalSeries, samplesPerSeries int) (*DB, []storage.SeriesRef) {
require.Positive(tb, samplesPerSeries, "samplesPerSeries must be > 0")
require.LessOrEqual(tb, samplesPerSeries, 1000, "samples must fit one head chunk range (1000 ms)")
opts := DefaultOptions()
opts.MinBlockDuration = 1000
opts.MaxBlockDuration = 1000
db, err := Open(tb.TempDir(), nil, nil, opts, nil)
require.NoError(tb, err)
db.DisableCompactions()
const appendBatch = 10_000
refs := make([]storage.SeriesRef, 0, totalSeries)
app := db.Appender(context.Background())
for i := range totalSeries {
lbls := labels.FromStrings("__name__", "metric", "instance", strconv.Itoa(i))
var ref storage.SeriesRef
for ts := int64(0); ts < int64(samplesPerSeries); ts++ {
ref, err = app.Append(ref, lbls, ts, float64(ts))
require.NoError(tb, err)
}
refs = append(refs, ref)
if (i+1)%appendBatch == 0 {
require.NoError(tb, app.Commit())
app = db.Appender(context.Background())
}
}
require.NoError(tb, app.Commit())
return db, refs
}
// pickRefsEvenly returns count refs sampled at approximately even intervals
// across refs. This distributes the selected refs over the full input range
// rather than clustering them, which is useful for benchmarks that want
// postings intersections to scan most of the postings list.
func pickRefsEvenly(refs []storage.SeriesRef, count int) []storage.SeriesRef {
if count >= len(refs) {
out := make([]storage.SeriesRef, len(refs))
copy(out, refs)
return out
}
out := make([]storage.SeriesRef, 0, count)
step := float64(len(refs)) / float64(count)
for i := range count {
out = append(out, refs[int(float64(i)*step)])
}
return out
}
// BenchmarkCompactSelectedSeries measures the end-to-end cost of compacting a
// selected subset of head series into blocks and evicting them from the head.
//
// The benchmark keeps the head size fixed and varies the selected fraction
// (100%, 50%, 30%, 10%, 1%, and 0.1%). A 100% selection approximates the
// workload of CompactStaleHead, while smaller fractions measure how much work
// is avoided when SelectedSeriesHead restricts compaction to a small subset of
// series.
//
// Each iteration rebuilds the DB because CompactSelectedSeries is destructive:
// selected series are evicted from the head and cannot be reused by subsequent
// iterations.
//
// Each series contains DefaultSamplesPerChunk samples, matching the TSDB's
// target chunk size so chunk-writing costs are representative of production
// workloads rather than degenerate single-sample chunks.
func BenchmarkCompactSelectedSeries(b *testing.B) {
const (
totalSeries = 100_000
samplesPerSeries = DefaultSamplesPerChunk
)
fractions := []float64{1.0, 0.5, 0.3, 0.1, 0.01, 0.001}
for _, fraction := range fractions {
selectedCount := max(1, int(float64(totalSeries)*fraction))
b.Run(fmt.Sprintf("totalSeries=%d/selectedSeries=%d", totalSeries, selectedCount), func(b *testing.B) {
b.ReportAllocs()
for b.Loop() {
b.StopTimer()
db, allRefs := setupDBForSelectedSeriesBenchmark(b, totalSeries, samplesPerSeries)
selectedRefs := pickRefsEvenly(allRefs, selectedCount)
b.StartTimer()
require.NoError(b, db.CompactSelectedSeries(selectedRefs))
b.StopTimer()
require.NoError(b, db.Close())
b.StartTimer()
}
})
}
}
// BenchmarkFilterSeriesAndSortPostings measures the cost of
// Head.filterSeriesAndSortPostings under the isSeriesWithoutOOO predicate.
// This is the filtering step performed by CompactSelectedSeries while holding
// db.cmtx, before invoking compactHeadViewLocked.
//
// The operation is O(N log N) in the number of candidate refs and is a likely
// contributor to the fixed-cost floor observed at low selectivity in
// BenchmarkCompactSelectedSeries.
//
// The benchmark varies the input size as a fraction of a fixed head size,
// mirroring BenchmarkCompactSelectedSeries so the results can be compared
// directly. The head is built once per sub-benchmark because the function is
// non-destructive.
//
// samplesPerSeries matches BenchmarkCompactSelectedSeries for setup
// consistency, although the filter only examines series-level state and is
// insensitive to its value.
func BenchmarkFilterSeriesAndSortPostings(b *testing.B) {
const (
totalSeries = 100_000
samplesPerSeries = DefaultSamplesPerChunk
)
fractions := []float64{1.0, 0.5, 0.3, 0.1, 0.01, 0.001}
for _, fraction := range fractions {
inputCount := max(1, int(float64(totalSeries)*fraction))
b.Run(fmt.Sprintf("totalSeries=%d/inputSeries=%d", totalSeries, inputCount), func(b *testing.B) {
db, allRefs := setupDBForSelectedSeriesBenchmark(b, totalSeries, samplesPerSeries)
b.Cleanup(func() { require.NoError(b, db.Close()) })
inputRefs := pickRefsEvenly(allRefs, inputCount)
b.ResetTimer()
b.ReportAllocs()
for b.Loop() {
_, err := db.head.filterSeriesAndSortPostings(index.NewListPostings(inputRefs), isSeriesWithoutOOO)
require.NoError(b, err)
}
})
}
}
// TestDisableAutoCompactions checks that we can
// disable and enable the auto compaction.
// This is needed for unit tests that rely on
// checking state before and after a compaction.
func TestDisableAutoCompactions(t *testing.T) {
db := newTestDB(t)
blockRange := db.compactor.(*LeveledCompactor).ranges[0]
label := labels.FromStrings("foo", "bar")
// Trigger a compaction to check that it was skipped and
// no new blocks were created when compaction is disabled.
db.DisableCompactions()
app := db.Appender(context.Background())
for i := range int64(3) {
_, err := app.Append(0, label, i*blockRange, 0)
require.NoError(t, err)
_, err = app.Append(0, label, i*blockRange+1000, 0)
require.NoError(t, err)
}
require.NoError(t, app.Commit())
select {
case db.compactc <- struct{}{}:
default:
}
for range 10 {
if prom_testutil.ToFloat64(db.metrics.compactionsSkipped) > 0.0 {
break
}
time.Sleep(10 * time.Millisecond)
}
require.Greater(t, prom_testutil.ToFloat64(db.metrics.compactionsSkipped), 0.0, "No compaction was skipped after the set timeout.")
require.Empty(t, db.blocks)
// Enable the compaction, trigger it and check that the block is persisted.
db.EnableCompactions()
select {
case db.compactc <- struct{}{}:
default:
}
for range 100 {
if len(db.Blocks()) > 0 {
break
}
time.Sleep(100 * time.Millisecond)
}
require.NotEmpty(t, db.Blocks(), "No block was persisted after the set timeout.")
}
// TestCancelCompactions ensures that when the db is closed
// any running compaction is cancelled to unblock closing the db.
func TestCancelCompactions(t *testing.T) {
synctest.Test(t, func(t *testing.T) {
compactionStarted := make(chan struct{})
compactionCanceled := make(chan struct{})
opts := DefaultOptions()
opts.NewCompactorFunc = func(ctx context.Context, _ prometheus.Registerer, _ *slog.Logger, _ []int64, _ chunkenc.Pool, _ *Options) (Compactor, error) {
return &mockCompactorFn{
planFn: func() ([]string, error) {
return []string{"block-a", "block-b"}, nil
},
compactFn: func() ([]ulid.ULID, error) {
close(compactionStarted)
<-ctx.Done()
close(compactionCanceled)
return nil, ctx.Err()
},
// writeFn is unused: this test never reaches the head, OOO,
// or stale-series compaction paths that would call Write.
writeFn: func() ([]ulid.ULID, error) {
return nil, nil
},
}, nil
}
db := newTestDB(t, withOpts(opts))
db.compactc <- struct{}{}
<-compactionStarted
require.NoError(t, db.Close())
<-compactionCanceled
// Wrapped context.Canceled must not be counted as a real compaction
// failure (verifies errors.Is at every level of the chain).
require.Equal(t, 0.0, prom_testutil.ToFloat64(db.metrics.compactionsFailed))
})
}
type blockPopulatorFunc func(context.Context, *CompactorMetrics, *slog.Logger, chunkenc.Pool, storage.VerticalChunkSeriesMergeFunc, []BlockReader, *BlockMeta, IndexWriter, ChunkWriter, IndexReaderPostingsFunc) error
func (f blockPopulatorFunc) PopulateBlock(ctx context.Context, metrics *CompactorMetrics, logger *slog.Logger, chunkPool chunkenc.Pool, mergeFunc storage.VerticalChunkSeriesMergeFunc, blocks []BlockReader, meta *BlockMeta, indexw IndexWriter, chunkw ChunkWriter, postingsFunc IndexReaderPostingsFunc) error {
return f(ctx, metrics, logger, chunkPool, mergeFunc, blocks, meta, indexw, chunkw, postingsFunc)
}
// TestCanceledCompactionDoesNotMarkBlocksFailed ensures that a compaction
// aborted via context cancellation does not mark its source blocks as
// Compaction.Failed. The context.Canceled error must be detected with
// errors.Is so wrapped values are still recognized at every level.
func TestCanceledCompactionDoesNotMarkBlocksFailed(t *testing.T) {
t.Parallel()
tmpdir := t.TempDir()
blockDirs := []string{
createBlock(t, tmpdir, genSeries(1, 1, 0, 100)),
createBlock(t, tmpdir, genSeries(1, 1, 100, 200)),
}
compactor, err := NewLeveledCompactor(t.Context(), nil, promslog.NewNopLogger(), []int64{200}, nil, nil)
require.NoError(t, err)
_, err = compactor.CompactWithBlockPopulator(tmpdir, blockDirs, nil, blockPopulatorFunc(
func(context.Context, *CompactorMetrics, *slog.Logger, chunkenc.Pool, storage.VerticalChunkSeriesMergeFunc, []BlockReader, *BlockMeta, IndexWriter, ChunkWriter, IndexReaderPostingsFunc) error {
return context.Canceled
},
))
require.ErrorIs(t, err, context.Canceled)
for _, dir := range blockDirs {
meta, _, err := readMetaFile(dir)
require.NoError(t, err)
require.Falsef(t, meta.Compaction.Failed, "block %s should not be marked as compaction failed", meta.ULID)
}
}
// 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
// the new block created from the compaction should be deleted.
require.Error(t, db.Compact(ctx))
require.Equal(t, 1.0, prom_testutil.ToFloat64(db.metrics.reloadsFailed), "'failed db reloadBlocks' count metrics mismatch")
require.Equal(t, 1.0, prom_testutil.ToFloat64(db.compactor.(*LeveledCompactor).metrics.Ran), "`compaction` count metric mismatch")
require.Equal(t, 1.0, prom_testutil.ToFloat64(db.metrics.compactionsFailed), "`compactions failed` count metric mismatch")
actBlocks, err = blockDirs(db.Dir())
require.NoError(t, err)
require.Equal(t, expBlocks, len(actBlocks)-1, "block count should be the same as before the compaction") // -1 to exclude the corrupted block.
})
}
}
func TestHeadCompactionWithHistograms(t *testing.T) {
for _, floatTest := range []bool{true, false} {
t.Run(fmt.Sprintf("float=%t", floatTest), func(t *testing.T) {
head, _ := newTestHead(t, DefaultBlockDuration, compression.None, false)
require.NoError(t, head.Init(0))
minute := func(m int) int64 { return int64(m) * time.Minute.Milliseconds() }
ctx := context.Background()
appendHistogram := func(
lbls labels.Labels, from, to int, h *histogram.Histogram, exp *[]chunks.Sample,
) {
t.Helper()
app := head.Appender(ctx)
for tsMinute := from; tsMinute <= to; tsMinute++ {
var err error
if floatTest {
_, err = app.AppendHistogram(0, lbls, minute(tsMinute), nil, h.ToFloat(nil))
efh := h.ToFloat(nil)
if tsMinute == from {
efh.CounterResetHint = histogram.UnknownCounterReset
} else {
efh.CounterResetHint = histogram.NotCounterReset
}
*exp = append(*exp, sample{t: minute(tsMinute), fh: efh})
} else {
_, err = app.AppendHistogram(0, lbls, minute(tsMinute), h, nil)
eh := h.Copy()
if tsMinute == from {
eh.CounterResetHint = histogram.UnknownCounterReset
} else {
eh.CounterResetHint = histogram.NotCounterReset
}
*exp = append(*exp, sample{t: minute(tsMinute), h: eh})
}
require.NoError(t, err)
}
require.NoError(t, app.Commit())
}
appendFloat := func(lbls labels.Labels, from, to int, exp *[]chunks.Sample) {
t.Helper()
app := head.Appender(ctx)
for tsMinute := from; tsMinute <= to; tsMinute++ {
_, err := app.Append(0, lbls, minute(tsMinute), float64(tsMinute))
require.NoError(t, err)
*exp = append(*exp, sample{t: minute(tsMinute), f: float64(tsMinute)})
}
require.NoError(t, app.Commit())
}
var (
series1 = labels.FromStrings("foo", "bar1")
series2 = labels.FromStrings("foo", "bar2")
series3 = labels.FromStrings("foo", "bar3")
series4 = labels.FromStrings("foo", "bar4")
exp1, exp2, exp3, exp4 []chunks.Sample
)
h := &histogram.Histogram{
Count: 15,
ZeroCount: 4,
ZeroThreshold: 0.001,
Sum: 35.5,
Schema: 1,
PositiveSpans: []histogram.Span{
{Offset: 0, Length: 2},
{Offset: 2, Length: 2},
},
PositiveBuckets: []int64{1, 1, -1, 0},
NegativeSpans: []histogram.Span{
{Offset: 0, Length: 1},
{Offset: 1, Length: 2},
},
NegativeBuckets: []int64{1, 2, -1},
}
// Series with only histograms.
appendHistogram(series1, 100, 105, h, &exp1)
// Series starting with float and then getting histograms.
appendFloat(series2, 100, 102, &exp2)
appendHistogram(series2, 103, 105, h.Copy(), &exp2)
appendFloat(series2, 106, 107, &exp2)
appendHistogram(series2, 108, 109, h.Copy(), &exp2)
// Series starting with histogram and then getting float.
appendHistogram(series3, 101, 103, h.Copy(), &exp3)
appendFloat(series3, 104, 106, &exp3)
appendHistogram(series3, 107, 108, h.Copy(), &exp3)
appendFloat(series3, 109, 110, &exp3)
// A float only series.
appendFloat(series4, 100, 102, &exp4)
// Compaction.
mint := head.MinTime()
maxt := head.MaxTime() + 1 // Block intervals are half-open: [b.MinTime, b.MaxTime).
compactor, err := NewLeveledCompactor(context.Background(), nil, nil, []int64{DefaultBlockDuration}, chunkenc.NewPool(), nil)
require.NoError(t, err)
ids, err := compactor.Write(head.opts.ChunkDirRoot, head, mint, maxt, nil)
require.NoError(t, err)
require.Len(t, ids, 1)
// Open the block and query it and check the histograms.
block, err := OpenBlock(nil, path.Join(head.opts.ChunkDirRoot, ids[0].String()), nil, nil)
require.NoError(t, err)
t.Cleanup(func() {
require.NoError(t, block.Close())
})
q, err := NewBlockQuerier(block, block.MinTime(), block.MaxTime())
require.NoError(t, err)
actHists := query(t, q, labels.MustNewMatcher(labels.MatchRegexp, "foo", "bar.*"))
require.Equal(t, map[string][]chunks.Sample{
series1.String(): exp1,
series2.String(): exp2,
series3.String(): exp3,
series4.String(): exp4,
}, actHists)
})
}
}
// Depending on numSeriesPerSchema, it can take few gigs of memory;
// the test adds all samples to appender before committing instead of
// buffering the writes to make it run faster.
func TestSparseHistogramSpaceSavings(t *testing.T) {
t.Skip()
type testcase struct {
numSeriesPerSchema int
numBuckets int
numSpans int
gapBetweenSpans int
}
cases := []testcase{
{1, 15, 1, 0},
{1, 50, 1, 0},
{1, 100, 1, 0},
{1, 15, 3, 5},
{1, 50, 3, 3},
{1, 100, 3, 2},
{100, 15, 1, 0},
{100, 50, 1, 0},
{100, 100, 1, 0},
{100, 15, 3, 5},
{100, 50, 3, 3},
{100, 100, 3, 2},
}
type testSummary struct {
oldBlockTotalSeries int
oldBlockIndexSize int64
oldBlockChunksSize int64
oldBlockTotalSize int64
sparseBlockTotalSeries int
sparseBlockIndexSize int64
sparseBlockChunksSize int64
sparseBlockTotalSize int64
numBuckets int
numSpans int
gapBetweenSpans int
}
var summaries []testSummary
allSchemas := []int{-4, -3, -2, -1, 0, 1, 2, 3, 4, 5, 6, 7, 8}
schemaDescription := []string{"minus_4", "minus_3", "minus_2", "minus_1", "0", "1", "2", "3", "4", "5", "6", "7", "8"}
numHistograms := 120 * 4 // 15s scrape interval.
timeStep := DefaultBlockDuration / int64(numHistograms)
for _, c := range cases {
t.Run(
fmt.Sprintf("series=%d,span=%d,gap=%d,buckets=%d",
len(allSchemas)*c.numSeriesPerSchema,
c.numSpans,
c.gapBetweenSpans,
c.numBuckets,
),
func(t *testing.T) {
oldHead, _ := newTestHead(t, DefaultBlockDuration, compression.None, false)
sparseHead, _ := newTestHead(t, DefaultBlockDuration, compression.None, false)
var allSparseSeries []struct {
baseLabels labels.Labels
hists []*histogram.Histogram
}
for sid, schema := range allSchemas {
for i := 0; i < c.numSeriesPerSchema; i++ {
lbls := labels.FromStrings(
"__name__", fmt.Sprintf("rpc_durations_%d_histogram_seconds", i),
"instance", "localhost:8080",
"job", fmt.Sprintf("sparse_histogram_schema_%s", schemaDescription[sid]),
)
allSparseSeries = append(allSparseSeries, struct {
baseLabels labels.Labels
hists []*histogram.Histogram
}{baseLabels: lbls, hists: generateCustomHistograms(numHistograms, c.numBuckets, c.numSpans, c.gapBetweenSpans, schema)})
}
}
oldApp := oldHead.Appender(context.Background())
sparseApp := sparseHead.Appender(context.Background())
numOldSeriesPerHistogram := 0
var oldULIDs []ulid.ULID
var sparseULIDs []ulid.ULID
var wg sync.WaitGroup
wg.Go(func() {
// Ingest sparse histograms.
for _, ah := range allSparseSeries {
var (
ref storage.SeriesRef
err error
)
for i := range numHistograms {
ts := int64(i) * timeStep
ref, err = sparseApp.AppendHistogram(ref, ah.baseLabels, ts, ah.hists[i], nil)
require.NoError(t, err)
}
}
require.NoError(t, sparseApp.Commit())
// Sparse head compaction.
mint := sparseHead.MinTime()
maxt := sparseHead.MaxTime() + 1 // Block intervals are half-open: [b.MinTime, b.MaxTime).
compactor, err := NewLeveledCompactor(context.Background(), nil, nil, []int64{DefaultBlockDuration}, chunkenc.NewPool(), nil)
require.NoError(t, err)
sparseULIDs, err = compactor.Write(sparseHead.opts.ChunkDirRoot, sparseHead, mint, maxt, nil)
require.NoError(t, err)
require.Len(t, sparseULIDs, 1)
})
wg.Add(1)
go func(c testcase) {
defer wg.Done()
// Ingest histograms the old way.
for _, ah := range allSparseSeries {
refs := make([]storage.SeriesRef, c.numBuckets+((c.numSpans-1)*c.gapBetweenSpans))
for i := range numHistograms {
ts := int64(i) * timeStep
h := ah.hists[i]
numOldSeriesPerHistogram = 0
it := h.CumulativeBucketIterator()
itIdx := 0
var err error
for it.Next() {
numOldSeriesPerHistogram++
b := it.At()
lbls := labels.NewBuilder(ah.baseLabels).Set("le", fmt.Sprintf("%.16f", b.Upper)).Labels()
refs[itIdx], err = oldApp.Append(refs[itIdx], lbls, ts, float64(b.Count))
require.NoError(t, err)
itIdx++
}
baseName := ah.baseLabels.Get(labels.MetricName)
// _count metric.
countLbls := labels.NewBuilder(ah.baseLabels).Set(labels.MetricName, baseName+"_count").Labels()
_, err = oldApp.Append(0, countLbls, ts, float64(h.Count))
require.NoError(t, err)
numOldSeriesPerHistogram++
// _sum metric.
sumLbls := labels.NewBuilder(ah.baseLabels).Set(labels.MetricName, baseName+"_sum").Labels()
_, err = oldApp.Append(0, sumLbls, ts, h.Sum)
require.NoError(t, err)
numOldSeriesPerHistogram++
}
}
require.NoError(t, oldApp.Commit())
// Old head compaction.
mint := oldHead.MinTime()
maxt := oldHead.MaxTime() + 1 // Block intervals are half-open: [b.MinTime, b.MaxTime).
compactor, err := NewLeveledCompactor(context.Background(), nil, nil, []int64{DefaultBlockDuration}, chunkenc.NewPool(), nil)
require.NoError(t, err)
oldULIDs, err = compactor.Write(oldHead.opts.ChunkDirRoot, oldHead, mint, maxt, nil)
require.NoError(t, err)
require.Len(t, oldULIDs, 1)
}(c)
wg.Wait()
oldBlockDir := filepath.Join(oldHead.opts.ChunkDirRoot, oldULIDs[0].String())
sparseBlockDir := filepath.Join(sparseHead.opts.ChunkDirRoot, sparseULIDs[0].String())
oldSize, err := fileutil.DirSize(oldBlockDir)
require.NoError(t, err)
oldIndexSize, err := fileutil.DirSize(filepath.Join(oldBlockDir, "index"))
require.NoError(t, err)
oldChunksSize, err := fileutil.DirSize(filepath.Join(oldBlockDir, "chunks"))
require.NoError(t, err)
sparseSize, err := fileutil.DirSize(sparseBlockDir)
require.NoError(t, err)
sparseIndexSize, err := fileutil.DirSize(filepath.Join(sparseBlockDir, "index"))
require.NoError(t, err)
sparseChunksSize, err := fileutil.DirSize(filepath.Join(sparseBlockDir, "chunks"))
require.NoError(t, err)
summaries = append(summaries, testSummary{
oldBlockTotalSeries: len(allSchemas) * c.numSeriesPerSchema * numOldSeriesPerHistogram,
oldBlockIndexSize: oldIndexSize,
oldBlockChunksSize: oldChunksSize,
oldBlockTotalSize: oldSize,
sparseBlockTotalSeries: len(allSchemas) * c.numSeriesPerSchema,
sparseBlockIndexSize: sparseIndexSize,
sparseBlockChunksSize: sparseChunksSize,
sparseBlockTotalSize: sparseSize,
numBuckets: c.numBuckets,
numSpans: c.numSpans,
gapBetweenSpans: c.gapBetweenSpans,
})
})
}
for _, s := range summaries {
fmt.Printf(`
Meta: NumBuckets=%d, NumSpans=%d, GapBetweenSpans=%d
Old Block: NumSeries=%d, IndexSize=%d, ChunksSize=%d, TotalSize=%d
Sparse Block: NumSeries=%d, IndexSize=%d, ChunksSize=%d, TotalSize=%d
Savings: Index=%.2f%%, Chunks=%.2f%%, Total=%.2f%%
`,
s.numBuckets, s.numSpans, s.gapBetweenSpans,
s.oldBlockTotalSeries, s.oldBlockIndexSize, s.oldBlockChunksSize, s.oldBlockTotalSize,
s.sparseBlockTotalSeries, s.sparseBlockIndexSize, s.sparseBlockChunksSize, s.sparseBlockTotalSize,
100*(1-float64(s.sparseBlockIndexSize)/float64(s.oldBlockIndexSize)),
100*(1-float64(s.sparseBlockChunksSize)/float64(s.oldBlockChunksSize)),
100*(1-float64(s.sparseBlockTotalSize)/float64(s.oldBlockTotalSize)),
)
}
}
func generateCustomHistograms(numHists, numBuckets, numSpans, gapBetweenSpans, schema int) (r []*histogram.Histogram) {
// First histogram with all the settings.
h := &histogram.Histogram{
Sum: 1000 * rand.Float64(),
Schema: int32(schema),
}
// Generate spans.
h.PositiveSpans = []histogram.Span{
{Offset: int32(rand.Intn(10)), Length: uint32(numBuckets)},
}
if numSpans > 1 {
spanWidth := numBuckets / numSpans
// First span gets those additional buckets.
h.PositiveSpans[0].Length = uint32(spanWidth + (numBuckets - spanWidth*numSpans))
for i := 0; i < numSpans-1; i++ {
h.PositiveSpans = append(h.PositiveSpans, histogram.Span{Offset: int32(rand.Intn(gapBetweenSpans) + 1), Length: uint32(spanWidth)})
}
}
// Generate buckets.
v := int64(rand.Intn(30) + 1)
h.PositiveBuckets = []int64{v}
count := v
firstHistValues := []int64{v}
for i := 0; i < numBuckets-1; i++ {
delta := int64(rand.Intn(20))
if rand.Int()%2 == 0 && firstHistValues[len(firstHistValues)-1] > delta {
// Randomly making delta negative such that curr value will be >0.
delta = -delta
}
currVal := firstHistValues[len(firstHistValues)-1] + delta
count += currVal
firstHistValues = append(firstHistValues, currVal)
h.PositiveBuckets = append(h.PositiveBuckets, delta)
}
h.Count = uint64(count)
r = append(r, h)
// Remaining histograms with same spans but changed bucket values.
for j := 0; j < numHists-1; j++ {
newH := h.Copy()
newH.Sum = float64(j+1) * 1000 * rand.Float64()
// Generate buckets.
count := int64(0)
currVal := int64(0)
for i := range newH.PositiveBuckets {
delta := int64(rand.Intn(10))
if i == 0 {
newH.PositiveBuckets[i] += delta
currVal = newH.PositiveBuckets[i]
continue
}
currVal += newH.PositiveBuckets[i]
if rand.Int()%2 == 0 && (currVal-delta) > firstHistValues[i] {
// Randomly making delta negative such that curr value will be >0
// and above the previous count since we are not doing resets here.
delta = -delta
}
newH.PositiveBuckets[i] += delta
currVal += delta
count += currVal
}
newH.Count = uint64(count)
r = append(r, newH)
h = newH
}
return r
}
func TestCompactBlockMetas(t *testing.T) {
parent1 := ulid.MustNew(100, nil)
parent2 := ulid.MustNew(200, nil)
parent3 := ulid.MustNew(300, nil)
parent4 := ulid.MustNew(400, nil)
input := []*BlockMeta{
{ULID: parent1, MinTime: 1000, MaxTime: 2000, Compaction: BlockMetaCompaction{Level: 2, Sources: []ulid.ULID{ulid.MustNew(1, nil), ulid.MustNew(10, nil)}}},
{ULID: parent2, MinTime: 200, MaxTime: 500, Compaction: BlockMetaCompaction{Level: 1}},
{ULID: parent3, MinTime: 500, MaxTime: 2500, Compaction: BlockMetaCompaction{Level: 3, Sources: []ulid.ULID{ulid.MustNew(5, nil), ulid.MustNew(6, nil)}}},
{ULID: parent4, MinTime: 100, MaxTime: 900, Compaction: BlockMetaCompaction{Level: 1}},
}
outUlid := ulid.MustNew(1000, nil)
output := CompactBlockMetas(outUlid, input...)
expected := &BlockMeta{
ULID: outUlid,
MinTime: 100,
MaxTime: 2500,
Stats: BlockStats{},
Compaction: BlockMetaCompaction{
Level: 4,
Sources: []ulid.ULID{ulid.MustNew(1, nil), ulid.MustNew(5, nil), ulid.MustNew(6, nil), ulid.MustNew(10, nil)},
Parents: []BlockDesc{
{ULID: parent1, MinTime: 1000, MaxTime: 2000},
{ULID: parent2, MinTime: 200, MaxTime: 500},
{ULID: parent3, MinTime: 500, MaxTime: 2500},
{ULID: parent4, MinTime: 100, MaxTime: 900},
},
},
}
require.Equal(t, expected, output)
}
// TestCompactBlockMetasHints verifies that CompactBlockMetas propagates the
// compaction hints to the merged block: the from-stale-series hint is kept when
// any source carries it (the planner only ever groups stale with stale), and the
// from-out-of-order hint is kept only when every source carries it (out-of-order
// blocks may be co-compacted with in-order blocks). Dropping either hint would
// wrongly advance inOrderBlocksMaxTime. See #18379.
func TestCompactBlockMetasHints(t *testing.T) {
parent1 := ulid.MustNew(100, nil)
parent2 := ulid.MustNew(200, nil)
outUlid := ulid.MustNew(1000, nil)
staleMeta := func(u ulid.ULID) *BlockMeta {
m := &BlockMeta{ULID: u, MinTime: 0, MaxTime: 1000, Compaction: BlockMetaCompaction{Level: 1}}
m.Compaction.SetStaleSeries()
return m
}
oooMeta := func(u ulid.ULID) *BlockMeta {
m := &BlockMeta{ULID: u, MinTime: 0, MaxTime: 1000, Compaction: BlockMetaCompaction{Level: 1}}
m.Compaction.SetOutOfOrder()
return m
}
plainMeta := func(u ulid.ULID) *BlockMeta {
return &BlockMeta{ULID: u, MinTime: 0, MaxTime: 1000, Compaction: BlockMetaCompaction{Level: 1}}
}
// staleOOOMeta is a single source carrying BOTH the from-stale-series and the
// from-out-of-order hints at once.
staleOOOMeta := func(u ulid.ULID) *BlockMeta {
m := &BlockMeta{ULID: u, MinTime: 0, MaxTime: 1000, Compaction: BlockMetaCompaction{Level: 1}}
m.Compaction.SetStaleSeries()
m.Compaction.SetOutOfOrder()
return m
}
t.Run("single source carrying both hints keeps both", func(t *testing.T) {
// One source with both hints: stale survives by any-source semantics, and
// out-of-order survives because the single (only) source is out-of-order, so
// every source is out-of-order.
out := CompactBlockMetas(outUlid, staleOOOMeta(parent1))
require.True(t, out.Compaction.FromStaleSeries(), "merged block must keep the from-stale-series hint")
require.True(t, out.Compaction.FromOutOfOrder(), "merged block must keep the from-out-of-order hint when the only source is out-of-order")
})
t.Run("stale+ooo source with an ooo-only source keeps both", func(t *testing.T) {
// Stale survives by any-source semantics; out-of-order survives because
// every source (one stale+ooo, one ooo-only) is out-of-order.
out := CompactBlockMetas(outUlid, staleOOOMeta(parent1), oooMeta(parent2))
require.True(t, out.Compaction.FromStaleSeries(), "merged block must keep the from-stale-series hint when any source is stale")
require.True(t, out.Compaction.FromOutOfOrder(), "merged block must keep the from-out-of-order hint when every source is out-of-order")
})
t.Run("stale+ooo source with a plain source keeps stale and drops ooo", func(t *testing.T) {
// Stale survives by any-source semantics; out-of-order is dropped because the
// plain source is in-order, so not every source is out-of-order.
out := CompactBlockMetas(outUlid, staleOOOMeta(parent1), plainMeta(parent2))
require.True(t, out.Compaction.FromStaleSeries(), "merged block must keep the from-stale-series hint when any source is stale")
require.False(t, out.Compaction.FromOutOfOrder(), "merged block must drop the from-out-of-order hint when a source is in-order")
})
t.Run("stale hint is preserved when a source is stale", func(t *testing.T) {
out := CompactBlockMetas(outUlid, staleMeta(parent1), staleMeta(parent2))
require.True(t, out.Compaction.FromStaleSeries(), "merged block must keep the from-stale-series hint")
require.False(t, out.Compaction.FromOutOfOrder())
})
t.Run("out-of-order hint is preserved when all sources are out-of-order", func(t *testing.T) {
out := CompactBlockMetas(outUlid, oooMeta(parent1), oooMeta(parent2))
require.True(t, out.Compaction.FromOutOfOrder(), "merged block must keep the from-out-of-order hint when all sources are out-of-order")
require.False(t, out.Compaction.FromStaleSeries())
})
t.Run("out-of-order hint is dropped when mixed with in-order", func(t *testing.T) {
out := CompactBlockMetas(outUlid, oooMeta(parent1), plainMeta(parent2))
require.False(t, out.Compaction.FromOutOfOrder(), "merged block must drop the from-out-of-order hint when it also contains in-order data")
})
t.Run("stale kept and out-of-order dropped when sources mix stale and out-of-order", func(t *testing.T) {
// Stale uses any-source semantics, so the stale hint survives; out-of-order
// uses all-sources semantics, so a mix of stale and out-of-order is not all
// out-of-order and the out-of-order hint is dropped.
out := CompactBlockMetas(outUlid, staleMeta(parent1), oooMeta(parent2))
require.True(t, out.Compaction.FromStaleSeries(), "merged block must keep the from-stale-series hint when any source is stale")
require.False(t, out.Compaction.FromOutOfOrder(), "merged block must drop the from-out-of-order hint when not every source is out-of-order")
})
t.Run("no hint when no source carries one", func(t *testing.T) {
out := CompactBlockMetas(outUlid, plainMeta(parent1), plainMeta(parent2))
require.False(t, out.Compaction.FromStaleSeries())
require.False(t, out.Compaction.FromOutOfOrder())
})
}
func TestCompactEmptyResultBlockWithTombstone(t *testing.T) {
ctx := context.Background()
tmpdir := t.TempDir()
blockDir := createBlock(t, tmpdir, genSeries(1, 1, 0, 10))
block, err := OpenBlock(nil, blockDir, nil, nil)
require.NoError(t, err)
// Write tombstone covering the whole block.
err = block.Delete(ctx, 0, 10, labels.MustNewMatcher(labels.MatchEqual, defaultLabelName, "0"))
require.NoError(t, err)
c, err := NewLeveledCompactor(ctx, nil, promslog.NewNopLogger(), []int64{0}, nil, nil)
require.NoError(t, err)
ulids, err := c.Compact(tmpdir, []string{blockDir}, []*Block{block})
require.NoError(t, err)
require.Nil(t, ulids)
require.NoError(t, block.Close())
}
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)
db.EnableCompactions()
// First trigger starts the delay period; no compaction yet.
triggerAutoCompaction()
require.NotZero(t, getDelayStart())
require.Equal(t, 0.0, compactorRanCount(db))
// Compaction doesn't run before the delay expires.
time.Sleep(delay / 2)
triggerAutoCompaction()
require.Equal(t, 0.0, compactorRanCount(db))
// Compaction runs once the delay expires.
time.Sleep(delay / 2)
triggerAutoCompaction()
// One compaction was run and one block was produced.
require.Equal(t, 1.0, compactorRanCount(db))
// The compaction delay doesn't block or wait on the trigger;
// 3 triggers were processed above without blocking.
require.GreaterOrEqual(t, prom_testutil.ToFloat64(db.metrics.compactionsTriggered)-prom_testutil.ToFloat64(db.metrics.compactionsSkipped), 3.0)
// The delay doesn't change the head blocks alignment.
require.Equal(t, db.compactor.(*LeveledCompactor).ranges[0]+1, db.head.MinTime())
// Second compaction: expect 2 more blocks.
// This ensures that the logic for compaction delay doesn't only work for the first compaction, but also takes into account the future compactions.
// This also ensures that no delay happens between consecutive compactions.
db.DisableCompactions()
appendSamples(t, db, 31, 41)
db.EnableCompactions()
triggerAutoCompaction()
require.Equal(t, 1.0, compactorRanCount(db))
time.Sleep(delay / 2)
triggerAutoCompaction()
require.Equal(t, 1.0, compactorRanCount(db))
time.Sleep(delay / 2)
triggerAutoCompaction()
require.Equal(t, 3.0, compactorRanCount(db))
// This test covers a special case. If auto compaction is in a delay period and a manual compaction is triggered,
// auto compaction should stop waiting for the delay if the head is no longer compactable.
// Of course, if the head is still compactable after the manual compaction, auto compaction will continue waiting for the same delay.
db.DisableCompactions()
appendSamples(t, db, 51)
require.True(t, db.head.compactable())
db.EnableCompactions()
// Trigger an auto compaction.
triggerAutoCompaction()
// That made auto compaction start waiting for the delay.
require.NotZero(t, getDelayStart())
// Trigger a manual compaction.
require.NoError(t, db.CompactHead(NewRangeHead(db.Head(), 0, 50.0)))
require.Equal(t, 4.0, compactorRanCount(db))
// Re-trigger an auto compaction.
triggerAutoCompaction()
// That made auto compaction stop waiting for the delay.
require.Zero(t, getDelayStart())
})
})
}
// 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)
})
}
}
Reference in a new issue View git blame Copy permalink