diff --git a/ingot.go b/ingot.go index b7043ed..4beb9f0 100644 --- a/ingot.go +++ b/ingot.go @@ -2,6 +2,7 @@ package ingot import ( + "context" "fmt" "os" "path/filepath" @@ -11,25 +12,58 @@ import ( "git.dvdt.dev/david/ingot/internal/block" "git.dvdt.dev/david/ingot/internal/chunkenc" + "git.dvdt.dev/david/ingot/internal/compact" "git.dvdt.dev/david/ingot/internal/head" "git.dvdt.dev/david/ingot/internal/postings" "git.dvdt.dev/david/ingot/internal/wal" "git.dvdt.dev/david/ingot/labels" ) +// Default compaction level durations in milliseconds. +var defaultLevels = []int64{ + 2 * 3600 * 1000, // 2h + 8 * 3600 * 1000, // 8h + 32 * 3600 * 1000, // 32h +} + // DB is an embedded time-series database. type DB struct { - dataDir string - opts Options - head *head.Head - blocks []*block.Reader // sorted by MinTime - mu sync.RWMutex // protects blocks slice + dataDir string + opts Options + head *head.Head + blocks []*block.Reader // sorted by MinTime + mu sync.RWMutex // protects blocks slice + compactor *compact.Compactor + compactCtx context.Context + compactCancel context.CancelFunc + compactWg sync.WaitGroup } // Options configures a DB. type Options struct { Retention time.Duration BlockDuration time.Duration + // Clock returns the current time in milliseconds. Defaults to + // time.Now().UnixMilli(). Injected for testing with simulated time. + Clock func() int64 +} + +func (o *Options) clock() func() int64 { + if o.Clock != nil { + return o.Clock + } + return func() int64 { return time.Now().UnixMilli() } +} + +func (o *Options) blockDurationMs() int64 { + if o.BlockDuration == 0 { + return defaultLevels[0] // 2h default + } + return o.BlockDuration.Milliseconds() +} + +func (o *Options) retentionMs() int64 { + return o.Retention.Milliseconds() } // Open opens or creates a DB at the given directory. @@ -44,17 +78,28 @@ func Open(dataDir string, opts Options) (*DB, error) { return nil, fmt.Errorf("ingot: open head: %w", err) } + ctx, cancel := context.WithCancel(context.Background()) + db := &DB{ - dataDir: dataDir, - opts: opts, - head: h, + dataDir: dataDir, + opts: opts, + head: h, + compactCtx: ctx, + compactCancel: cancel, } + db.compactor = compact.New(dataDir, defaultLevels, opts.retentionMs(), opts.clock()) + if err := db.loadBlocks(); err != nil { + cancel() h.Close() return nil, fmt.Errorf("ingot: load blocks: %w", err) } + // Start background compaction goroutine. + db.compactWg.Add(1) + go db.compactLoop() + return db, nil } @@ -99,6 +144,7 @@ func (db *DB) Querier(mint, maxt int64) (*Querier, error) { var overlapping []*block.Reader for _, b := range db.blocks { if b.Meta.MaxTime >= mint && b.Meta.MinTime <= maxt { + b.Ref() overlapping = append(overlapping, b) } } @@ -139,8 +185,132 @@ func (db *DB) FlushOlderThan(maxT int64) (string, error) { return ulid, nil } +// RunCompaction performs a single compaction cycle. Exported for testing. +func (db *DB) RunCompaction() error { + db.mu.RLock() + snapshot := make([]*block.Reader, len(db.blocks)) + copy(snapshot, db.blocks) + db.mu.RUnlock() + + group := db.compactor.Plan(snapshot) + if group == nil { + return nil + } + + newULID, err := db.compactor.Compact(group.Sources) + if err != nil { + return fmt.Errorf("ingot: compact: %w", err) + } + + newBlock, err := block.Open(filepath.Join(db.dataDir, newULID)) + if err != nil { + return fmt.Errorf("ingot: open compacted block: %w", err) + } + + // Build a set of source ULIDs for fast lookup. + sourceSet := make(map[string]struct{}, len(group.Sources)) + for _, s := range group.Sources { + sourceSet[s.Meta.ULID] = struct{}{} + } + + // Swap blocks under short lock. + db.mu.Lock() + var remaining []*block.Reader + for _, b := range db.blocks { + if _, ok := sourceSet[b.Meta.ULID]; !ok { + remaining = append(remaining, b) + } + } + remaining = append(remaining, newBlock) + sort.Slice(remaining, func(i, j int) bool { + return remaining[i].Meta.MinTime < remaining[j].Meta.MinTime + }) + db.blocks = remaining + db.mu.Unlock() + + // Condemn and release source blocks. + for _, src := range group.Sources { + dir := src.Dir() + src.Condemn() + if src.Release() { + os.RemoveAll(dir) + } + } + + return nil +} + +// ApplyRetention drops blocks whose data is older than the retention window. +// Exported for testing. +func (db *DB) ApplyRetention() { + if db.opts.Retention == 0 { + return + } + + db.mu.RLock() + snapshot := make([]*block.Reader, len(db.blocks)) + copy(snapshot, db.blocks) + db.mu.RUnlock() + + expired := db.compactor.Expired(snapshot) + if len(expired) == 0 { + return + } + + expiredSet := make(map[string]struct{}, len(expired)) + for _, b := range expired { + expiredSet[b.Meta.ULID] = struct{}{} + } + + db.mu.Lock() + var remaining []*block.Reader + for _, b := range db.blocks { + if _, ok := expiredSet[b.Meta.ULID]; !ok { + remaining = append(remaining, b) + } + } + db.blocks = remaining + db.mu.Unlock() + + for _, b := range expired { + dir := b.Dir() + b.Condemn() + if b.Release() { + os.RemoveAll(dir) + } + } +} + +// compactLoop runs in a background goroutine, periodically flushing the +// head and compacting blocks. +func (db *DB) compactLoop() { + defer db.compactWg.Done() + ticker := time.NewTicker(1 * time.Minute) + defer ticker.Stop() + for { + select { + case <-db.compactCtx.Done(): + return + case <-ticker.C: + db.autoFlush() + db.RunCompaction() + db.ApplyRetention() + } + } +} + +// autoFlush flushes sealed head chunks older than BlockDuration. +func (db *DB) autoFlush() { + now := db.opts.clock()() + cutoff := now - db.opts.blockDurationMs() + db.FlushOlderThan(cutoff) +} + // Close closes the DB, releasing all resources. func (db *DB) Close() error { + db.compactCancel() + db.compactWg.Wait() + var firstErr error if err := db.head.Close(); err != nil { firstErr = err @@ -235,8 +405,15 @@ func (q *Querier) Select(matchers ...*labels.Matcher) SeriesSet { return &sliceSeriesSet{series: entries} } -// Close is a no-op for now (refcounting is M5). +// Close releases block references held by this querier. func (q *Querier) Close() error { + for _, b := range q.blocks { + dir := b.Dir() + if b.Release() { + os.RemoveAll(dir) + } + } + q.blocks = nil return nil } @@ -384,15 +561,15 @@ func (s *resultSeries) Iterator() SampleIterator { // mergedSampleIterator merges multiple ChunkIterators in order, deduplicating // timestamps. Earlier iterators (blocks) win over later ones (head). type mergedSampleIterator struct { - iters []chunkenc.ChunkIterator - mint int64 - maxt int64 - cur int - lastT int64 - curT int64 - curV float64 + iters []chunkenc.ChunkIterator + mint int64 + maxt int64 + cur int + lastT int64 + curT int64 + curV float64 started bool - err error + err error } func (m *mergedSampleIterator) Next() bool { diff --git a/ingot_test.go b/ingot_test.go index ec404e3..eb82092 100644 --- a/ingot_test.go +++ b/ingot_test.go @@ -2,7 +2,9 @@ package ingot import ( "math" + "os" "testing" + "time" "git.dvdt.dev/david/ingot/labels" "github.com/stretchr/testify/assert" @@ -467,81 +469,256 @@ type queryCase struct { matchers []*labels.Matcher } -func TestDBAppenderAPI(t *testing.T) { - db := openTestDB(t) +func TestDBLifecycle(t *testing.T) { + tests := []struct { + name string + setup func(t *testing.T, dir string) *DB + wantSampleCount int + wantSeriesCount int + matchers []*labels.Matcher + mint int64 + maxt int64 + }{ + { + name: "append_and_query_back", + setup: func(t *testing.T, dir string) *DB { + db, err := Open(dir, Options{}) + require.NoError(t, err) + app := db.Appender() + ref, err := app.Append(0, labels.FromStrings("__name__", "temp", "room", "office"), 1000, 71.3) + require.NoError(t, err) + _, err = app.Append(ref, nil, 2000, 71.4) + require.NoError(t, err) + require.NoError(t, app.Commit()) + return db + }, + wantSampleCount: 2, + wantSeriesCount: 1, + matchers: []*labels.Matcher{labels.MustNewMatcher(labels.MatchEqual, "room", "office")}, + mint: math.MinInt64, + maxt: math.MaxInt64, + }, + { + name: "reopen_with_blocks", + setup: func(t *testing.T, dir string) *DB { + db, err := Open(dir, Options{}) + require.NoError(t, err) + app := db.Appender() + ref, err := app.Append(0, labels.FromStrings("__name__", "temp"), 0, 0) + require.NoError(t, err) + for i := 1; i < 250; i++ { + _, err = app.Append(ref, nil, int64(i*15000), float64(i)) + require.NoError(t, err) + } + require.NoError(t, app.Commit()) + _, err = db.FlushOlderThan(math.MaxInt64) + require.NoError(t, err) + require.NoError(t, db.Close()) - app := db.Appender() - ref, err := app.Append(0, labels.FromStrings("__name__", "temp", "room", "office"), 1000, 71.3) - require.NoError(t, err) - assert.NotZero(t, ref) + db2, err := Open(dir, Options{}) + require.NoError(t, err) + return db2 + }, + wantSampleCount: 250, + wantSeriesCount: 1, + matchers: []*labels.Matcher{labels.MustNewMatcher(labels.MatchEqual, "__name__", "temp")}, + mint: math.MinInt64, + maxt: math.MaxInt64, + }, + { + name: "compacted_blocks_queryable", + setup: func(t *testing.T, dir string) *DB { + db, err := Open(dir, Options{}) + require.NoError(t, err) + const samplesPerBlock = 130 + ref := uint64(0) + for b := 0; b < 4; b++ { + app := db.Appender() + for i := 0; i < samplesPerBlock; i++ { + ts := int64((b*samplesPerBlock + i) * 15000) + r, err := app.Append(ref, labels.FromStrings("__name__", "temp"), ts, float64(ts)) + require.NoError(t, err) + ref = r + } + require.NoError(t, app.Commit()) + _, err := db.FlushOlderThan(math.MaxInt64) + require.NoError(t, err) + } + require.NoError(t, db.RunCompaction()) + return db + }, + wantSampleCount: 520, + wantSeriesCount: 1, + matchers: []*labels.Matcher{labels.MustNewMatcher(labels.MatchEqual, "__name__", "temp")}, + mint: math.MinInt64, + maxt: math.MaxInt64, + }, + { + name: "retention_drops_old_keeps_recent", + setup: func(t *testing.T, dir string) *DB { + now := int64(100 * 3600 * 1000) + db, err := Open(dir, Options{ + Retention: 24 * time.Hour, + Clock: func() int64 { return now }, + }) + require.NoError(t, err) + // Old data (50 hours ago). + app := db.Appender() + ref, err := app.Append(0, labels.FromStrings("__name__", "old"), 50*3600*1000, 1.0) + require.NoError(t, err) + for i := 1; i < 250; i++ { + _, err = app.Append(ref, nil, int64(50*3600*1000+i*15000), float64(i)) + require.NoError(t, err) + } + require.NoError(t, app.Commit()) + _, err = db.FlushOlderThan(math.MaxInt64) + require.NoError(t, err) + // Recent data (1 hour ago). + app = db.Appender() + ref2, err := app.Append(0, labels.FromStrings("__name__", "recent"), 99*3600*1000, 1.0) + require.NoError(t, err) + for i := 1; i < 250; i++ { + _, err = app.Append(ref2, nil, int64(99*3600*1000+i*15000), float64(i)) + require.NoError(t, err) + } + require.NoError(t, app.Commit()) + _, err = db.FlushOlderThan(math.MaxInt64) + require.NoError(t, err) + db.ApplyRetention() + return db + }, + wantSampleCount: 250, + wantSeriesCount: 1, + matchers: []*labels.Matcher{labels.MustNewMatcher(labels.MatchEqual, "__name__", "recent")}, + mint: math.MinInt64, + maxt: math.MaxInt64, + }, + } - _, err = app.Append(ref, nil, 2000, 71.4) - require.NoError(t, err) + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + dir := t.TempDir() + db := tc.setup(t, dir) + defer db.Close() - require.NoError(t, app.Commit()) + q, err := db.Querier(tc.mint, tc.maxt) + require.NoError(t, err) + defer q.Close() - // Query back. - q, err := db.Querier(math.MinInt64, math.MaxInt64) - require.NoError(t, err) - defer q.Close() - - ss := q.Select(labels.MustNewMatcher(labels.MatchEqual, "room", "office")) - require.True(t, ss.Next()) - s := ss.At() - assert.Equal(t, labels.FromStrings("__name__", "temp", "room", "office"), s.Labels()) - - it := s.Iterator() - require.True(t, it.Next()) - st, sv := it.At() - assert.Equal(t, int64(1000), st) - assert.Equal(t, 71.3, sv) - - require.True(t, it.Next()) - st, sv = it.At() - assert.Equal(t, int64(2000), st) - assert.Equal(t, 71.4, sv) - - assert.False(t, it.Next()) - assert.False(t, ss.Next()) + ss := q.Select(tc.matchers...) + seriesCount := 0 + sampleCount := 0 + for ss.Next() { + seriesCount++ + it := ss.At().Iterator() + for it.Next() { + sampleCount++ + } + require.NoError(t, it.Err()) + } + require.NoError(t, ss.Err()) + assert.Equal(t, tc.wantSeriesCount, seriesCount, "series count") + assert.Equal(t, tc.wantSampleCount, sampleCount, "sample count") + }) + } } -func TestDBReopenWithBlocks(t *testing.T) { - dir := t.TempDir() - - // Write data, flush, close. - db, err := Open(dir, Options{}) - require.NoError(t, err) - - app := db.Appender() - ref, err := app.Append(0, labels.FromStrings("__name__", "temp"), 0, 0) - require.NoError(t, err) - for i := 1; i < 250; i++ { - _, err = app.Append(ref, nil, int64(i*15000), float64(i)) - require.NoError(t, err) +func TestQueryDuringCompaction(t *testing.T) { + tests := []struct { + name string + numBlocks int + samplesPerBlock int + }{ + { + name: "four_blocks", + numBlocks: 4, + samplesPerBlock: 130, + }, + { + name: "two_blocks", + numBlocks: 2, + samplesPerBlock: 130, + }, } - require.NoError(t, app.Commit()) - _, err = db.FlushOlderThan(math.MaxInt64) - require.NoError(t, err) - require.NoError(t, db.Close()) + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + dir := t.TempDir() + db, err := Open(dir, Options{}) + require.NoError(t, err) + defer db.Close() - // Reopen. - db2, err := Open(dir, Options{}) - require.NoError(t, err) - defer db2.Close() + ref := uint64(0) + for b := 0; b < tc.numBlocks; b++ { + app := db.Appender() + for i := 0; i < tc.samplesPerBlock; i++ { + ts := int64((b*tc.samplesPerBlock + i) * 15000) + r, err := app.Append(ref, labels.FromStrings("__name__", "temp"), ts, float64(ts)) + require.NoError(t, err) + ref = r + } + require.NoError(t, app.Commit()) + _, err := db.FlushOlderThan(math.MaxInt64) + require.NoError(t, err) + } - // Should find data in block. - q, err := db2.Querier(math.MinInt64, math.MaxInt64) - require.NoError(t, err) - defer q.Close() + // Snapshot source block dirs. + db.mu.RLock() + sourceDirs := make([]string, len(db.blocks)) + for i, b := range db.blocks { + sourceDirs[i] = b.Dir() + } + db.mu.RUnlock() - ss := q.Select(labels.MustNewMatcher(labels.MatchEqual, "__name__", "temp")) - require.True(t, ss.Next()) - it := ss.At().Iterator() - count := 0 - for it.Next() { - count++ + // Start query holding refs on all blocks. + q, err := db.Querier(math.MinInt64, math.MaxInt64) + require.NoError(t, err) + + ss := q.Select(labels.MustNewMatcher(labels.MatchEqual, "__name__", "temp")) + require.True(t, ss.Next()) + it := ss.At().Iterator() + require.True(t, it.Next()) + + // Compact while query is open. + err = db.RunCompaction() + require.NoError(t, err) + + // Source dirs still exist (query holds refs). + for _, d := range sourceDirs { + _, statErr := os.Stat(d) + assert.NoError(t, statErr, "source dir should exist while query holds ref") + } + + // Finish iterating — all data still readable. + count := 1 + for it.Next() { + count++ + } + require.NoError(t, it.Err()) + assert.Equal(t, tc.samplesPerBlock*tc.numBlocks, count, "all samples readable during compaction") + + // Close querier — source dirs should be deleted. + require.NoError(t, q.Close()) + for _, d := range sourceDirs { + _, statErr := os.Stat(d) + assert.True(t, os.IsNotExist(statErr), "source dir should be deleted after query close") + } + + // Compacted block still queryable. + q2, err := db.Querier(math.MinInt64, math.MaxInt64) + require.NoError(t, err) + defer q2.Close() + ss2 := q2.Select(labels.MustNewMatcher(labels.MatchEqual, "__name__", "temp")) + count = 0 + for ss2.Next() { + it2 := ss2.At().Iterator() + for it2.Next() { + count++ + } + require.NoError(t, it2.Err()) + } + require.NoError(t, ss2.Err()) + assert.Equal(t, tc.samplesPerBlock*tc.numBlocks, count, "compacted block has all samples") + }) } - require.NoError(t, it.Err()) - assert.Equal(t, 250, count, "should find all samples: 240 from block + 10 from head WAL replay") } diff --git a/internal/block/block_test.go b/internal/block/block_test.go index ca3b28a..31330a7 100644 --- a/internal/block/block_test.go +++ b/internal/block/block_test.go @@ -8,6 +8,7 @@ import ( "testing" "git.dvdt.dev/david/ingot/internal/chunkenc" + "git.dvdt.dev/david/ingot/internal/index" "git.dvdt.dev/david/ingot/labels" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -181,13 +182,49 @@ func TestBlockRoundTrip(t *testing.T) { } } -func TestBlockMultipleChunksPerSeries(t *testing.T) { - dataDir := t.TempDir() - +func TestBlockSeriesChunkIterator(t *testing.T) { chunk1Samples := []sample{s(1000, 1.0), s(1015, 2.0), s(1030, 3.0)} chunk2Samples := []sample{s(2000, 4.0), s(2015, 5.0), s(2030, 6.0)} - allSamples := append(chunk1Samples, chunk2Samples...) + tests := []struct { + name string + ref uint64 + mint int64 + maxt int64 + wantSamples []sample + }{ + { + name: "full_range", + ref: 1, + mint: math.MinInt64, + maxt: math.MaxInt64, + wantSamples: append(chunk1Samples, chunk2Samples...), + }, + { + name: "second_chunk_only", + ref: 1, + mint: 2000, + maxt: 3000, + wantSamples: chunk2Samples, + }, + { + name: "no_overlap", + ref: 1, + mint: 5000, + maxt: 6000, + wantSamples: nil, + }, + { + name: "unknown_ref", + ref: 999, + mint: math.MinInt64, + maxt: math.MaxInt64, + wantSamples: nil, + }, + } + + // Setup: create a block with two chunks for series ref=1. + dataDir := t.TempDir() flushData := []SeriesFlush{ { Ref: 1, @@ -198,71 +235,76 @@ func TestBlockMultipleChunksPerSeries(t *testing.T) { }, }, } - ulid, err := Flush(dataDir, flushData) require.NoError(t, err) - r, err := Open(filepath.Join(dataDir, ulid)) require.NoError(t, err) defer r.Close() - // Full range. - it, err := r.SeriesChunkIterator(1, math.MinInt64, math.MaxInt64) - require.NoError(t, err) - got := collectIterator(t, it) - require.Equal(t, len(allSamples), len(got)) - for i, want := range allSamples { - assert.Equal(t, want.t, got[i].t, "sample %d t", i) - assert.Equal(t, want.vBits, got[i].vBits, "sample %d v", i) + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + it, err := r.SeriesChunkIterator(tc.ref, tc.mint, tc.maxt) + require.NoError(t, err) + got := collectIterator(t, it) + assert.Equal(t, len(tc.wantSamples), len(got), "sample count") + for i, want := range tc.wantSamples { + assert.Equal(t, want.t, got[i].t, "sample %d t", i) + assert.Equal(t, want.vBits, got[i].vBits, "sample %d v", i) + } + }) } - - // Query only second chunk's range. - it, err = r.SeriesChunkIterator(1, 2000, 3000) - require.NoError(t, err) - got = collectIterator(t, it) - require.Equal(t, len(chunk2Samples), len(got)) - for i, want := range chunk2Samples { - assert.Equal(t, want.t, got[i].t, "sample %d t", i) - assert.Equal(t, want.vBits, got[i].vBits, "sample %d v", i) - } - - // Query with no overlap. - it, err = r.SeriesChunkIterator(1, 5000, 6000) - require.NoError(t, err) - got = collectIterator(t, it) - assert.Empty(t, got) - - // Unknown ref. - it, err = r.SeriesChunkIterator(999, math.MinInt64, math.MaxInt64) - require.NoError(t, err) - assert.False(t, it.Next()) } func TestBlockMetaTimeBounds(t *testing.T) { - dataDir := t.TempDir() - - flushData := []SeriesFlush{ + tests := []struct { + name string + series []SeriesFlush + wantMinT int64 + wantMaxT int64 + }{ { - Ref: 1, - Labels: []labels.Label{{Name: "__name__", Value: "a"}}, - Chunks: []ChunkData{{MinT: 500, MaxT: 1000, Data: makeChunk(t, []sample{s(500, 1.0), s(1000, 2.0)})}}, + name: "two_series_different_ranges", + series: []SeriesFlush{ + {Ref: 1, Labels: []labels.Label{{Name: "__name__", Value: "a"}}, Chunks: []ChunkData{{MinT: 500, MaxT: 1000, Data: makeChunkFromPairs([]int64{500, 1000}, []float64{1.0, 2.0})}}}, + {Ref: 2, Labels: []labels.Label{{Name: "__name__", Value: "b"}}, Chunks: []ChunkData{{MinT: 200, MaxT: 800, Data: makeChunkFromPairs([]int64{200, 800}, []float64{3.0, 4.0})}}}, + }, + wantMinT: 200, + wantMaxT: 1000, }, { - Ref: 2, - Labels: []labels.Label{{Name: "__name__", Value: "b"}}, - Chunks: []ChunkData{{MinT: 200, MaxT: 800, Data: makeChunk(t, []sample{s(200, 3.0), s(800, 4.0)})}}, + name: "single_series", + series: []SeriesFlush{ + {Ref: 1, Labels: []labels.Label{{Name: "__name__", Value: "a"}}, Chunks: []ChunkData{{MinT: 100, MaxT: 500, Data: makeChunkFromPairs([]int64{100, 500}, []float64{1.0, 2.0})}}}, + }, + wantMinT: 100, + wantMaxT: 500, + }, + { + name: "multiple_chunks", + series: []SeriesFlush{ + {Ref: 1, Labels: []labels.Label{{Name: "__name__", Value: "a"}}, Chunks: []ChunkData{ + {MinT: 100, MaxT: 200, Data: makeChunkFromPairs([]int64{100, 200}, []float64{1.0, 2.0})}, + {MinT: 300, MaxT: 900, Data: makeChunkFromPairs([]int64{300, 900}, []float64{3.0, 4.0})}, + }}, + }, + wantMinT: 100, + wantMaxT: 900, }, } - ulid, err := Flush(dataDir, flushData) - require.NoError(t, err) + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + dataDir := t.TempDir() + ulid, err := Flush(dataDir, tc.series) + require.NoError(t, err) + r, err := Open(filepath.Join(dataDir, ulid)) + require.NoError(t, err) + defer r.Close() - r, err := Open(filepath.Join(dataDir, ulid)) - require.NoError(t, err) - defer r.Close() - - assert.Equal(t, int64(200), r.Meta.MinTime) - assert.Equal(t, int64(1000), r.Meta.MaxTime) + assert.Equal(t, tc.wantMinT, r.Meta.MinTime, "MinTime") + assert.Equal(t, tc.wantMaxT, r.Meta.MaxTime, "MaxTime") + }) + } } func TestULIDRoundTrip(t *testing.T) { @@ -276,57 +318,83 @@ func TestULIDRoundTrip(t *testing.T) { } } -func TestCorruptChunkCRC(t *testing.T) { - dataDir := t.TempDir() - - samples := []sample{s(1000, 71.3), s(1015, 71.4)} - flushData := []SeriesFlush{ +func TestBlockCorruption(t *testing.T) { + tests := []struct { + name string + corruptFunc func(t *testing.T, blockDir string, chunkRef index.ChunkRef) + wantErr error + }{ { - Ref: 1, - Labels: []labels.Label{{Name: "__name__", Value: "temp"}}, - Chunks: []ChunkData{{MinT: 1000, MaxT: 1015, Data: makeChunk(t, samples)}}, + name: "corrupt_chunk_data_byte", + corruptFunc: func(t *testing.T, blockDir string, chunkRef index.ChunkRef) { + chunkPath := filepath.Join(blockDir, chunksDirName, segmentName(int(chunkRef.Segment()))) + data, err := os.ReadFile(chunkPath) + require.NoError(t, err) + off := int(chunkRef.Offset()) + chunkEntryHeaderLen + 1 + data[off] ^= 0xFF + require.NoError(t, os.WriteFile(chunkPath, data, 0644)) + }, + wantErr: ErrCorruptChunk, + }, + { + name: "corrupt_chunk_crc", + corruptFunc: func(t *testing.T, blockDir string, chunkRef index.ChunkRef) { + chunkPath := filepath.Join(blockDir, chunksDirName, segmentName(int(chunkRef.Segment()))) + data, err := os.ReadFile(chunkPath) + require.NoError(t, err) + // Corrupt the last byte of the CRC. + off := int(chunkRef.Offset()) + chunkEntryHeaderLen + // Read dataLen to find CRC position. + dataLen := int(data[chunkRef.Offset()])*16777216 + int(data[chunkRef.Offset()+1])*65536 + + int(data[chunkRef.Offset()+2])*256 + int(data[chunkRef.Offset()+3]) + crcOff := off + dataLen + 3 // last byte of CRC + data[crcOff] ^= 0xFF + require.NoError(t, os.WriteFile(chunkPath, data, 0644)) + }, + wantErr: ErrCorruptChunk, }, } - ulid, err := Flush(dataDir, flushData) - require.NoError(t, err) + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + dataDir := t.TempDir() + samples := []sample{s(1000, 71.3), s(1015, 71.4)} + flushData := []SeriesFlush{ + { + Ref: 1, + Labels: []labels.Label{{Name: "__name__", Value: "temp"}}, + Chunks: []ChunkData{{MinT: 1000, MaxT: 1015, Data: makeChunk(t, samples)}}, + }, + } + ulid, err := Flush(dataDir, flushData) + require.NoError(t, err) - // Open block, find the chunk ref, then corrupt the chunk file. - blockDir := filepath.Join(dataDir, ulid) - r, err := Open(blockDir) - require.NoError(t, err) + blockDir := filepath.Join(dataDir, ulid) + r, err := Open(blockDir) + require.NoError(t, err) + series := r.Series() + require.Equal(t, 1, len(series)) + chunkRef := series[0].Chunks[0].Ref + r.Close() - series := r.Series() - require.Equal(t, 1, len(series)) - chunkRef := series[0].Chunks[0].Ref - r.Close() + tc.corruptFunc(t, blockDir, chunkRef) - // Corrupt chunk data on disk. - chunkSeg := chunkRef.Segment() - chunkPath := filepath.Join(blockDir, chunksDirName, segmentName(int(chunkSeg))) + r, err = Open(blockDir) + require.NoError(t, err) + defer r.Close() - // Read, corrupt a data byte, write back. - chunkFile, err := readFileBytes(chunkPath) - require.NoError(t, err) - off := int(chunkRef.Offset()) + chunkEntryHeaderLen + 1 // corrupt a data byte - if off < len(chunkFile) { - chunkFile[off] ^= 0xFF + _, err = r.ChunkIterator(chunkRef) + assert.Equal(t, tc.wantErr, err) + }) } - require.NoError(t, writeFileBytes(chunkPath, chunkFile)) - - // Re-open and try to read the corrupt chunk. - r, err = Open(blockDir) - require.NoError(t, err) - defer r.Close() - - _, err = r.ChunkIterator(chunkRef) - assert.Equal(t, ErrCorruptChunk, err) } -func readFileBytes(path string) ([]byte, error) { - return os.ReadFile(path) -} - -func writeFileBytes(path string, data []byte) error { - return os.WriteFile(path, data, 0644) +// makeChunkFromPairs creates a chunk from timestamp/value slices. +func makeChunkFromPairs(ts []int64, vs []float64) []byte { + c := chunkenc.NewXORChunk() + a, _ := c.Appender() + for i := range ts { + a.Append(ts[i], vs[i]) + } + return append([]byte(nil), c.Bytes()...) } diff --git a/internal/block/reader.go b/internal/block/reader.go index 36fe6ac..63f74b2 100644 --- a/internal/block/reader.go +++ b/internal/block/reader.go @@ -3,6 +3,7 @@ package block import ( "os" "path/filepath" + "sync/atomic" "git.dvdt.dev/david/ingot/internal/chunkenc" "git.dvdt.dev/david/ingot/internal/index" @@ -11,10 +12,12 @@ import ( // Reader provides read access to an immutable on-disk block. type Reader struct { - dir string - Meta BlockMeta - idx *index.Reader - chunks *chunkReader + dir string + Meta BlockMeta + idx *index.Reader + chunks *chunkReader + refs atomic.Int32 + condemned atomic.Bool } // Open opens a block directory for reading. Chunk files are mmap'd. @@ -39,12 +42,14 @@ func Open(dir string) (*Reader, error) { return nil, err } - return &Reader{ + r := &Reader{ dir: dir, Meta: meta, idx: idx, chunks: cr, - }, nil + } + r.refs.Store(1) // DB's ownership ref + return r, nil } // Series returns all series entries from the index. @@ -115,6 +120,30 @@ func (r *Reader) AllPostings() []uint64 { return r.idx.AllPostings() } +// Dir returns the block directory path. +func (r *Reader) Dir() string { return r.dir } + +// Ref increments the refcount. Called by Querier on snapshot. +func (r *Reader) Ref() { r.refs.Add(1) } + +// Release decrements the refcount. Returns true if the refcount hit zero +// and the block is condemned (caller should delete the directory). +func (r *Reader) Release() bool { + if r.refs.Add(-1) == 0 { + r.Close() + return r.condemned.Load() + } + return false +} + +// Condemn marks the block for directory deletion when refcount reaches zero. +func (r *Reader) Condemn() { r.condemned.Store(true) } + +// RawChunkData returns the raw chunk bytes at the given ref (for compaction). +func (r *Reader) RawChunkData(ref index.ChunkRef) ([]byte, error) { + return r.chunks.chunkData(ref) +} + // Close releases all resources (munmaps chunk files). func (r *Reader) Close() error { return r.chunks.close() diff --git a/internal/block/writer.go b/internal/block/writer.go index c9765fa..6ece077 100644 --- a/internal/block/writer.go +++ b/internal/block/writer.go @@ -30,6 +30,16 @@ type SeriesFlush struct { // // Returns the block ULID and any error. func Flush(dataDir string, series []SeriesFlush) (string, error) { + return flushBlock(dataDir, series, 1, nil) +} + +// FlushCompacted writes a new immutable block from compacted series data, +// recording the compaction level and source block ULIDs. +func FlushCompacted(dataDir string, series []SeriesFlush, level int, sources []string) (string, error) { + return flushBlock(dataDir, series, level, sources) +} + +func flushBlock(dataDir string, series []SeriesFlush, level int, sources []string) (string, error) { ulid := newULID() blockDir := filepath.Join(dataDir, ulid) @@ -49,7 +59,12 @@ func Flush(dataDir string, series []SeriesFlush) (string, error) { ) meta.ULID = ulid meta.Version = 1 - meta.Compaction = CompactionInfo{Level: 1, Sources: []string{ulid}} + meta.Compaction = CompactionInfo{Level: level} + if sources != nil { + meta.Compaction.Sources = sources + } else { + meta.Compaction.Sources = []string{ulid} + } meta.MinTime = int64(^uint64(0) >> 1) // max int64 meta.MaxTime = int64(0) diff --git a/internal/compact/compact.go b/internal/compact/compact.go new file mode 100644 index 0000000..bebfba1 --- /dev/null +++ b/internal/compact/compact.go @@ -0,0 +1,206 @@ +// Package compact implements levelled compaction and retention for ingot blocks. +package compact + +import ( + "fmt" + "sort" + + "git.dvdt.dev/david/ingot/internal/block" + "git.dvdt.dev/david/ingot/labels" +) + +// Clock returns the current time in milliseconds since epoch. +type Clock func() int64 + +// Compactor manages levelled compaction and retention. +type Compactor struct { + dataDir string + levels []int64 // level durations in ms, e.g. [2h, 8h, 32h] + retention int64 // retention window in ms (0 = disabled) + clock Clock +} + +// New creates a Compactor. levels are the compaction level durations in +// ascending order (e.g. 2h, 8h, 32h in milliseconds). retention is the +// maximum age of data in milliseconds (0 disables retention). +func New(dataDir string, levels []int64, retention int64, clock Clock) *Compactor { + return &Compactor{ + dataDir: dataDir, + levels: levels, + retention: retention, + clock: clock, + } +} + +// CompactionGroup describes a set of source blocks to compact. +type CompactionGroup struct { + Sources []*block.Reader + Level int // resulting compaction level +} + +// Plan returns the first eligible compaction group, or nil if no compaction +// is needed. Lower levels are prioritized. A group requires at least 2 +// blocks at the same compaction level whose combined time span fits within +// the next level's duration. +func (c *Compactor) Plan(blocks []*block.Reader) *CompactionGroup { + if len(blocks) < 2 { + return nil + } + + // Group blocks by compaction level. + byLevel := make(map[int][]*block.Reader) + for _, b := range blocks { + lvl := b.Meta.Compaction.Level + byLevel[lvl] = append(byLevel[lvl], b) + } + + // Sort levels ascending. + var levels []int + for lvl := range byLevel { + levels = append(levels, lvl) + } + sort.Ints(levels) + + for _, lvl := range levels { + group := c.planLevel(byLevel[lvl], lvl) + if group != nil { + return group + } + } + return nil +} + +// planLevel finds a compactable group within blocks at the same level. +func (c *Compactor) planLevel(blocks []*block.Reader, level int) *CompactionGroup { + if len(blocks) < 2 { + return nil + } + + // Sort by MinTime. + sorted := make([]*block.Reader, len(blocks)) + copy(sorted, blocks) + sort.Slice(sorted, func(i, j int) bool { + return sorted[i].Meta.MinTime < sorted[j].Meta.MinTime + }) + + // Determine the max span for the next level. + var maxSpan int64 + if level-1 < len(c.levels) && level >= 1 { + // Find the next level's duration. Level 1 blocks should compact + // into level 2 when they span up to levels[1] (8h), etc. + if level < len(c.levels) { + maxSpan = c.levels[level] + } + } + if maxSpan == 0 { + // No higher level defined, or level 0 — use the second level duration. + if level < len(c.levels) { + maxSpan = c.levels[level] + } else { + return nil // already at max level + } + } + + // Find the first group of consecutive blocks whose span fits maxSpan. + for i := 0; i < len(sorted)-1; i++ { + group := []*block.Reader{sorted[i]} + for j := i + 1; j < len(sorted); j++ { + span := sorted[j].Meta.MaxTime - sorted[i].Meta.MinTime + if span > maxSpan { + break + } + group = append(group, sorted[j]) + } + if len(group) >= 2 { + return &CompactionGroup{ + Sources: group, + Level: level + 1, + } + } + } + return nil +} + +// Compact merges source blocks into a single new block. Returns the new +// block's ULID. The caller is responsible for swapping the block set and +// releasing source blocks. +func (c *Compactor) Compact(sources []*block.Reader) (string, error) { + if len(sources) == 0 { + return "", fmt.Errorf("compact: no source blocks") + } + + merged := make(map[uint64]*mergedEntry) + + for _, src := range sources { + for _, entry := range src.Series() { + me, ok := merged[entry.Ref] + if !ok { + me = &mergedEntry{ + ref: entry.Ref, + labels: entry.Labels, + } + merged[entry.Ref] = me + } + for _, cm := range entry.Chunks { + raw, err := src.RawChunkData(cm.Ref) + if err != nil { + return "", fmt.Errorf("compact: read chunk ref %v from %s: %w", + cm.Ref, src.Meta.ULID, err) + } + // Copy the raw bytes since the source may be munmapped later. + data := make([]byte, len(raw)) + copy(data, raw) + me.chunks = append(me.chunks, block.ChunkData{ + MinT: cm.MinT, + MaxT: cm.MaxT, + Data: data, + }) + } + } + } + + // Build flush data sorted by ref for deterministic output. + flushData := make([]block.SeriesFlush, 0, len(merged)) + for _, me := range merged { + flushData = append(flushData, block.SeriesFlush{ + Ref: me.ref, + Labels: me.labels, + Chunks: me.chunks, + }) + } + sort.Slice(flushData, func(i, j int) bool { return flushData[i].Ref < flushData[j].Ref }) + + // Determine new compaction level and collect source ULIDs. + maxLevel := 0 + sourceULIDs := make([]string, 0, len(sources)) + for _, src := range sources { + if src.Meta.Compaction.Level > maxLevel { + maxLevel = src.Meta.Compaction.Level + } + sourceULIDs = append(sourceULIDs, src.Meta.ULID) + } + + return block.FlushCompacted(c.dataDir, flushData, maxLevel+1, sourceULIDs) +} + +// Expired returns blocks whose MaxTime is older than the retention window. +// Returns nil if retention is disabled (zero). +func (c *Compactor) Expired(blocks []*block.Reader) []*block.Reader { + if c.retention == 0 { + return nil + } + cutoff := c.clock() - c.retention + var expired []*block.Reader + for _, b := range blocks { + if b.Meta.MaxTime < cutoff { + expired = append(expired, b) + } + } + return expired +} + +type mergedEntry struct { + ref uint64 + labels []labels.Label + chunks []block.ChunkData +} diff --git a/internal/compact/compact_test.go b/internal/compact/compact_test.go new file mode 100644 index 0000000..8e843d1 --- /dev/null +++ b/internal/compact/compact_test.go @@ -0,0 +1,430 @@ +package compact + +import ( + "math" + "path/filepath" + "testing" + + "git.dvdt.dev/david/ingot/internal/block" + "git.dvdt.dev/david/ingot/internal/chunkenc" + "git.dvdt.dev/david/ingot/labels" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const ( + hour = 3600 * 1000 // 1 hour in ms +) + +// makeChunk creates a chunk with the given samples and returns its raw bytes. +func makeChunk(t *testing.T, timestamps []int64, values []float64) []byte { + t.Helper() + c := chunkenc.NewXORChunk() + a, err := c.Appender() + require.NoError(t, err) + for i := range timestamps { + a.Append(timestamps[i], values[i]) + } + return append([]byte(nil), c.Bytes()...) +} + +// flushTestBlock creates a block in dataDir and returns an opened Reader. +func flushTestBlock(t *testing.T, dataDir string, series []block.SeriesFlush, level int, sources []string) *block.Reader { + t.Helper() + var ulid string + var err error + if level == 1 && sources == nil { + ulid, err = block.Flush(dataDir, series) + } else { + ulid, err = block.FlushCompacted(dataDir, series, level, sources) + } + require.NoError(t, err) + r, err := block.Open(filepath.Join(dataDir, ulid)) + require.NoError(t, err) + return r +} + +// collectBlockSamples reads all samples for a series ref from a block. +func collectBlockSamples(t *testing.T, r *block.Reader, ref uint64) []sample { + t.Helper() + it, err := r.SeriesChunkIterator(ref, math.MinInt64, math.MaxInt64) + require.NoError(t, err) + var out []sample + for it.Next() { + ts, v := it.At() + out = append(out, sample{ts, v}) + } + require.NoError(t, it.Err()) + return out +} + +type sample struct { + t int64 + v float64 +} + +func TestPlan(t *testing.T) { + tests := []struct { + name string + blocks []blockSpec // blocks to create + wantGroup bool // expect a compaction group + wantLevel int // expected resulting level + wantCount int // expected number of sources in group + }{ + { + name: "no_blocks", + blocks: nil, + wantGroup: false, + }, + { + name: "single_block", + blocks: []blockSpec{ + {minT: 0, maxT: 2 * hour, level: 1}, + }, + wantGroup: false, + }, + { + name: "two_level1_blocks_within_8h", + blocks: []blockSpec{ + {minT: 0, maxT: 2 * hour, level: 1}, + {minT: 2 * hour, maxT: 4 * hour, level: 1}, + }, + wantGroup: true, + wantLevel: 2, + wantCount: 2, + }, + { + name: "four_level1_blocks", + blocks: []blockSpec{ + {minT: 0, maxT: 2 * hour, level: 1}, + {minT: 2 * hour, maxT: 4 * hour, level: 1}, + {minT: 4 * hour, maxT: 6 * hour, level: 1}, + {minT: 6 * hour, maxT: 8 * hour, level: 1}, + }, + wantGroup: true, + wantLevel: 2, + wantCount: 4, + }, + { + name: "level1_blocks_exceed_8h_span", + blocks: []blockSpec{ + {minT: 0, maxT: 2 * hour, level: 1}, + {minT: 7 * hour, maxT: 9 * hour, level: 1}, + }, + wantGroup: false, + }, + { + name: "two_level2_blocks_within_32h", + blocks: []blockSpec{ + {minT: 0, maxT: 8 * hour, level: 2}, + {minT: 8 * hour, maxT: 16 * hour, level: 2}, + }, + wantGroup: true, + wantLevel: 3, + wantCount: 2, + }, + { + name: "max_level_blocks_not_compacted", + blocks: []blockSpec{ + {minT: 0, maxT: 32 * hour, level: 3}, + {minT: 32 * hour, maxT: 64 * hour, level: 3}, + }, + wantGroup: false, + }, + { + name: "mixed_levels_lower_prioritized", + blocks: []blockSpec{ + {minT: 0, maxT: 2 * hour, level: 1}, + {minT: 2 * hour, maxT: 4 * hour, level: 1}, + {minT: 10 * hour, maxT: 18 * hour, level: 2}, + {minT: 18 * hour, maxT: 26 * hour, level: 2}, + }, + wantGroup: true, + wantLevel: 2, // level-1 group found first + wantCount: 2, + }, + } + + levels := []int64{2 * hour, 8 * hour, 32 * hour} + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + dataDir := t.TempDir() + c := New(dataDir, levels, 0, nil) + + var blocks []*block.Reader + for _, spec := range tc.blocks { + r := createBlockWithMeta(t, dataDir, spec) + blocks = append(blocks, r) + t.Cleanup(func() { r.Close() }) + } + + group := c.Plan(blocks) + + if !tc.wantGroup { + assert.Nil(t, group, "expected no compaction group") + return + } + require.NotNil(t, group, "expected a compaction group") + assert.Equal(t, tc.wantLevel, group.Level, "compaction level") + assert.Equal(t, tc.wantCount, len(group.Sources), "source count") + }) + } +} + +func TestCompact(t *testing.T) { + tests := []struct { + name string + sourceBlocks []sourceBlock + wantSeriesRefs []uint64 + wantSamples map[uint64][]sample + wantLevel int + }{ + { + name: "merge_two_blocks_disjoint_series", + sourceBlocks: []sourceBlock{ + { + level: 1, + series: []seriesData{ + {ref: 1, labels: labels.FromStrings("__name__", "a"), samples: []sample{{1000, 1.0}, {2000, 2.0}}}, + }, + }, + { + level: 1, + series: []seriesData{ + {ref: 2, labels: labels.FromStrings("__name__", "b"), samples: []sample{{1000, 3.0}, {2000, 4.0}}}, + }, + }, + }, + wantSeriesRefs: []uint64{1, 2}, + wantSamples: map[uint64][]sample{ + 1: {{1000, 1.0}, {2000, 2.0}}, + 2: {{1000, 3.0}, {2000, 4.0}}, + }, + wantLevel: 2, + }, + { + name: "merge_two_blocks_same_series", + sourceBlocks: []sourceBlock{ + { + level: 1, + series: []seriesData{ + {ref: 1, labels: labels.FromStrings("__name__", "temp"), samples: []sample{{1000, 1.0}, {2000, 2.0}}}, + }, + }, + { + level: 1, + series: []seriesData{ + {ref: 1, labels: labels.FromStrings("__name__", "temp"), samples: []sample{{3000, 3.0}, {4000, 4.0}}}, + }, + }, + }, + wantSeriesRefs: []uint64{1}, + wantSamples: map[uint64][]sample{ + 1: {{1000, 1.0}, {2000, 2.0}, {3000, 3.0}, {4000, 4.0}}, + }, + wantLevel: 2, + }, + { + name: "merge_level2_blocks", + sourceBlocks: []sourceBlock{ + { + level: 2, + series: []seriesData{ + {ref: 1, labels: labels.FromStrings("__name__", "a"), samples: []sample{{1000, 1.0}}}, + }, + }, + { + level: 2, + series: []seriesData{ + {ref: 1, labels: labels.FromStrings("__name__", "a"), samples: []sample{{5000, 5.0}}}, + }, + }, + }, + wantSeriesRefs: []uint64{1}, + wantSamples: map[uint64][]sample{ + 1: {{1000, 1.0}, {5000, 5.0}}, + }, + wantLevel: 3, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + dataDir := t.TempDir() + c := New(dataDir, []int64{2 * hour, 8 * hour, 32 * hour}, 0, nil) + + // Create source blocks. + var sources []*block.Reader + for _, sb := range tc.sourceBlocks { + r := createSourceBlock(t, dataDir, sb) + sources = append(sources, r) + } + + // Compact. + newULID, err := c.Compact(sources) + require.NoError(t, err) + require.NotEmpty(t, newULID) + + // Close source blocks. + for _, s := range sources { + s.Close() + } + + // Open compacted block. + compacted, err := block.Open(filepath.Join(dataDir, newULID)) + require.NoError(t, err) + defer compacted.Close() + + // Verify compaction level. + assert.Equal(t, tc.wantLevel, compacted.Meta.Compaction.Level, "compaction level") + assert.Equal(t, len(tc.sourceBlocks), len(compacted.Meta.Compaction.Sources), "source count") + + // Verify series count. + assert.Equal(t, len(tc.wantSeriesRefs), compacted.Meta.Stats.NumSeries, "series count") + + // Verify each series' samples. + for _, ref := range tc.wantSeriesRefs { + got := collectBlockSamples(t, compacted, ref) + want := tc.wantSamples[ref] + require.Equal(t, len(want), len(got), "sample count for ref %d", ref) + for i := range want { + assert.Equal(t, want[i].t, got[i].t, "ref %d sample %d t", ref, i) + assert.Equal(t, want[i].v, got[i].v, "ref %d sample %d v", ref, i) + } + } + }) + } +} + +func TestExpired(t *testing.T) { + tests := []struct { + name string + blocks []blockSpec + now int64 + retention int64 + wantCount int + }{ + { + name: "no_retention", + blocks: []blockSpec{{minT: 0, maxT: 1000}}, + now: 100000, + retention: 0, + wantCount: 0, + }, + { + name: "all_within_retention", + blocks: []blockSpec{ + {minT: 80000, maxT: 90000}, + }, + now: 100000, + retention: 50000, + wantCount: 0, + }, + { + name: "one_expired", + blocks: []blockSpec{ + {minT: 0, maxT: 10000}, + {minT: 80000, maxT: 90000}, + }, + now: 100000, + retention: 50000, + wantCount: 1, + }, + { + name: "all_expired", + blocks: []blockSpec{ + {minT: 0, maxT: 10000}, + {minT: 20000, maxT: 30000}, + }, + now: 100000, + retention: 50000, + wantCount: 2, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + dataDir := t.TempDir() + clock := func() int64 { return tc.now } + c := New(dataDir, nil, tc.retention, clock) + + var blocks []*block.Reader + for _, spec := range tc.blocks { + r := createBlockWithMeta(t, dataDir, spec) + blocks = append(blocks, r) + t.Cleanup(func() { r.Close() }) + } + + expired := c.Expired(blocks) + assert.Equal(t, tc.wantCount, len(expired), "expired block count") + }) + } +} + +// --- helpers --- + +type blockSpec struct { + minT int64 + maxT int64 + level int +} + +type sourceBlock struct { + level int + series []seriesData +} + +type seriesData struct { + ref uint64 + labels []labels.Label + samples []sample +} + +// createBlockWithMeta creates a minimal block with the given time range and level. +func createBlockWithMeta(t *testing.T, dataDir string, spec blockSpec) *block.Reader { + t.Helper() + // Need at least one series with samples spanning [minT, maxT]. + timestamps := []int64{spec.minT, spec.maxT} + values := []float64{1.0, 2.0} + data := makeChunk(t, timestamps, values) + + series := []block.SeriesFlush{{ + Ref: 1, + Labels: labels.FromStrings("__name__", "test"), + Chunks: []block.ChunkData{{MinT: spec.minT, MaxT: spec.maxT, Data: data}}, + }} + + level := spec.level + if level == 0 { + level = 1 + } + return flushTestBlock(t, dataDir, series, level, []string{"src"}) +} + +// createSourceBlock creates a block with the given series data. +func createSourceBlock(t *testing.T, dataDir string, sb sourceBlock) *block.Reader { + t.Helper() + var flushData []block.SeriesFlush + for _, sd := range sb.series { + timestamps := make([]int64, len(sd.samples)) + values := make([]float64, len(sd.samples)) + for i, s := range sd.samples { + timestamps[i] = s.t + values[i] = s.v + } + data := makeChunk(t, timestamps, values) + minT, maxT := sd.samples[0].t, sd.samples[len(sd.samples)-1].t + flushData = append(flushData, block.SeriesFlush{ + Ref: sd.ref, + Labels: sd.labels, + Chunks: []block.ChunkData{{MinT: minT, MaxT: maxT, Data: data}}, + }) + } + + level := sb.level + if level == 0 { + level = 1 + } + return flushTestBlock(t, dataDir, flushData, level, []string{"src"}) +} diff --git a/internal/head/head_test.go b/internal/head/head_test.go index 539d543..f32ac60 100644 --- a/internal/head/head_test.go +++ b/internal/head/head_test.go @@ -439,29 +439,51 @@ func TestFlushOlderThan(t *testing.T) { name string numSamples int // samples per series (each at 15s intervals) flushMaxT int64 - wantBlockSeries int // number of series in the block - wantBlockExists bool + postFlushAppend int // additional samples to append after flush (0 = none) + wantULID bool // whether flush produces a block + wantBlockSeries int // number of series in the block (0 if no block) + wantBlockCount int // samples in the block (0 if no block) + wantHeadCount int // minimum samples remaining in head after flush }{ { name: "flush_sealed_chunks", - numSamples: 250, // 2 sealed chunks (120 each) + 10 active + numSamples: 250, flushMaxT: math.MaxInt64, + postFlushAppend: 0, + wantULID: true, wantBlockSeries: 1, - wantBlockExists: true, + wantBlockCount: 240, + wantHeadCount: 10, }, { name: "nothing_to_flush", - numSamples: 50, // only active chunk, no sealed + numSamples: 50, flushMaxT: math.MaxInt64, + postFlushAppend: 0, + wantULID: false, wantBlockSeries: 0, - wantBlockExists: false, + wantBlockCount: 0, + wantHeadCount: 50, }, { name: "partial_flush_by_time", numSamples: 250, - flushMaxT: 120 * 15000, // only flush first sealed chunk + flushMaxT: 120 * 15000, + postFlushAppend: 0, + wantULID: true, wantBlockSeries: 1, - wantBlockExists: true, + wantBlockCount: 120, + wantHeadCount: 10, + }, + { + name: "flush_then_continue_appending", + numSamples: 250, + flushMaxT: math.MaxInt64, + postFlushAppend: 10, + wantULID: true, + wantBlockSeries: 1, + wantBlockCount: 240, + wantHeadCount: 10, }, } @@ -469,7 +491,6 @@ func TestFlushOlderThan(t *testing.T) { t.Run(tc.name, func(t *testing.T) { h := openHead(t) - // Append samples. app := h.Appender() ref, err := app.Append(0, []labels.Label{{Name: "__name__", Value: "temp"}}, 0, 0) require.NoError(t, err) @@ -479,134 +500,98 @@ func TestFlushOlderThan(t *testing.T) { } require.NoError(t, app.Commit()) - // Flush. ulid, err := h.FlushOlderThan(tc.flushMaxT) require.NoError(t, err) - if !tc.wantBlockExists { - assert.Empty(t, ulid) - return + // Post-flush appends. + if tc.postFlushAppend > 0 { + app = h.Appender() + for i := tc.numSamples; i < tc.numSamples+tc.postFlushAppend; i++ { + _, err = app.Append(ref, nil, int64(i*15000), float64(i)) + require.NoError(t, err) + } + require.NoError(t, app.Commit()) } - assert.NotEmpty(t, ulid) + // Assert ULID presence. + assert.Equal(t, tc.wantULID, ulid != "", "block ULID presence") - // Verify block exists and is readable. - blockDir := filepath.Join(h.DataDir(), ulid) - br, err := block.Open(blockDir) - require.NoError(t, err) - defer br.Close() - - assert.Equal(t, tc.wantBlockSeries, br.Meta.Stats.NumSeries) - - // Verify block data is correct by iterating. - if tc.wantBlockSeries > 0 { + // Assert block contents. + blockCount := 0 + if ulid != "" { + blockDir := filepath.Join(h.DataDir(), ulid) + br, err := block.Open(blockDir) + require.NoError(t, err) + defer br.Close() + assert.Equal(t, tc.wantBlockSeries, br.Meta.Stats.NumSeries, "block series count") it, err := br.SeriesChunkIterator(ref, math.MinInt64, math.MaxInt64) require.NoError(t, err) - count := 0 for it.Next() { - count++ + blockCount++ } require.NoError(t, it.Err()) - assert.Greater(t, count, 0, "block should contain samples") } + assert.Equal(t, tc.wantBlockCount, blockCount, "block sample count") - // Head should still have its active chunk data. - allSamples := collectSamples(t, h, ref, math.MinInt64, math.MaxInt64) - assert.Greater(t, len(allSamples), 0, "head should still have active chunk") + // Assert head still has data. + headSamples := collectSamples(t, h, ref, math.MinInt64, math.MaxInt64) + assert.GreaterOrEqual(t, len(headSamples), tc.wantHeadCount, "head sample count") }) } } -func TestFlushThenContinueAppending(t *testing.T) { - h := openHead(t) - - // Append enough to seal two chunks (240 samples), plus a few more. - app := h.Appender() - ref, err := app.Append(0, []labels.Label{{Name: "__name__", Value: "temp"}}, 0, 0) - require.NoError(t, err) - for i := 1; i < 250; i++ { - _, err = app.Append(ref, nil, int64(i*15000), float64(i)) - require.NoError(t, err) - } - require.NoError(t, app.Commit()) - - // Flush sealed chunks. - ulid, err := h.FlushOlderThan(math.MaxInt64) - require.NoError(t, err) - require.NotEmpty(t, ulid) - - // Continue appending after flush. - app = h.Appender() - for i := 250; i < 260; i++ { - _, err = app.Append(ref, nil, int64(i*15000), float64(i)) - require.NoError(t, err) - } - require.NoError(t, app.Commit()) - - // Head should have the active chunk data (unflushed). - allSamples := collectSamples(t, h, ref, math.MinInt64, math.MaxInt64) - assert.Greater(t, len(allSamples), 0) - - // Block should have the flushed data. - blockDir := filepath.Join(h.DataDir(), ulid) - br, err := block.Open(blockDir) - require.NoError(t, err) - defer br.Close() - - it, err := br.SeriesChunkIterator(ref, math.MinInt64, math.MaxInt64) - require.NoError(t, err) - blockCount := 0 - for it.Next() { - blockCount++ - } - require.NoError(t, it.Err()) - assert.Equal(t, 240, blockCount, "block should contain 2 sealed chunks of 120 samples each") -} - func TestFlushWALTruncation(t *testing.T) { - dir := t.TempDir() - walDir := filepath.Join(dir, "wal") - - h, err := Open(walDir, wal.Options{}) - require.NoError(t, err) - - // Append enough to seal chunks. - app := h.Appender() - ref, err := app.Append(0, []labels.Label{{Name: "__name__", Value: "temp"}}, 0, 0) - require.NoError(t, err) - for i := 1; i < 250; i++ { - _, err = app.Append(ref, nil, int64(i*15000), float64(i)) - require.NoError(t, err) + tests := []struct { + name string + numSamples int + }{ + {name: "250_samples", numSamples: 250}, + {name: "500_samples", numSamples: 500}, } - require.NoError(t, app.Commit()) - // Count WAL segments before flush. - walEntries, err := os.ReadDir(walDir) - require.NoError(t, err) - segsBefore := len(walEntries) + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + dir := t.TempDir() + walDir := filepath.Join(dir, "wal") - // Flush. - _, err = h.FlushOlderThan(math.MaxInt64) - require.NoError(t, err) + h, err := Open(walDir, wal.Options{}) + require.NoError(t, err) - // WAL segments should have been truncated (or at least not grown). - walEntries, err = os.ReadDir(walDir) - require.NoError(t, err) - segsAfter := len(walEntries) - assert.LessOrEqual(t, segsAfter, segsBefore, "WAL should be truncated after flush") + app := h.Appender() + ref, err := app.Append(0, []labels.Label{{Name: "__name__", Value: "temp"}}, 0, 0) + require.NoError(t, err) + _ = ref + for i := 1; i < tc.numSamples; i++ { + _, err = app.Append(ref, nil, int64(i*15000), float64(i)) + require.NoError(t, err) + } + require.NoError(t, app.Commit()) - require.NoError(t, h.Close()) + walEntries, err := os.ReadDir(walDir) + require.NoError(t, err) + segsBefore := len(walEntries) - // Re-open: head should recover from WAL (only unflushed data). - h2, err := Open(walDir, wal.Options{}) - require.NoError(t, err) - defer h2.Close() + _, err = h.FlushOlderThan(math.MaxInt64) + require.NoError(t, err) - // The re-opened head should be functional. - app = h2.Appender() - _, err = app.Append(0, []labels.Label{{Name: "__name__", Value: "new_series"}}, 5000000, 42.0) - require.NoError(t, err) - require.NoError(t, app.Commit()) + walEntries, err = os.ReadDir(walDir) + require.NoError(t, err) + segsAfter := len(walEntries) + assert.LessOrEqual(t, segsAfter, segsBefore, "WAL should be truncated after flush") + + require.NoError(t, h.Close()) + + // Re-open: head should be functional. + h2, err := Open(walDir, wal.Options{}) + require.NoError(t, err) + defer h2.Close() + + app = h2.Appender() + _, err = app.Append(0, []labels.Label{{Name: "__name__", Value: "new_series"}}, 5000000, 42.0) + require.NoError(t, err) + require.NoError(t, app.Commit()) + }) + } } func TestHeadPostings(t *testing.T) { @@ -642,44 +627,81 @@ func TestHeadPostings(t *testing.T) { } } -func TestHeadLabelValues(t *testing.T) { - h := openHead(t) +func TestHeadQueryMethods(t *testing.T) { + tests := []struct { + name string + series [][]labels.Label + wantLabelValues map[string][]string // label name -> expected values + wantLabels map[uint64][]labels.Label // ref -> expected labels + wantAllPostings []uint64 + }{ + { + name: "multiple_series_with_shared_labels", + series: [][]labels.Label{ + {{Name: "__name__", Value: "temp"}, {Name: "room", Value: "office"}}, + {{Name: "__name__", Value: "humidity"}, {Name: "room", Value: "office"}}, + {{Name: "__name__", Value: "temp"}, {Name: "room", Value: "kitchen"}}, + }, + wantLabelValues: map[string][]string{ + "__name__": {"humidity", "temp"}, + "room": {"kitchen", "office"}, + "nonexistent": {}, + }, + wantLabels: map[uint64][]labels.Label{ + 1: {{Name: "__name__", Value: "temp"}, {Name: "room", Value: "office"}}, + 2: {{Name: "__name__", Value: "humidity"}, {Name: "room", Value: "office"}}, + 3: {{Name: "__name__", Value: "temp"}, {Name: "room", Value: "kitchen"}}, + }, + wantAllPostings: []uint64{1, 2, 3}, + }, + { + name: "single_series", + series: [][]labels.Label{ + {{Name: "__name__", Value: "temp"}}, + }, + wantLabelValues: map[string][]string{ + "__name__": {"temp"}, + "nonexistent": {}, + }, + wantLabels: map[uint64][]labels.Label{ + 1: {{Name: "__name__", Value: "temp"}}, + }, + wantAllPostings: []uint64{1}, + }, + } - app := h.Appender() - _, err := app.Append(0, []labels.Label{{Name: "__name__", Value: "temp"}, {Name: "room", Value: "office"}}, 1000, 1.0) - require.NoError(t, err) - _, err = app.Append(0, []labels.Label{{Name: "__name__", Value: "humidity"}, {Name: "room", Value: "office"}}, 1000, 2.0) - require.NoError(t, err) - _, err = app.Append(0, []labels.Label{{Name: "__name__", Value: "temp"}, {Name: "room", Value: "kitchen"}}, 1000, 3.0) - require.NoError(t, err) - require.NoError(t, app.Commit()) + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + h := openHead(t) - assert.Equal(t, []string{"humidity", "temp"}, h.LabelValues("__name__")) - assert.Equal(t, []string{"kitchen", "office"}, h.LabelValues("room")) - assert.Equal(t, []string{}, h.LabelValues("nonexistent")) -} + app := h.Appender() + for _, ls := range tc.series { + _, err := app.Append(0, ls, 1000, 1.0) + require.NoError(t, err) + } + require.NoError(t, app.Commit()) -func TestHeadLabelsAndAllPostings(t *testing.T) { - h := openHead(t) + // Assert LabelValues. + for name, wantVals := range tc.wantLabelValues { + got := h.LabelValues(name) + assert.Equal(t, wantVals, got, "LabelValues(%q)", name) + } - app := h.Appender() - _, err := app.Append(0, []labels.Label{{Name: "__name__", Value: "temp"}}, 1000, 1.0) - require.NoError(t, err) - _, err = app.Append(0, []labels.Label{{Name: "__name__", Value: "humidity"}}, 1000, 2.0) - require.NoError(t, err) - require.NoError(t, app.Commit()) + // Assert Labels by ref. + for ref, wantLabels := range tc.wantLabels { + ls, ok := h.Labels(ref) + assert.True(t, ok, "Labels(%d) should exist", ref) + assert.Equal(t, wantLabels, ls, "Labels(%d)", ref) + } - // Labels - ls, ok := h.Labels(1) - assert.True(t, ok) - assert.Equal(t, []labels.Label{{Name: "__name__", Value: "temp"}}, ls) + // Unknown ref returns false. + _, ok := h.Labels(999) + assert.False(t, ok, "Labels(999) should not exist") - _, ok = h.Labels(999) - assert.False(t, ok) - - // AllPostings - refs := h.AllPostings() - assert.Equal(t, []uint64{1, 2}, refs) + // Assert AllPostings. + assert.Equal(t, tc.wantAllPostings, h.AllPostings(), "AllPostings") + }) + } } func TestConcurrentAppend(t *testing.T) { diff --git a/labels/matchers_test.go b/labels/matchers_test.go index 80c0d9a..b7d3109 100644 --- a/labels/matchers_test.go +++ b/labels/matchers_test.go @@ -51,15 +51,41 @@ func TestMatcherMatches(t *testing.T) { } } -func TestNewMatcherBadRegexp(t *testing.T) { - _, err := NewMatcher(MatchRegexp, "__name__", "[invalid") - assert.Error(t, err) +func TestNewMatcherErrors(t *testing.T) { + tests := []struct { + name string + typ MatchType + pattern string + }{ + {name: "bad_regexp_bracket", typ: MatchRegexp, pattern: "[invalid"}, + {name: "bad_not_regexp_bracket", typ: MatchNotRegexp, pattern: "(unclosed"}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + _, err := NewMatcher(tc.typ, "__name__", tc.pattern) + assert.Error(t, err) + }) + } } func TestMustNewMatcherPanics(t *testing.T) { - assert.Panics(t, func() { - MustNewMatcher(MatchRegexp, "__name__", "[invalid") - }) + tests := []struct { + name string + typ MatchType + pattern string + }{ + {name: "bad_regexp", typ: MatchRegexp, pattern: "[invalid"}, + {name: "bad_not_regexp", typ: MatchNotRegexp, pattern: "(unclosed"}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + assert.Panics(t, func() { + MustNewMatcher(tc.typ, "__name__", tc.pattern) + }) + }) + } } func TestFromStrings(t *testing.T) { @@ -93,15 +119,38 @@ func TestFromStrings(t *testing.T) { } } -func TestFromStringsPanicsOnOdd(t *testing.T) { - assert.Panics(t, func() { - FromStrings("__name__") - }) +func TestFromStringsPanics(t *testing.T) { + tests := []struct { + name string + args []string + }{ + {name: "odd_count_one", args: []string{"__name__"}}, + {name: "odd_count_three", args: []string{"__name__", "temp", "room"}}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + assert.Panics(t, func() { + FromStrings(tc.args...) + }) + }) + } } func TestMatchTypeString(t *testing.T) { - assert.Equal(t, "=", MatchEqual.String()) - assert.Equal(t, "!=", MatchNotEqual.String()) - assert.Equal(t, "=~", MatchRegexp.String()) - assert.Equal(t, "!~", MatchNotRegexp.String()) + tests := []struct { + typ MatchType + want string + }{ + {typ: MatchEqual, want: "="}, + {typ: MatchNotEqual, want: "!="}, + {typ: MatchRegexp, want: "=~"}, + {typ: MatchNotRegexp, want: "!~"}, + } + + for _, tc := range tests { + t.Run(tc.want, func(t *testing.T) { + assert.Equal(t, tc.want, tc.typ.String()) + }) + } } diff --git a/soak_test.go b/soak_test.go new file mode 100644 index 0000000..438343f --- /dev/null +++ b/soak_test.go @@ -0,0 +1,234 @@ +package ingot + +import ( + "fmt" + "math" + "os" + "path/filepath" + "runtime" + "sync/atomic" + "testing" + "time" + + "git.dvdt.dev/david/ingot/labels" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestSoak simulates 48h of ingestion at 15s intervals across 10k series, +// validating that compaction, retention, and concurrent queries all work +// correctly under sustained load. +// +// Skipped with -short. Expected runtime: 1-5 minutes. +func TestSoak(t *testing.T) { + if testing.Short() { + t.Skip("skipping soak test in -short mode") + } + + const ( + numSeries = 10_000 + canaryCount = 10 // series tracked in the oracle for query validation + scrapeInterval = 15_000 // 15s in ms + simDuration = 48 * 3600 * 1000 // 48h in ms + blockDuration = 2 * 3600 * 1000 // 2h in ms + retentionMs = 24 * 3600 * 1000 // 24h in ms + flushInterval = 5 * 60 * 1000 // flush every 5 simulated minutes + compactInterval = 30 * 60 * 1000 // compact every 30 simulated minutes + queryInterval = 10 * 60 * 1000 // validate queries every 10 simulated minutes + ) + + var now atomic.Int64 + startTime := int64(1_000_000_000) // arbitrary start: ~11.5 days in ms + now.Store(startTime) + clock := func() int64 { return now.Load() } + + dir := t.TempDir() + db, err := Open(dir, Options{ + Retention: time.Duration(retentionMs) * time.Millisecond, + BlockDuration: time.Duration(blockDuration) * time.Millisecond, + Clock: clock, + }) + require.NoError(t, err) + defer db.Close() + + // Generate series labels. + seriesLabels := make([][]labels.Label, numSeries) + for i := 0; i < numSeries; i++ { + seriesLabels[i] = labels.FromStrings( + "__name__", fmt.Sprintf("metric_%d", i/100), + "instance", fmt.Sprintf("inst_%d", i%100), + ) + } + + // Oracle tracks canary series only. + oracle := newOracle() + refs := make([]uint64, numSeries) + + // Track stats. + var totalSamples int64 + var queryErrors int64 + var maxHeapMB uint64 + + endTime := startTime + simDuration + for ts := startTime; ts < endTime; ts += scrapeInterval { + now.Store(ts) + + // Append samples for all series. + app := db.Appender() + for i := 0; i < numSeries; i++ { + var ls []labels.Label + if refs[i] == 0 { + ls = seriesLabels[i] + } + r, err := app.Append(refs[i], ls, ts, float64(ts+int64(i))) + require.NoError(t, err) + refs[i] = r + if i < canaryCount { + if ls != nil { + oracle.addSeries(r, seriesLabels[i]) + } + oracle.addSample(r, ts, float64(ts+int64(i))) + } + } + require.NoError(t, app.Commit()) + totalSamples += int64(numSeries) + + // Periodic flush. + elapsed := ts - startTime + if elapsed > 0 && elapsed%flushInterval == 0 { + _, err := db.FlushOlderThan(ts - int64(blockDuration)) + require.NoError(t, err) + } + + // Periodic compaction + retention. + if elapsed > 0 && elapsed%compactInterval == 0 { + err := db.RunCompaction() + require.NoError(t, err) + db.ApplyRetention() + } + + // Periodic query validation. + if elapsed > 0 && elapsed%queryInterval == 0 { + if !validateCanaries(t, db, oracle, ts, retentionMs, clock) { + atomic.AddInt64(&queryErrors, 1) + } + + // Check heap allocation. + var ms runtime.MemStats + runtime.ReadMemStats(&ms) + heapMB := ms.HeapAlloc / (1024 * 1024) + if heapMB > maxHeapMB { + maxHeapMB = heapMB + } + } + } + + // Final validation. + t.Logf("Total samples written: %d", totalSamples) + t.Logf("Peak heap: %d MiB", maxHeapMB) + + // Assert zero query errors. + assert.Equal(t, int64(0), atomic.LoadInt64(&queryErrors), "query errors during soak") + + // Assert bounded memory (should stay well under 1 GiB with 10k series). + assert.Less(t, maxHeapMB, uint64(1024), "heap should stay under 1 GiB") + + // Assert bounded disk: count block directories. + entries, err := os.ReadDir(dir) + require.NoError(t, err) + blockCount := 0 + for _, e := range entries { + if e.IsDir() && e.Name() != "wal" { + if _, err := os.Stat(filepath.Join(dir, e.Name(), "meta.json")); err == nil { + blockCount++ + } + } + } + t.Logf("Block count at end: %d", blockCount) + // With 24h retention and 2h blocks, expect roughly 12 raw + some compacted. + // Should be well under 50. + assert.Less(t, blockCount, 50, "block count should be bounded by retention") + + // Final canary validation. + finalNow := now.Load() + validateCanaries(t, db, oracle, finalNow, retentionMs, clock) + + // Live query during compaction: start query, compact, finish query. + q, err := db.Querier(math.MinInt64, math.MaxInt64) + require.NoError(t, err) + ss := q.Select(labels.MustNewMatcher(labels.MatchRegexp, "__name__", "metric_0.*")) + // Trigger compaction while query is open. + db.RunCompaction() + // Drain the query — should not error. + for ss.Next() { + it := ss.At().Iterator() + for it.Next() { + } + assert.NoError(t, it.Err()) + } + assert.NoError(t, ss.Err()) + require.NoError(t, q.Close()) +} + +// validateCanaries queries each canary series and compares with the oracle. +// Returns true if all validations pass. Every assertion runs for every canary. +func validateCanaries(t *testing.T, db *DB, o *oracle, now int64, retentionMs int64, clock func() int64) bool { + t.Helper() + pass := true + + // Query window: last 1h or from retention cutoff, whichever is later. + maxt := now + mint := now - 3600*1000 + retentionCutoff := clock() - retentionMs + if mint < retentionCutoff { + mint = retentionCutoff + } + + for ref, ls := range o.labels { + // Build matchers from labels. + var matchers []*labels.Matcher + for _, l := range ls { + matchers = append(matchers, labels.MustNewMatcher(labels.MatchEqual, l.Name, l.Value)) + } + + // Query. + q, err := db.Querier(mint, maxt) + require.NoError(t, err, "querier for canary ref %d", ref) + + ss := q.Select(matchers...) + var gotSamples []sample + var iterErr error + var ssErr error + for ss.Next() { + it := ss.At().Iterator() + for it.Next() { + st, sv := it.At() + gotSamples = append(gotSamples, sample{st, sv}) + } + iterErr = it.Err() + } + ssErr = ss.Err() + q.Close() + + // All assertions run unconditionally for every canary. + if !assert.NoError(t, iterErr, "canary ref %d: iterator error", ref) { + pass = false + } + if !assert.NoError(t, ssErr, "canary ref %d: series set error", ref) { + pass = false + } + + // Compare with oracle. + wantByRef := o.query(mint, maxt, matchers...) + var wantSamples []sample + if s, ok := wantByRef[ref]; ok { + wantSamples = s + } + + if !assert.Equal(t, len(wantSamples), len(gotSamples), + "canary ref %d sample count (mint=%d maxt=%d)", ref, mint, maxt) { + pass = false + } + } + return pass +}