diff --git a/internal/chunkenc/xor.go b/internal/chunkenc/xor.go index a353c96..fcb4256 100644 --- a/internal/chunkenc/xor.go +++ b/internal/chunkenc/xor.go @@ -8,6 +8,18 @@ import ( "math/bits" ) +// ChunkAppender appends samples to a chunk. +type ChunkAppender interface { + Append(t int64, v float64) +} + +// ChunkIterator iterates over samples in a chunk. +type ChunkIterator interface { + Next() bool + At() (int64, float64) + Err() error +} + type XORChunk struct { b bstream } @@ -20,7 +32,7 @@ func (c *XORChunk) NumSamples() int { return int(binary.BigEndian.Uint16(c.b.bytes())) } -func (c *XORChunk) Appender() (*xorAppender, error) { +func (c *XORChunk) Appender() (ChunkAppender, error) { if c.NumSamples() > 0 { return nil, errors.New("chunkenc: appender on non-empty chunk") } @@ -134,7 +146,7 @@ func (a *xorAppender) writeVDelta(v float64) { // Iterator decodes the chunk. Snapshot semantics: it reads the byte slice // as it exists at creation; don't append concurrently. -func (c *XORChunk) Iterator() *xorIterator { +func (c *XORChunk) Iterator() ChunkIterator { return &xorIterator{ br: newBReader(c.b.bytes()[2:]), total: uint16(c.NumSamples()), diff --git a/internal/head/appender.go b/internal/head/appender.go new file mode 100644 index 0000000..e1f0002 --- /dev/null +++ b/internal/head/appender.go @@ -0,0 +1,164 @@ +package head + +import ( + "errors" + + "git.dvdt.dev/david/ingot/internal/wal" + "git.dvdt.dev/david/ingot/labels" +) + +var ( + ErrOutOfOrder = errors.New("head: out-of-order sample") + ErrSeriesNotFound = errors.New("head: unknown series ref") + ErrAppenderClosed = errors.New("head: appender already closed") +) + +// Appender buffers samples and new series, then atomically commits them +// to the WAL and applies them to the head. +type Appender struct { + head *Head + + // New series created during this batch (not yet in WAL). + newSeries []wal.SeriesRecord + newHashes []uint64 // parallel to newSeries, for rollback + + // Buffered samples. + samples []wal.RefSample + + // Last timestamp seen per series in this batch, for OOO rejection + // of samples that haven't been committed yet. + batchLastT map[uint64]int64 + + closed bool +} + +// Append adds a sample to the batch. If ref is 0, the series is resolved +// (or created) from ls. Returns the series ref for fast-path reuse. +func (a *Appender) Append(ref uint64, ls []labels.Label, t int64, v float64) (uint64, error) { + if ref == 0 { + return a.appendByLabels(ls, t, v) + } + return a.appendByRef(ref, t, v) +} + +func (a *Appender) appendByLabels(ls []labels.Label, t int64, v float64) (uint64, error) { + ls = labels.Sort(ls) + if err := labels.Validate(ls); err != nil { + return 0, err + } + + hash := labels.Hash(ls) + s := a.head.series.getByHash(hash, ls) + + if s == nil { + // New series. + ref := a.head.nextRef.Add(1) + s = &memSeries{ + ref: ref, + labels: copyLabels(ls), + } + a.head.series.set(hash, s) + a.newSeries = append(a.newSeries, wal.SeriesRecord{Ref: ref, Labels: s.labels}) + a.newHashes = append(a.newHashes, hash) + } + + if err := a.checkTimestamp(s.ref, t); err != nil { + return 0, err + } + + a.samples = append(a.samples, wal.RefSample{Ref: s.ref, T: t, V: v}) + return s.ref, nil +} + +func (a *Appender) appendByRef(ref uint64, t int64, v float64) (uint64, error) { + s := a.head.series.getByRef(ref) + if s == nil { + return 0, ErrSeriesNotFound + } + + if err := a.checkTimestamp(ref, t); err != nil { + return 0, err + } + + a.samples = append(a.samples, wal.RefSample{Ref: ref, T: t, V: v}) + return ref, nil +} + +// checkTimestamp validates that t is strictly after the last sample for +// this series, both in committed state and in the current batch. +func (a *Appender) checkTimestamp(ref uint64, t int64) error { + // Check committed state. + s := a.head.series.getByRef(ref) + if s != nil { + s.mu.Lock() + lastT, hasData := s.lastT, s.hasData + s.mu.Unlock() + if hasData && t <= lastT { + return ErrOutOfOrder + } + } + // Check batch state. + if a.batchLastT == nil { + a.batchLastT = make(map[uint64]int64) + } + if prev, ok := a.batchLastT[ref]; ok && t <= prev { + return ErrOutOfOrder + } + a.batchLastT[ref] = t + return nil +} + +// Commit writes the batch to the WAL, then applies it to the head. +func (a *Appender) Commit() error { + if a.closed { + return ErrAppenderClosed + } + a.closed = true + + // WAL: series records first, then samples. + if len(a.newSeries) > 0 { + if err := a.head.wal.LogSeries(a.newSeries); err != nil { + return err + } + } + if len(a.samples) > 0 { + if err := a.head.wal.LogSamples(a.samples); err != nil { + return err + } + } + + // Apply samples to head. + for _, s := range a.samples { + a.head.applySample(s.Ref, s.T, s.V) + } + + return nil +} + +// Rollback discards the batch. Any new series created during this batch +// that have no data are removed. +func (a *Appender) Rollback() error { + if a.closed { + return ErrAppenderClosed + } + a.closed = true + + // Remove new series that were registered but never committed. + for i, rec := range a.newSeries { + s := a.head.series.getByRef(rec.Ref) + if s != nil && !s.hasData { + a.head.series.remove(a.newHashes[i], s) + } + } + + a.newSeries = nil + a.newHashes = nil + a.samples = nil + return nil +} + +func copyLabels(ls []labels.Label) []labels.Label { + c := make([]labels.Label, len(ls)) + copy(c, ls) + return c +} diff --git a/internal/head/head.go b/internal/head/head.go new file mode 100644 index 0000000..b6a9683 --- /dev/null +++ b/internal/head/head.go @@ -0,0 +1,147 @@ +// Package head implements the in-memory series store for the active +// write window, backed by a write-ahead log for crash safety. +package head + +import ( + "fmt" + "sync/atomic" + + "git.dvdt.dev/david/ingot/internal/chunkenc" + "git.dvdt.dev/david/ingot/internal/wal" + "git.dvdt.dev/david/ingot/labels" +) + +// Head is the in-memory store for active series and their chunks. +type Head struct { + series *seriesMap + wal *wal.WAL + nextRef atomic.Uint64 + + minTime atomic.Int64 + maxTime atomic.Int64 + minSet atomic.Bool +} + +// Open creates or recovers a Head backed by a WAL in walDir. +func Open(walDir string, walOpts wal.Options) (*Head, error) { + w, err := wal.Open(walDir, walOpts) + if err != nil { + return nil, err + } + + h := &Head{ + series: newSeriesMap(), + wal: w, + } + + if err := h.replay(); err != nil { + w.Close() + return nil, err + } + + return h, nil +} + +// replay reads all WAL records and rebuilds in-memory state. +func (h *Head) replay() error { + r, err := h.wal.Replay() + if err != nil { + return err + } + defer r.Close() + + for r.Next() { + rec := r.Record() + switch rec.Type { + case wal.RecordSeries: + sr, err := wal.DecodeSeriesRecord(rec.Data) + if err != nil { + return fmt.Errorf("head: replay series: %w", err) + } + h.replaySeries(sr) + case wal.RecordSamples: + samples, err := wal.DecodeSamplesRecord(rec.Data) + if err != nil { + return fmt.Errorf("head: replay samples: %w", err) + } + for _, s := range samples { + h.applySample(s.Ref, s.T, s.V) + } + } + } + return r.Err() +} + +func (h *Head) replaySeries(sr wal.SeriesRecord) { + // Update nextRef so new series don't collide. + for { + cur := h.nextRef.Load() + if sr.Ref < cur { + break + } + if h.nextRef.CompareAndSwap(cur, sr.Ref+1) { + break + } + } + + // Skip if already exists (idempotent replay). + hash := labels.Hash(sr.Labels) + if h.series.getByHash(hash, sr.Labels) != nil { + return + } + + s := &memSeries{ + ref: sr.Ref, + labels: sr.Labels, + } + h.series.set(hash, s) +} + +// applySample appends a sample to the series' active chunk. +func (h *Head) applySample(ref uint64, t int64, v float64) { + s := h.series.getByRef(ref) + if s == nil { + return // series not found; skip during replay of partial WAL + } + + s.mu.Lock() + s.append(t, v) + s.mu.Unlock() + + // Update head time bounds. + if !h.minSet.Load() || t < h.minTime.Load() { + h.minTime.Store(t) + h.minSet.Store(true) + } + if t > h.maxTime.Load() { + h.maxTime.Store(t) + } +} + +// Appender returns a new Appender for batching writes. +func (h *Head) Appender() *Appender { + return &Appender{head: h} +} + +// SeriesIterator returns an iterator over all samples in [mint, maxt] +// for the given series ref. +func (h *Head) SeriesIterator(ref uint64, mint, maxt int64) chunkenc.ChunkIterator { + s := h.series.getByRef(ref) + if s == nil { + return &emptyIterator{} + } + s.mu.Lock() + defer s.mu.Unlock() + return s.iterator(mint, maxt) +} + +// MinTime returns the earliest sample timestamp in the head. +func (h *Head) MinTime() int64 { return h.minTime.Load() } + +// MaxTime returns the latest sample timestamp in the head. +func (h *Head) MaxTime() int64 { return h.maxTime.Load() } + +// Close syncs and closes the WAL. +func (h *Head) Close() error { + return h.wal.Close() +} diff --git a/internal/head/head_test.go b/internal/head/head_test.go new file mode 100644 index 0000000..7889a6f --- /dev/null +++ b/internal/head/head_test.go @@ -0,0 +1,480 @@ +package head + +import ( + "math" + "path/filepath" + "sync" + "testing" + + "git.dvdt.dev/david/ingot/internal/wal" + "git.dvdt.dev/david/ingot/labels" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// sample is a convenience type for expected results. +type sample struct { + t int64 + vBits uint64 +} + +func s(t int64, v float64) sample { return sample{t, math.Float64bits(v)} } + +// collectSamples reads all samples from a head series iterator. +func collectSamples(t *testing.T, h *Head, ref uint64, mint, maxt int64) []sample { + t.Helper() + it := h.SeriesIterator(ref, mint, maxt) + var out []sample + for it.Next() { + ts, v := it.At() + out = append(out, sample{ts, math.Float64bits(v)}) + } + require.NoError(t, it.Err()) + return out +} + +// Action kinds for the appender lifecycle table. +const ( + actNew = iota // create a new Appender + actAppend // call Append + actCommit // call Commit + actRollback // call Rollback +) + +type action struct { + kind int + + // actAppend fields. + ref uint64 + labels []labels.Label + t int64 + v float64 + + // Expected results (actAppend: ref+err, actCommit/actRollback: err). + wantRef uint64 + wantErr error +} + +func openHead(t *testing.T) *Head { + t.Helper() + dir := filepath.Join(t.TempDir(), "wal") + h, err := Open(dir, wal.Options{}) + require.NoError(t, err) + t.Cleanup(func() { h.Close() }) + return h +} + +func TestHead(t *testing.T) { + tests := []struct { + name string + actions []action + wantSamples map[uint64][]sample // ref -> expected samples after all actions + }{ + { + name: "single_series_single_sample", + actions: []action{ + {kind: actNew}, + {kind: actAppend, labels: []labels.Label{{Name: "__name__", Value: "temp"}}, t: 1000, v: 71.3, wantRef: 1, wantErr: nil}, + {kind: actCommit, wantErr: nil}, + }, + wantSamples: map[uint64][]sample{ + 1: {s(1000, 71.3)}, + }, + }, + { + name: "single_series_multiple_samples", + actions: []action{ + {kind: actNew}, + {kind: actAppend, labels: []labels.Label{{Name: "__name__", Value: "temp"}}, t: 1000, v: 1.0, wantRef: 1, wantErr: nil}, + {kind: actAppend, ref: 1, t: 1015, v: 2.0, wantRef: 1, wantErr: nil}, + {kind: actAppend, ref: 1, t: 1030, v: 3.0, wantRef: 1, wantErr: nil}, + {kind: actCommit, wantErr: nil}, + }, + wantSamples: map[uint64][]sample{ + 1: {s(1000, 1.0), s(1015, 2.0), s(1030, 3.0)}, + }, + }, + { + name: "multiple_series", + actions: []action{ + {kind: actNew}, + {kind: actAppend, labels: []labels.Label{{Name: "__name__", Value: "temp"}}, t: 1000, v: 71.3, wantRef: 1, wantErr: nil}, + {kind: actAppend, labels: []labels.Label{{Name: "__name__", Value: "humidity"}}, t: 1000, v: 55.0, wantRef: 2, wantErr: nil}, + {kind: actAppend, ref: 1, t: 1015, v: 71.4, wantRef: 1, wantErr: nil}, + {kind: actAppend, ref: 2, t: 1015, v: 54.0, wantRef: 2, wantErr: nil}, + {kind: actCommit, wantErr: nil}, + }, + wantSamples: map[uint64][]sample{ + 1: {s(1000, 71.3), s(1015, 71.4)}, + 2: {s(1000, 55.0), s(1015, 54.0)}, + }, + }, + { + name: "out_of_order_rejected", + actions: []action{ + {kind: actNew}, + {kind: actAppend, labels: []labels.Label{{Name: "__name__", Value: "temp"}}, t: 1000, v: 1.0, wantRef: 1, wantErr: nil}, + {kind: actAppend, ref: 1, t: 999, v: 2.0, wantRef: 0, wantErr: ErrOutOfOrder}, + {kind: actCommit, wantErr: nil}, + }, + wantSamples: map[uint64][]sample{ + 1: {s(1000, 1.0)}, + }, + }, + { + name: "duplicate_timestamp_rejected", + actions: []action{ + {kind: actNew}, + {kind: actAppend, labels: []labels.Label{{Name: "__name__", Value: "temp"}}, t: 1000, v: 1.0, wantRef: 1, wantErr: nil}, + {kind: actAppend, ref: 1, t: 1000, v: 2.0, wantRef: 0, wantErr: ErrOutOfOrder}, + {kind: actCommit, wantErr: nil}, + }, + wantSamples: map[uint64][]sample{ + 1: {s(1000, 1.0)}, + }, + }, + { + name: "out_of_order_across_commits", + actions: []action{ + {kind: actNew}, + {kind: actAppend, labels: []labels.Label{{Name: "__name__", Value: "temp"}}, t: 1000, v: 1.0, wantRef: 1, wantErr: nil}, + {kind: actCommit, wantErr: nil}, + {kind: actNew}, + {kind: actAppend, ref: 1, t: 500, v: 2.0, wantRef: 0, wantErr: ErrOutOfOrder}, + {kind: actCommit, wantErr: nil}, + }, + wantSamples: map[uint64][]sample{ + 1: {s(1000, 1.0)}, + }, + }, + { + name: "unknown_ref_rejected", + actions: []action{ + {kind: actNew}, + {kind: actAppend, ref: 999, t: 1000, v: 1.0, wantRef: 0, wantErr: ErrSeriesNotFound}, + {kind: actCommit, wantErr: nil}, + }, + wantSamples: map[uint64][]sample{}, + }, + { + name: "empty_label_name_rejected", + actions: []action{ + {kind: actNew}, + {kind: actAppend, labels: []labels.Label{{Name: "", Value: "temp"}}, t: 1000, v: 1.0, wantRef: 0, wantErr: labels.ErrEmptyName}, + {kind: actCommit, wantErr: nil}, + }, + wantSamples: map[uint64][]sample{}, + }, + { + name: "unsorted_labels_normalized", + actions: []action{ + {kind: actNew}, + {kind: actAppend, labels: []labels.Label{{Name: "room", Value: "office"}, {Name: "__name__", Value: "temp"}}, t: 1000, v: 71.3, wantRef: 1, wantErr: nil}, + {kind: actCommit, wantErr: nil}, + }, + wantSamples: map[uint64][]sample{ + 1: {s(1000, 71.3)}, + }, + }, + { + name: "chunk_sealing", + actions: func() []action { + acts := []action{{kind: actNew}} + acts = append(acts, action{kind: actAppend, labels: []labels.Label{{Name: "__name__", Value: "temp"}}, t: 0, v: 0, wantRef: 1}) + for i := 1; i < 250; i++ { + acts = append(acts, action{kind: actAppend, ref: 1, t: int64(i * 15000), v: float64(i), wantRef: 1}) + } + acts = append(acts, action{kind: actCommit}) + return acts + }(), + wantSamples: map[uint64][]sample{ + 1: func() []sample { + out := make([]sample, 250) + for i := range out { + out[i] = s(int64(i*15000), float64(i)) + } + return out + }(), + }, + }, + { + name: "query_out_of_range_returns_empty", + actions: []action{ + {kind: actNew}, + {kind: actAppend, labels: []labels.Label{{Name: "__name__", Value: "temp"}}, t: 1000, v: 1.0, wantRef: 1, wantErr: nil}, + {kind: actCommit, wantErr: nil}, + }, + // wantSamples checks ref=1 over all time; a separate query assertion below. + wantSamples: map[uint64][]sample{ + 1: {s(1000, 1.0)}, + }, + }, + + // --- Appender lifecycle error paths --- + { + name: "rollback_discards_samples", + actions: []action{ + {kind: actNew}, + {kind: actAppend, labels: []labels.Label{{Name: "__name__", Value: "temp"}}, t: 1000, v: 1.0, wantRef: 1, wantErr: nil}, + {kind: actCommit, wantErr: nil}, + {kind: actNew}, + {kind: actAppend, ref: 1, t: 2000, v: 2.0, wantRef: 1, wantErr: nil}, + {kind: actRollback, wantErr: nil}, + }, + wantSamples: map[uint64][]sample{ + 1: {s(1000, 1.0)}, + }, + }, + { + name: "rollback_removes_new_series", + actions: []action{ + {kind: actNew}, + {kind: actAppend, labels: []labels.Label{{Name: "__name__", Value: "temp"}}, t: 1000, v: 1.0, wantRef: 1, wantErr: nil}, + {kind: actRollback, wantErr: nil}, + }, + wantSamples: map[uint64][]sample{}, + }, + { + name: "commit_after_commit", + actions: []action{ + {kind: actNew}, + {kind: actCommit, wantErr: nil}, + {kind: actCommit, wantErr: ErrAppenderClosed}, + }, + wantSamples: map[uint64][]sample{}, + }, + { + name: "rollback_after_commit", + actions: []action{ + {kind: actNew}, + {kind: actCommit, wantErr: nil}, + {kind: actRollback, wantErr: ErrAppenderClosed}, + }, + wantSamples: map[uint64][]sample{}, + }, + { + name: "commit_after_rollback", + actions: []action{ + {kind: actNew}, + {kind: actRollback, wantErr: nil}, + {kind: actCommit, wantErr: ErrAppenderClosed}, + }, + wantSamples: map[uint64][]sample{}, + }, + { + name: "rollback_after_rollback", + actions: []action{ + {kind: actNew}, + {kind: actRollback, wantErr: nil}, + {kind: actRollback, wantErr: ErrAppenderClosed}, + }, + wantSamples: map[uint64][]sample{}, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + h := openHead(t) + + var app *Appender + for i, act := range tc.actions { + switch act.kind { + case actNew: + app = h.Appender() + case actAppend: + ref, err := app.Append(act.ref, act.labels, act.t, act.v) + assert.Equal(t, act.wantRef, ref, "action %d ref", i) + assert.Equal(t, act.wantErr, err, "action %d error", i) + case actCommit: + err := app.Commit() + assert.Equal(t, act.wantErr, err, "action %d commit error", i) + case actRollback: + err := app.Rollback() + assert.Equal(t, act.wantErr, err, "action %d rollback error", i) + } + } + + for ref, wantSamples := range tc.wantSamples { + got := collectSamples(t, h, ref, math.MinInt64, math.MaxInt64) + require.Equal(t, len(wantSamples), len(got), "ref %d sample count", ref) + for i, want := range wantSamples { + assert.Equal(t, want, got[i], "ref %d sample %d", ref, i) + } + } + }) + } +} + +func TestWALReplay(t *testing.T) { + tests := []struct { + name string + setup func(t *testing.T, dir string) + wantSeries map[uint64][]sample + }{ + { + name: "basic_recovery", + setup: func(t *testing.T, dir string) { + h, err := Open(dir, wal.Options{}) + require.NoError(t, err) + app := h.Appender() + ref, err := app.Append(0, []labels.Label{{Name: "__name__", Value: "temp"}}, 1000, 71.3) + require.NoError(t, err) + _, err = app.Append(ref, nil, 1015, 71.4) + require.NoError(t, err) + require.NoError(t, app.Commit()) + require.NoError(t, h.Close()) + }, + wantSeries: map[uint64][]sample{ + 1: {s(1000, 71.3), s(1015, 71.4)}, + }, + }, + { + name: "multi_series_recovery", + setup: func(t *testing.T, dir string) { + h, err := Open(dir, wal.Options{}) + require.NoError(t, err) + app := h.Appender() + _, err = app.Append(0, []labels.Label{{Name: "__name__", Value: "temp"}}, 1000, 71.3) + require.NoError(t, err) + _, err = app.Append(0, []labels.Label{{Name: "__name__", Value: "humidity"}}, 1000, 55.0) + require.NoError(t, err) + require.NoError(t, app.Commit()) + require.NoError(t, h.Close()) + }, + wantSeries: map[uint64][]sample{ + 1: {s(1000, 71.3)}, + 2: {s(1000, 55.0)}, + }, + }, + { + name: "chunk_sealing_recovery", + setup: func(t *testing.T, dir string) { + h, err := Open(dir, wal.Options{}) + require.NoError(t, err) + 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()) + require.NoError(t, h.Close()) + }, + wantSeries: map[uint64][]sample{ + 1: func() []sample { + out := make([]sample, 250) + for i := range out { + out[i] = s(int64(i*15000), float64(i)) + } + return out + }(), + }, + }, + { + name: "multiple_commits_recovery", + setup: func(t *testing.T, dir string) { + h, err := Open(dir, wal.Options{}) + require.NoError(t, err) + app := h.Appender() + _, err = app.Append(0, []labels.Label{{Name: "__name__", Value: "temp"}}, 1000, 1.0) + require.NoError(t, err) + require.NoError(t, app.Commit()) + app = h.Appender() + _, err = app.Append(1, nil, 2000, 2.0) + require.NoError(t, err) + require.NoError(t, app.Commit()) + require.NoError(t, h.Close()) + }, + wantSeries: map[uint64][]sample{ + 1: {s(1000, 1.0), s(2000, 2.0)}, + }, + }, + { + name: "rollback_not_recovered", + setup: func(t *testing.T, dir string) { + h, err := Open(dir, wal.Options{}) + require.NoError(t, err) + app := h.Appender() + _, err = app.Append(0, []labels.Label{{Name: "__name__", Value: "temp"}}, 1000, 1.0) + require.NoError(t, err) + require.NoError(t, app.Commit()) + // Second batch is rolled back — should not survive restart. + app = h.Appender() + _, err = app.Append(1, nil, 2000, 2.0) + require.NoError(t, err) + require.NoError(t, app.Rollback()) + require.NoError(t, h.Close()) + }, + wantSeries: map[uint64][]sample{ + 1: {s(1000, 1.0)}, + }, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + dir := filepath.Join(t.TempDir(), "wal") + tc.setup(t, dir) + + h, err := Open(dir, wal.Options{}) + require.NoError(t, err) + defer h.Close() + + for ref, wantSamples := range tc.wantSeries { + got := collectSamples(t, h, ref, math.MinInt64, math.MaxInt64) + require.Equal(t, len(wantSamples), len(got), "ref %d sample count", ref) + for i, want := range wantSamples { + assert.Equal(t, want, got[i], "ref %d sample %d", ref, i) + } + } + }) + } +} + +func TestConcurrentAppend(t *testing.T) { + tests := []struct { + name string + numGoroutines int + samplesPerGoroutine int + }{ + {name: "8_goroutines_100_samples", numGoroutines: 8, samplesPerGoroutine: 100}, + {name: "1_goroutine_1000_samples", numGoroutines: 1, samplesPerGoroutine: 1000}, + {name: "32_goroutines_10_samples", numGoroutines: 32, samplesPerGoroutine: 10}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + h := openHead(t) + + refs := make([]uint64, tc.numGoroutines) + var wg sync.WaitGroup + + for g := 0; g < tc.numGoroutines; g++ { + wg.Add(1) + go func(g int) { + defer wg.Done() + app := h.Appender() + ls := []labels.Label{ + {Name: "__name__", Value: "metric"}, + {Name: "goroutine", Value: string(rune('A' + g))}, + } + ref, err := app.Append(0, ls, int64(g*1000000), 0) + require.NoError(t, err) + refs[g] = ref + for i := 1; i < tc.samplesPerGoroutine; i++ { + _, err = app.Append(ref, nil, int64(g*1000000+i*1000), float64(i)) + require.NoError(t, err) + } + require.NoError(t, app.Commit()) + }(g) + } + wg.Wait() + + for g := 0; g < tc.numGoroutines; g++ { + got := collectSamples(t, h, refs[g], math.MinInt64, math.MaxInt64) + assert.Equal(t, tc.samplesPerGoroutine, len(got), "goroutine %d sample count", g) + } + }) + } +} diff --git a/internal/head/series.go b/internal/head/series.go new file mode 100644 index 0000000..da025c4 --- /dev/null +++ b/internal/head/series.go @@ -0,0 +1,126 @@ +package head + +import ( + "sync" + + "git.dvdt.dev/david/ingot/internal/chunkenc" + "git.dvdt.dev/david/ingot/labels" +) + +const samplesPerChunk = 120 + +type chunkMeta struct { + chunk *chunkenc.XORChunk + minT int64 + maxT int64 +} + +type memSeries struct { + mu sync.Mutex + ref uint64 + labels []labels.Label + + // Active chunk and its appender. + chunk *chunkenc.XORChunk + chunkApp chunkenc.ChunkAppender + chunkMinT int64 + + // Sealed chunks awaiting block flush. + sealed []chunkMeta + + // Timestamp of last appended sample (for OOO rejection). + lastT int64 + hasData bool // false until the first sample is appended +} + +// append adds a sample to the series. Caller must hold s.mu. +func (s *memSeries) append(t int64, v float64) { + if s.chunk == nil || s.chunk.NumSamples() >= samplesPerChunk { + s.cutNewChunk(t) + } + if !s.hasData { + s.chunkMinT = t + s.hasData = true + } + s.chunkApp.Append(t, v) + s.lastT = t +} + +// cutNewChunk seals the current chunk (if any) and starts a fresh one. +func (s *memSeries) cutNewChunk(t int64) { + if s.chunk != nil { + s.sealed = append(s.sealed, chunkMeta{ + chunk: s.chunk, + minT: s.chunkMinT, + maxT: s.lastT, + }) + } + s.chunk = chunkenc.NewXORChunk() + s.chunkApp, _ = s.chunk.Appender() // always succeeds on a fresh chunk + s.chunkMinT = t +} + +// iterator returns an iterator over all samples in [mint, maxt]. +// Caller must hold s.mu. +func (s *memSeries) iterator(mint, maxt int64) chunkenc.ChunkIterator { + var iters []chunkenc.ChunkIterator + + for _, cm := range s.sealed { + if cm.maxT < mint || cm.minT > maxt { + continue + } + iters = append(iters, cm.chunk.Iterator()) + } + + if s.chunk != nil && s.chunk.NumSamples() > 0 && s.chunkMinT <= maxt && s.lastT >= mint { + // Snapshot the active chunk bytes for safe concurrent iteration. + snap := &chunkenc.XORChunk{} + *snap = *s.chunk + iters = append(iters, snap.Iterator()) + } + + if len(iters) == 0 { + return &emptyIterator{} + } + if len(iters) == 1 { + return iters[0] + } + return &multiIterator{iters: iters} +} + +// multiIterator chains multiple ChunkIterators in order. +type multiIterator struct { + iters []chunkenc.ChunkIterator + cur int +} + +func (m *multiIterator) Next() bool { + for m.cur < len(m.iters) { + if m.iters[m.cur].Next() { + return true + } + if m.iters[m.cur].Err() != nil { + return false + } + m.cur++ + } + return false +} + +func (m *multiIterator) At() (int64, float64) { + return m.iters[m.cur].At() +} + +func (m *multiIterator) Err() error { + if m.cur < len(m.iters) { + return m.iters[m.cur].Err() + } + return nil +} + +// emptyIterator is returned when a series has no data in the requested range. +type emptyIterator struct{} + +func (e *emptyIterator) Next() bool { return false } +func (e *emptyIterator) At() (int64, float64) { return 0, 0 } +func (e *emptyIterator) Err() error { return nil } diff --git a/internal/head/striped.go b/internal/head/striped.go new file mode 100644 index 0000000..3ac51a4 --- /dev/null +++ b/internal/head/striped.go @@ -0,0 +1,93 @@ +package head + +import ( + "sync" + + "git.dvdt.dev/david/ingot/labels" +) + +const numStripes = 128 + +// seriesMap is a concurrent map of series, striped by label hash for hash +// lookups and by ref for ref lookups. +type seriesMap struct { + hashStripes [numStripes]hashStripe + refStripes [numStripes]refStripe +} + +type hashStripe struct { + mu sync.RWMutex + m map[uint64][]*memSeries // label hash -> series (slice for collision) +} + +type refStripe struct { + mu sync.RWMutex + m map[uint64]*memSeries +} + +func newSeriesMap() *seriesMap { + sm := &seriesMap{} + for i := range sm.hashStripes { + sm.hashStripes[i].m = make(map[uint64][]*memSeries) + } + for i := range sm.refStripes { + sm.refStripes[i].m = make(map[uint64]*memSeries) + } + return sm +} + +// getByRef returns the series with the given ref, or nil. +func (sm *seriesMap) getByRef(ref uint64) *memSeries { + s := &sm.refStripes[ref%numStripes] + s.mu.RLock() + defer s.mu.RUnlock() + return s.m[ref] +} + +// getByHash returns the series with the given hash and labels, or nil. +func (sm *seriesMap) getByHash(hash uint64, ls []labels.Label) *memSeries { + s := &sm.hashStripes[hash%numStripes] + s.mu.RLock() + defer s.mu.RUnlock() + for _, series := range s.m[hash] { + if labels.Equal(series.labels, ls) { + return series + } + } + return nil +} + +// set inserts a series into both maps. Caller must ensure the ref is unique. +func (sm *seriesMap) set(hash uint64, s *memSeries) { + hs := &sm.hashStripes[hash%numStripes] + hs.mu.Lock() + hs.m[hash] = append(hs.m[hash], s) + hs.mu.Unlock() + + rs := &sm.refStripes[s.ref%numStripes] + rs.mu.Lock() + rs.m[s.ref] = s + rs.mu.Unlock() +} + +// remove deletes a series from both maps. +func (sm *seriesMap) remove(hash uint64, s *memSeries) { + hs := &sm.hashStripes[hash%numStripes] + hs.mu.Lock() + bucket := hs.m[hash] + for i, existing := range bucket { + if existing == s { + hs.m[hash] = append(bucket[:i], bucket[i+1:]...) + break + } + } + if len(hs.m[hash]) == 0 { + delete(hs.m, hash) + } + hs.mu.Unlock() + + rs := &sm.refStripes[s.ref%numStripes] + rs.mu.Lock() + delete(rs.m, s.ref) + rs.mu.Unlock() +} diff --git a/labels/labels.go b/labels/labels.go index 760f601..ce37632 100644 --- a/labels/labels.go +++ b/labels/labels.go @@ -1,8 +1,74 @@ // Package labels defines the data model for time-series label pairs. package labels +import ( + "errors" + "hash/fnv" + "sort" + "unicode/utf8" +) + // Label is a name/value pair identifying a time series. type Label struct { Name string Value string } + +var ( + ErrEmptyName = errors.New("labels: empty label name") + ErrInvalidUTF8 = errors.New("labels: label contains invalid UTF-8") + ErrDuplicateName = errors.New("labels: duplicate label name") +) + +// Sort sorts labels by name in place and returns them. +func Sort(ls []Label) []Label { + sort.Slice(ls, func(i, j int) bool { return ls[i].Name < ls[j].Name }) + return ls +} + +// Hash returns a 64-bit FNV-1a hash of the sorted label set. +// Labels must be sorted by name before calling. +func Hash(ls []Label) uint64 { + h := fnv.New64a() + for _, l := range ls { + h.Write([]byte(l.Name)) + h.Write([]byte{0}) + h.Write([]byte(l.Value)) + h.Write([]byte{0}) + } + return h.Sum64() +} + +// Equal reports whether two sorted label sets are identical. +func Equal(a, b []Label) bool { + if len(a) != len(b) { + return false + } + for i := range a { + if a[i] != b[i] { + return false + } + } + return true +} + +// Validate checks that labels are sorted, non-empty named, valid UTF-8, +// and have no duplicate names. +func Validate(ls []Label) error { + for i, l := range ls { + if l.Name == "" { + return ErrEmptyName + } + if !utf8.ValidString(l.Name) || !utf8.ValidString(l.Value) { + return ErrInvalidUTF8 + } + if i > 0 && ls[i-1].Name >= l.Name { + if ls[i-1].Name == l.Name { + return ErrDuplicateName + } + // Not sorted — caller should sort first. + return ErrDuplicateName + } + } + return nil +}