From 42b03db2faac76ee5d82da23f2c1e00410ba0621 Mon Sep 17 00:00:00 2001 From: David Rinaldo Date: Sat, 4 Jul 2026 14:59:39 -0400 Subject: [PATCH] Immutable blocks with mmap reads Flush sealed head chunks to ULID-named block directories on disk. Each block contains CRC'd chunk segment files (mmap'd for reads), a binary index (symbol table, series, postings with TOC), and a meta.json written last as the immutability gate. WAL is truncated after block fsync, preserving the crash-safety ording invariant. --- internal/block/block_test.go | 332 +++++++++++++++++++++++++++++++++ internal/block/chunk_reader.go | 174 +++++++++++++++++ internal/block/chunk_writer.go | 133 +++++++++++++ internal/block/meta.go | 55 ++++++ internal/block/reader.go | 147 +++++++++++++++ internal/block/ulid.go | 119 ++++++++++++ internal/block/writer.go | 141 ++++++++++++++ internal/chunkenc/xor.go | 9 + internal/chunkenc/xor_test.go | 52 ++++++ internal/head/head.go | 77 +++++++- internal/head/head_test.go | 177 ++++++++++++++++++ internal/head/striped.go | 12 ++ internal/index/index.go | 54 ++++++ internal/index/index_test.go | 208 +++++++++++++++++++++ internal/index/reader.go | 215 +++++++++++++++++++++ internal/index/writer.go | 273 +++++++++++++++++++++++++++ 16 files changed, 2176 insertions(+), 2 deletions(-) create mode 100644 internal/block/block_test.go create mode 100644 internal/block/chunk_reader.go create mode 100644 internal/block/chunk_writer.go create mode 100644 internal/block/meta.go create mode 100644 internal/block/reader.go create mode 100644 internal/block/ulid.go create mode 100644 internal/block/writer.go create mode 100644 internal/index/index.go create mode 100644 internal/index/index_test.go create mode 100644 internal/index/reader.go create mode 100644 internal/index/writer.go diff --git a/internal/block/block_test.go b/internal/block/block_test.go new file mode 100644 index 0000000..ca3b28a --- /dev/null +++ b/internal/block/block_test.go @@ -0,0 +1,332 @@ +package block + +import ( + "math" + "math/rand" + "os" + "path/filepath" + "testing" + + "git.dvdt.dev/david/ingot/internal/chunkenc" + "git.dvdt.dev/david/ingot/labels" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type sample struct { + t int64 + vBits uint64 +} + +func s(t int64, v float64) sample { return sample{t, math.Float64bits(v)} } + +// makeChunk creates a chunk with the given samples and returns its raw bytes. +func makeChunk(t *testing.T, samples []sample) []byte { + t.Helper() + c := chunkenc.NewXORChunk() + a, err := c.Appender() + require.NoError(t, err) + for _, s := range samples { + a.Append(s.t, math.Float64frombits(s.vBits)) + } + return append([]byte(nil), c.Bytes()...) +} + +// collectIterator reads all samples from a ChunkIterator. +func collectIterator(t *testing.T, it chunkenc.ChunkIterator) []sample { + t.Helper() + 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 +} + +func TestBlockRoundTrip(t *testing.T) { + tests := []struct { + name string + series []struct { + ref uint64 + labels []labels.Label + samples []sample + } + }{ + { + name: "single_series_single_chunk", + series: []struct { + ref uint64 + labels []labels.Label + samples []sample + }{ + { + ref: 1, + labels: []labels.Label{{Name: "__name__", Value: "temp"}}, + samples: []sample{ + s(1000, 71.3), s(1015, 71.4), s(1030, 71.5), + }, + }, + }, + }, + { + name: "multiple_series", + series: []struct { + ref uint64 + labels []labels.Label + samples []sample + }{ + { + ref: 1, + labels: []labels.Label{{Name: "__name__", Value: "temp"}, {Name: "room", Value: "office"}}, + samples: []sample{ + s(1000, 71.3), s(1015, 71.4), s(1030, 71.5), + }, + }, + { + ref: 2, + labels: []labels.Label{{Name: "__name__", Value: "humidity"}, {Name: "room", Value: "office"}}, + samples: []sample{ + s(1000, 55.0), s(1015, 54.5), s(1030, 54.0), + }, + }, + }, + }, + { + name: "large_chunk", + series: []struct { + ref uint64 + labels []labels.Label + samples []sample + }{ + { + ref: 1, + labels: []labels.Label{{Name: "__name__", Value: "counter"}}, + samples: func() []sample { + rnd := rand.New(rand.NewSource(42)) + out := make([]sample, 120) + ts, v := int64(0), 0.0 + for i := range out { + out[i] = s(ts, v) + ts += 15000 + int64(rnd.Intn(100)) - 50 + v += rnd.Float64() + } + return out + }(), + }, + }, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + dataDir := t.TempDir() + + // Build SeriesFlush data. + var flushData []SeriesFlush + for _, s := range tc.series { + data := makeChunk(t, s.samples) + minT, maxT := s.samples[0].t, s.samples[len(s.samples)-1].t + flushData = append(flushData, SeriesFlush{ + Ref: s.ref, + Labels: s.labels, + Chunks: []ChunkData{{MinT: minT, MaxT: maxT, Data: data}}, + }) + } + + // Write block. + ulid, err := Flush(dataDir, flushData) + require.NoError(t, err) + require.NotEmpty(t, ulid) + + // Open block for reading. + blockDir := filepath.Join(dataDir, ulid) + r, err := Open(blockDir) + require.NoError(t, err) + defer r.Close() + + // Verify meta. + assert.Equal(t, ulid, r.Meta.ULID) + assert.Equal(t, 1, r.Meta.Version) + assert.Equal(t, len(tc.series), r.Meta.Stats.NumSeries) + assert.Equal(t, len(tc.series), r.Meta.Stats.NumChunks) + + // Verify each series' data via iteration. + for _, s := range tc.series { + it, err := r.SeriesChunkIterator(s.ref, math.MinInt64, math.MaxInt64) + require.NoError(t, err) + got := collectIterator(t, it) + require.Equal(t, len(s.samples), len(got), "ref %d sample count", s.ref) + for i, want := range s.samples { + assert.Equal(t, want.t, got[i].t, "ref %d sample %d t", s.ref, i) + assert.Equal(t, want.vBits, got[i].vBits, "ref %d sample %d v", s.ref, i) + } + } + + // Verify postings. + for _, s := range tc.series { + for _, l := range s.labels { + refs := r.Postings(l.Name, l.Value) + assert.Contains(t, refs, s.ref, "postings for %s=%s", l.Name, l.Value) + } + } + + // Verify labels lookup. + for _, s := range tc.series { + ls, ok := r.Labels(s.ref) + assert.True(t, ok) + assert.Equal(t, s.labels, ls) + } + }) + } +} + +func TestBlockMultipleChunksPerSeries(t *testing.T) { + dataDir := t.TempDir() + + 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...) + + flushData := []SeriesFlush{ + { + Ref: 1, + Labels: []labels.Label{{Name: "__name__", Value: "temp"}}, + Chunks: []ChunkData{ + {MinT: 1000, MaxT: 1030, Data: makeChunk(t, chunk1Samples)}, + {MinT: 2000, MaxT: 2030, Data: makeChunk(t, chunk2Samples)}, + }, + }, + } + + 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) + } + + // 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{ + { + 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)})}}, + }, + { + 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)})}}, + }, + } + + ulid, err := Flush(dataDir, flushData) + require.NoError(t, err) + + 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) +} + +func TestULIDRoundTrip(t *testing.T) { + for i := 0; i < 100; i++ { + u := newULID() + assert.Equal(t, 26, len(u), "ULID length") + + // Should parse without error. + _, err := parseULID(u) + assert.NoError(t, err, "parse ULID %q", u) + } +} + +func TestCorruptChunkCRC(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) + + series := r.Series() + require.Equal(t, 1, len(series)) + chunkRef := series[0].Chunks[0].Ref + r.Close() + + // Corrupt chunk data on disk. + chunkSeg := chunkRef.Segment() + chunkPath := filepath.Join(blockDir, chunksDirName, segmentName(int(chunkSeg))) + + // 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 + } + 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) +} diff --git a/internal/block/chunk_reader.go b/internal/block/chunk_reader.go new file mode 100644 index 0000000..ac48db6 --- /dev/null +++ b/internal/block/chunk_reader.go @@ -0,0 +1,174 @@ +package block + +import ( + "encoding/binary" + "errors" + "fmt" + "hash/crc32" + "os" + "path/filepath" + "sort" + "syscall" + + "git.dvdt.dev/david/ingot/internal/chunkenc" + "git.dvdt.dev/david/ingot/internal/index" +) + +var ( + ErrInvalidChunkMagic = errors.New("block: invalid chunk file magic") + ErrInvalidChunkVersion = errors.New("block: unsupported chunk file version") + ErrCorruptChunk = errors.New("block: corrupt chunk (CRC mismatch)") + ErrChunkNotFound = errors.New("block: chunk ref out of bounds") +) + +// chunkReader reads chunk data from mmap'd segment files. +type chunkReader struct { + segments map[int][]byte // segment index -> mmap'd data +} + +func newChunkReader(blockDir string) (*chunkReader, error) { + chunksDir := filepath.Join(blockDir, chunksDirName) + entries, err := os.ReadDir(chunksDir) + if err != nil { + return nil, err + } + + cr := &chunkReader{segments: make(map[int][]byte)} + + for _, e := range entries { + if e.IsDir() { + continue + } + idx := parseSegmentName(e.Name()) + if idx < 0 { + continue + } + + data, err := mmapFile(filepath.Join(chunksDir, e.Name())) + if err != nil { + cr.close() + return nil, fmt.Errorf("block: mmap segment %s: %w", e.Name(), err) + } + + // Validate header. + if len(data) < chunkHeaderLen { + cr.close() + return nil, ErrInvalidChunkMagic + } + magic := binary.BigEndian.Uint32(data[:4]) + if magic != chunkMagic { + cr.close() + return nil, ErrInvalidChunkMagic + } + if data[4] != chunkVersion { + cr.close() + return nil, ErrInvalidChunkVersion + } + + cr.segments[idx] = data + } + + return cr, nil +} + +// chunkData reads the raw chunk bytes at the given ref, validates CRC. +func (cr *chunkReader) chunkData(ref index.ChunkRef) ([]byte, error) { + seg := int(ref.Segment()) + off := int(ref.Offset()) + + data, ok := cr.segments[seg] + if !ok { + return nil, ErrChunkNotFound + } + + if off+chunkEntryHeaderLen > len(data) { + return nil, ErrChunkNotFound + } + + dataLen := int(binary.BigEndian.Uint32(data[off : off+4])) + encoding := data[off+4] + off += chunkEntryHeaderLen + + end := off + dataLen + chunkEntryCRCLen + if end > len(data) { + return nil, ErrChunkNotFound + } + + chunkBytes := data[off : off+dataLen] + off += dataLen + + // Validate CRC. + wantCRC := binary.BigEndian.Uint32(data[off : off+4]) + crc := crc32.New(castagnoliTable) + crc.Write([]byte{encoding}) + crc.Write(chunkBytes) + if crc.Sum32() != wantCRC { + return nil, ErrCorruptChunk + } + + return chunkBytes, nil +} + +// chunkIterator returns a ChunkIterator for the chunk at the given ref. +func (cr *chunkReader) chunkIterator(ref index.ChunkRef) (chunkenc.ChunkIterator, error) { + data, err := cr.chunkData(ref) + if err != nil { + return nil, err + } + return chunkenc.XORChunkFromBytes(data).Iterator(), nil +} + +func (cr *chunkReader) close() error { + for _, data := range cr.segments { + syscall.Munmap(data) + } + cr.segments = nil + return nil +} + +// segmentIndices returns sorted segment indices. +func (cr *chunkReader) segmentIndices() []int { + idxs := make([]int, 0, len(cr.segments)) + for idx := range cr.segments { + idxs = append(idxs, idx) + } + sort.Ints(idxs) + return idxs +} + +func mmapFile(path string) ([]byte, error) { + f, err := os.Open(path) + if err != nil { + return nil, err + } + defer f.Close() + + info, err := f.Stat() + if err != nil { + return nil, err + } + if info.Size() == 0 { + return nil, fmt.Errorf("block: empty file %s", path) + } + + data, err := syscall.Mmap(int(f.Fd()), 0, int(info.Size()), + syscall.PROT_READ, syscall.MAP_PRIVATE) + if err != nil { + return nil, err + } + return data, nil +} + +func parseSegmentName(name string) int { + if len(name) != 6 { + return -1 + } + n := 0 + for _, c := range name { + if c < '0' || c > '9' { + return -1 + } + n = n*10 + int(c-'0') + } + return n +} diff --git a/internal/block/chunk_writer.go b/internal/block/chunk_writer.go new file mode 100644 index 0000000..97301a5 --- /dev/null +++ b/internal/block/chunk_writer.go @@ -0,0 +1,133 @@ +package block + +import ( + "encoding/binary" + "hash/crc32" + "os" + "path/filepath" + + "git.dvdt.dev/david/ingot/internal/index" +) + +const ( + chunkMagic uint32 = 0x0FACE5C4 + chunkVersion byte = 1 + chunkHeaderLen = 5 // 4-byte magic + 1-byte version + chunkEntryHeaderLen = 5 // 4-byte length + 1-byte encoding + chunkEntryCRCLen = 4 + encodingXOR byte = 1 + chunkSegmentMaxSize = 512 * 1024 * 1024 // 512 MiB + chunksDirName = "chunks" +) + +var castagnoliTable = crc32.MakeTable(crc32.Castagnoli) + +// chunkWriter writes chunk data to segmented chunk files inside a block directory. +type chunkWriter struct { + dir string + segmentIdx int + segmentOff int + f *os.File +} + +func newChunkWriter(blockDir string) (*chunkWriter, error) { + chunksDir := filepath.Join(blockDir, chunksDirName) + if err := os.MkdirAll(chunksDir, 0755); err != nil { + return nil, err + } + cw := &chunkWriter{dir: chunksDir, segmentIdx: 1} + if err := cw.newSegment(); err != nil { + return nil, err + } + return cw, nil +} + +func (cw *chunkWriter) newSegment() error { + if cw.f != nil { + if err := cw.f.Sync(); err != nil { + return err + } + if err := cw.f.Close(); err != nil { + return err + } + } + + name := segmentName(cw.segmentIdx) + f, err := os.Create(filepath.Join(cw.dir, name)) + if err != nil { + return err + } + cw.f = f + cw.segmentOff = 0 + + // Write chunk file header. + var hdr [chunkHeaderLen]byte + binary.BigEndian.PutUint32(hdr[:4], chunkMagic) + hdr[4] = chunkVersion + n, err := f.Write(hdr[:]) + cw.segmentOff += n + return err +} + +// writeChunk writes a chunk entry and returns its ChunkRef. +// Format: dataLen(4) | encoding(1) | data(dataLen) | CRC32C(4) +func (cw *chunkWriter) writeChunk(data []byte) (index.ChunkRef, error) { + entrySize := chunkEntryHeaderLen + len(data) + chunkEntryCRCLen + + // Rotate if this entry would exceed segment max size. + if cw.segmentOff+entrySize > chunkSegmentMaxSize { + cw.segmentIdx++ + if err := cw.newSegment(); err != nil { + return 0, err + } + } + + ref := index.NewChunkRef(uint32(cw.segmentIdx), uint32(cw.segmentOff)) + + // Header: length + encoding. + var hdr [chunkEntryHeaderLen]byte + binary.BigEndian.PutUint32(hdr[:4], uint32(len(data))) + hdr[4] = encodingXOR + if _, err := cw.f.Write(hdr[:]); err != nil { + return 0, err + } + + // Chunk data. + if _, err := cw.f.Write(data); err != nil { + return 0, err + } + + // CRC over encoding + data. + crc := crc32.New(castagnoliTable) + crc.Write(hdr[4:5]) // encoding byte + crc.Write(data) + var crcBuf [4]byte + binary.BigEndian.PutUint32(crcBuf[:], crc.Sum32()) + if _, err := cw.f.Write(crcBuf[:]); err != nil { + return 0, err + } + + cw.segmentOff += entrySize + return ref, nil +} + +func (cw *chunkWriter) close() error { + if cw.f == nil { + return nil + } + if err := cw.f.Sync(); err != nil { + cw.f.Close() + return err + } + return cw.f.Close() +} + +func segmentName(idx int) string { + var buf [6]byte + s := idx + for i := 5; i >= 0; i-- { + buf[i] = '0' + byte(s%10) + s /= 10 + } + return string(buf[:]) +} diff --git a/internal/block/meta.go b/internal/block/meta.go new file mode 100644 index 0000000..fd22164 --- /dev/null +++ b/internal/block/meta.go @@ -0,0 +1,55 @@ +package block + +import ( + "encoding/json" + "os" + "path/filepath" +) + +// BlockMeta describes a block on disk. +type BlockMeta struct { + ULID string `json:"ulid"` + MinTime int64 `json:"minTime"` + MaxTime int64 `json:"maxTime"` + Stats BlockStats `json:"stats"` + Compaction CompactionInfo `json:"compaction"` + Version int `json:"version"` +} + +// BlockStats holds summary statistics for a block. +type BlockStats struct { + NumSamples int `json:"numSamples"` + NumSeries int `json:"numSeries"` + NumChunks int `json:"numChunks"` +} + +// CompactionInfo records the block's compaction lineage. +type CompactionInfo struct { + Level int `json:"level"` + Sources []string `json:"sources"` +} + +const metaFilename = "meta.json" + +// readMeta reads a block's meta.json from the given block directory. +func readMeta(dir string) (BlockMeta, error) { + data, err := os.ReadFile(filepath.Join(dir, metaFilename)) + if err != nil { + return BlockMeta{}, err + } + var m BlockMeta + if err := json.Unmarshal(data, &m); err != nil { + return BlockMeta{}, err + } + return m, nil +} + +// writeMeta writes a block's meta.json to the given block directory. +// This is the last file written when creating a block — the immutability gate. +func writeMeta(dir string, m BlockMeta) error { + data, err := json.MarshalIndent(m, "", " ") + if err != nil { + return err + } + return os.WriteFile(filepath.Join(dir, metaFilename), data, 0644) +} diff --git a/internal/block/reader.go b/internal/block/reader.go new file mode 100644 index 0000000..771a739 --- /dev/null +++ b/internal/block/reader.go @@ -0,0 +1,147 @@ +package block + +import ( + "os" + "path/filepath" + + "git.dvdt.dev/david/ingot/internal/chunkenc" + "git.dvdt.dev/david/ingot/internal/index" + "git.dvdt.dev/david/ingot/labels" +) + +// Reader provides read access to an immutable on-disk block. +type Reader struct { + dir string + Meta BlockMeta + idx *index.Reader + chunks *chunkReader +} + +// Open opens a block directory for reading. Chunk files are mmap'd. +func Open(dir string) (*Reader, error) { + meta, err := readMeta(dir) + if err != nil { + return nil, err + } + + // Read the index file into memory. + indexData, err := os.ReadFile(filepath.Join(dir, "index")) + if err != nil { + return nil, err + } + idx, err := index.NewReader(indexData) + if err != nil { + return nil, err + } + + cr, err := newChunkReader(dir) + if err != nil { + return nil, err + } + + return &Reader{ + dir: dir, + Meta: meta, + idx: idx, + chunks: cr, + }, nil +} + +// Series returns all series entries from the index. +func (r *Reader) Series() []index.SeriesEntry { + return r.idx.Series() +} + +// SeriesByRef looks up a series by ref. +func (r *Reader) SeriesByRef(ref uint64) (index.SeriesEntry, bool) { + return r.idx.SeriesByRef(ref) +} + +// Postings returns sorted series refs matching label name=value. +func (r *Reader) Postings(name, value string) []uint64 { + return r.idx.Postings(name, value) +} + +// ChunkIterator returns an iterator for a chunk at the given ref. +func (r *Reader) ChunkIterator(ref index.ChunkRef) (chunkenc.ChunkIterator, error) { + return r.chunks.chunkIterator(ref) +} + +// SeriesChunkIterator returns an iterator over all chunks for a series in a +// time range. Chunks are iterated in order. +func (r *Reader) SeriesChunkIterator(ref uint64, mint, maxt int64) (chunkenc.ChunkIterator, error) { + entry, ok := r.idx.SeriesByRef(ref) + if !ok { + return &emptyIterator{}, nil + } + + var iters []chunkenc.ChunkIterator + for _, cm := range entry.Chunks { + if cm.MaxT < mint || cm.MinT > maxt { + continue + } + it, err := r.chunks.chunkIterator(cm.Ref) + if err != nil { + return nil, err + } + iters = append(iters, it) + } + + if len(iters) == 0 { + return &emptyIterator{}, nil + } + if len(iters) == 1 { + return iters[0], nil + } + return &multiIterator{iters: iters}, nil +} + +// Labels returns the labels for a series by ref. +func (r *Reader) Labels(ref uint64) ([]labels.Label, bool) { + entry, ok := r.idx.SeriesByRef(ref) + if !ok { + return nil, false + } + return entry.Labels, true +} + +// Close releases all resources (munmaps chunk files). +func (r *Reader) Close() error { + return r.chunks.close() +} + +// 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 +} + +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/block/ulid.go b/internal/block/ulid.go new file mode 100644 index 0000000..78318f3 --- /dev/null +++ b/internal/block/ulid.go @@ -0,0 +1,119 @@ +package block + +import ( + "crypto/rand" + "encoding/binary" + "errors" + "strings" + "time" +) + +// ULID encoding alphabet (Crockford's Base32). +const ulidEncoding = "0123456789ABCDEFGHJKMNPQRSTVWXYZ" + +// newULID generates a ULID with the current time and random entropy. +// Zero external dependencies — the encoding is simple enough to inline. +func newULID() string { + var b [16]byte + + // Timestamp: upper 48 bits = milliseconds since epoch. + ms := uint64(time.Now().UnixMilli()) + b[0] = byte(ms >> 40) + b[1] = byte(ms >> 32) + b[2] = byte(ms >> 24) + b[3] = byte(ms >> 16) + b[4] = byte(ms >> 8) + b[5] = byte(ms) + + // Randomness: lower 80 bits. + rand.Read(b[6:]) + + return encodeULID(b) +} + +// parseULID decodes a ULID string back to its 16-byte binary form. +func parseULID(s string) ([16]byte, error) { + var b [16]byte + if len(s) != 26 { + return b, errInvalidULID + } + + // Decode 26 base32 chars -> 130 bits -> 16 bytes + 2 spare bits. + var val [26]byte + for i := 0; i < 26; i++ { + idx := strings.IndexByte(ulidEncoding, s[i]) + if idx < 0 { + // Try lowercase. + idx = strings.IndexByte(ulidEncoding, s[i]&^0x20) + } + if idx < 0 { + return b, errInvalidULID + } + val[i] = byte(idx) + } + + // Pack 5-bit groups into bytes. + b[0] = val[0]<<5 | val[1] + b[1] = val[2]<<3 | val[3]>>2 + b[2] = val[3]<<6 | val[4]<<1 | val[5]>>4 + b[3] = val[5]<<4 | val[6]>>1 + b[4] = val[6]<<7 | val[7]<<2 | val[8]>>3 + b[5] = val[8]<<5 | val[9] + b[6] = val[10]<<3 | val[11]>>2 + b[7] = val[11]<<6 | val[12]<<1 | val[13]>>4 + b[8] = val[13]<<4 | val[14]>>1 + b[9] = val[14]<<7 | val[15]<<2 | val[16]>>3 + b[10] = val[16]<<5 | val[17] + b[11] = val[18]<<3 | val[19]>>2 + b[12] = val[19]<<6 | val[20]<<1 | val[21]>>4 + b[13] = val[21]<<4 | val[22]>>1 + b[14] = val[22]<<7 | val[23]<<2 | val[24]>>3 + b[15] = val[24]<<5 | val[25] + + return b, nil +} + +// ulidTime extracts the millisecond timestamp from a ULID string. +func ulidTime(s string) (int64, error) { + b, err := parseULID(s) + if err != nil { + return 0, err + } + // Timestamp is in the upper 6 bytes. + ms := int64(binary.BigEndian.Uint64(append([]byte{0, 0}, b[:6]...))) + return ms, nil +} + +func encodeULID(b [16]byte) string { + // 16 bytes = 128 bits, encoded in 26 base32 chars (130 bits, 2 spare). + var dst [26]byte + dst[0] = ulidEncoding[(b[0]&0xE0)>>5] + dst[1] = ulidEncoding[b[0]&0x1F] + dst[2] = ulidEncoding[(b[1]&0xF8)>>3] + dst[3] = ulidEncoding[(b[1]&0x07)<<2|(b[2]&0xC0)>>6] + dst[4] = ulidEncoding[(b[2]&0x3E)>>1] + dst[5] = ulidEncoding[(b[2]&0x01)<<4|(b[3]&0xF0)>>4] + dst[6] = ulidEncoding[(b[3]&0x0F)<<1|(b[4]&0x80)>>7] + dst[7] = ulidEncoding[(b[4]&0x7C)>>2] + dst[8] = ulidEncoding[(b[4]&0x03)<<3|(b[5]&0xE0)>>5] + dst[9] = ulidEncoding[b[5]&0x1F] + dst[10] = ulidEncoding[(b[6]&0xF8)>>3] + dst[11] = ulidEncoding[(b[6]&0x07)<<2|(b[7]&0xC0)>>6] + dst[12] = ulidEncoding[(b[7]&0x3E)>>1] + dst[13] = ulidEncoding[(b[7]&0x01)<<4|(b[8]&0xF0)>>4] + dst[14] = ulidEncoding[(b[8]&0x0F)<<1|(b[9]&0x80)>>7] + dst[15] = ulidEncoding[(b[9]&0x7C)>>2] + dst[16] = ulidEncoding[(b[9]&0x03)<<3|(b[10]&0xE0)>>5] + dst[17] = ulidEncoding[b[10]&0x1F] + dst[18] = ulidEncoding[(b[11]&0xF8)>>3] + dst[19] = ulidEncoding[(b[11]&0x07)<<2|(b[12]&0xC0)>>6] + dst[20] = ulidEncoding[(b[12]&0x3E)>>1] + dst[21] = ulidEncoding[(b[12]&0x01)<<4|(b[13]&0xF0)>>4] + dst[22] = ulidEncoding[(b[13]&0x0F)<<1|(b[14]&0x80)>>7] + dst[23] = ulidEncoding[(b[14]&0x7C)>>2] + dst[24] = ulidEncoding[(b[14]&0x03)<<3|(b[15]&0xE0)>>5] + dst[25] = ulidEncoding[b[15]&0x1F] + return string(dst[:]) +} + +var errInvalidULID = errors.New("block: invalid ULID") diff --git a/internal/block/writer.go b/internal/block/writer.go new file mode 100644 index 0000000..c9765fa --- /dev/null +++ b/internal/block/writer.go @@ -0,0 +1,141 @@ +package block + +import ( + "os" + "path/filepath" + + "git.dvdt.dev/david/ingot/internal/index" + "git.dvdt.dev/david/ingot/labels" +) + +// ChunkData describes a single chunk to be flushed to a block. +type ChunkData struct { + MinT int64 + MaxT int64 + Data []byte // raw XOR chunk bytes (including 2-byte sample count header) +} + +// SeriesFlush describes a series with its sealed chunks for block writing. +type SeriesFlush struct { + Ref uint64 + Labels []labels.Label + Chunks []ChunkData +} + +// Flush writes a new immutable block from the given series data. +// It creates a ULID-named directory under dataDir containing: +// - chunks/ with segment files +// - index file +// - meta.json (written last as the immutability gate) +// +// Returns the block ULID and any error. +func Flush(dataDir string, series []SeriesFlush) (string, error) { + ulid := newULID() + blockDir := filepath.Join(dataDir, ulid) + + if err := os.MkdirAll(blockDir, 0755); err != nil { + return "", err + } + + // Write chunk files and collect index entries. + cw, err := newChunkWriter(blockDir) + if err != nil { + return "", err + } + + var ( + indexEntries []index.SeriesEntry + meta BlockMeta + ) + meta.ULID = ulid + meta.Version = 1 + meta.Compaction = CompactionInfo{Level: 1, Sources: []string{ulid}} + meta.MinTime = int64(^uint64(0) >> 1) // max int64 + meta.MaxTime = int64(0) + + for _, sf := range series { + var chunks []index.ChunkMeta + for _, cd := range sf.Chunks { + ref, err := cw.writeChunk(cd.Data) + if err != nil { + cw.close() + return "", err + } + chunks = append(chunks, index.ChunkMeta{ + MinT: cd.MinT, + MaxT: cd.MaxT, + Ref: ref, + }) + if cd.MinT < meta.MinTime { + meta.MinTime = cd.MinT + } + if cd.MaxT > meta.MaxTime { + meta.MaxTime = cd.MaxT + } + meta.Stats.NumChunks++ + // Count samples from the chunk's 2-byte header. + if len(cd.Data) >= 2 { + meta.Stats.NumSamples += int(uint16(cd.Data[0])<<8 | uint16(cd.Data[1])) + } + } + indexEntries = append(indexEntries, index.SeriesEntry{ + Ref: sf.Ref, + Labels: sf.Labels, + Chunks: chunks, + }) + meta.Stats.NumSeries++ + } + + if err := cw.close(); err != nil { + return "", err + } + + // Write index file. + indexPath := filepath.Join(blockDir, "index") + indexFile, err := os.Create(indexPath) + if err != nil { + return "", err + } + + iw := index.NewWriter(indexFile) + for _, e := range indexEntries { + iw.AddSeries(e) + } + if _, err := iw.WriteTo(); err != nil { + indexFile.Close() + return "", err + } + if err := indexFile.Sync(); err != nil { + indexFile.Close() + return "", err + } + if err := indexFile.Close(); err != nil { + return "", err + } + + // Fsync the block directory to ensure all files are durable. + if err := syncDir(blockDir); err != nil { + return "", err + } + + // Write meta.json last — the immutability gate. + if err := writeMeta(blockDir, meta); err != nil { + return "", err + } + + // Fsync the data directory so the block directory entry is visible. + if err := syncDir(dataDir); err != nil { + return "", err + } + + return ulid, nil +} + +func syncDir(dir string) error { + d, err := os.Open(dir) + if err != nil { + return err + } + defer d.Close() + return d.Sync() +} diff --git a/internal/chunkenc/xor.go b/internal/chunkenc/xor.go index fcb4256..df49220 100644 --- a/internal/chunkenc/xor.go +++ b/internal/chunkenc/xor.go @@ -28,6 +28,15 @@ func NewXORChunk() *XORChunk { return &XORChunk{b: bstream{stream: make([]byte, 2), count: 0}} } +// XORChunkFromBytes creates a read-only XORChunk from raw bytes. +// The data must include the 2-byte sample count header (as returned by Bytes). +// The returned chunk supports Iterator, NumSamples, and Bytes but not Appender. +func XORChunkFromBytes(data []byte) *XORChunk { + cp := make([]byte, len(data)) + copy(cp, data) + return &XORChunk{b: bstream{stream: cp}} +} + func (c *XORChunk) NumSamples() int { return int(binary.BigEndian.Uint16(c.b.bytes())) } diff --git a/internal/chunkenc/xor_test.go b/internal/chunkenc/xor_test.go index fb7316c..0814660 100644 --- a/internal/chunkenc/xor_test.go +++ b/internal/chunkenc/xor_test.go @@ -298,6 +298,58 @@ func TestXORChunk(t *testing.T) { } } +func TestXORChunkFromBytes(t *testing.T) { + rnd := rand.New(rand.NewSource(99)) + + tests := []struct { + name string + samples [][2]float64 + }{ + {"single", [][2]float64{{1000, 71.3}}}, + {"two", [][2]float64{{1000, 71.3}, {1015, 71.4}}}, + {"full_chunk", func() [][2]float64 { + out := make([][2]float64, 120) + ts, v := int64(0), 70.0 + for i := range out { + out[i] = [2]float64{float64(ts), v} + ts += 15000 + int64(rnd.Intn(100)) - 50 + v += rnd.Float64() - 0.5 + } + return out + }()}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + orig := NewXORChunk() + a, _ := orig.Appender() + for _, s := range tc.samples { + a.Append(int64(s[0]), s[1]) + } + + reconstituted := XORChunkFromBytes(orig.Bytes()) + + assert.Equal(t, orig.NumSamples(), reconstituted.NumSamples()) + assert.Equal(t, orig.Bytes(), reconstituted.Bytes()) + + // Verify iteration produces identical samples. + it := reconstituted.Iterator() + for i, s := range tc.samples { + assert.True(t, it.Next(), "sample %d", i) + gotT, gotV := it.At() + assert.Equal(t, int64(s[0]), gotT, "sample %d t", i) + assert.Equal(t, math.Float64bits(s[1]), math.Float64bits(gotV), "sample %d v", i) + } + assert.False(t, it.Next()) + assert.NoError(t, it.Err()) + + // Appender on non-empty reconstituted chunk should fail. + _, err := reconstituted.Appender() + assert.Error(t, err) + }) + } +} + func FuzzXORIterator(f *testing.F) { c := NewXORChunk() a, _ := c.Appender() diff --git a/internal/head/head.go b/internal/head/head.go index b6a9683..0a09097 100644 --- a/internal/head/head.go +++ b/internal/head/head.go @@ -4,8 +4,10 @@ package head import ( "fmt" + "path/filepath" "sync/atomic" + "git.dvdt.dev/david/ingot/internal/block" "git.dvdt.dev/david/ingot/internal/chunkenc" "git.dvdt.dev/david/ingot/internal/wal" "git.dvdt.dev/david/ingot/labels" @@ -13,6 +15,7 @@ import ( // Head is the in-memory store for active series and their chunks. type Head struct { + dataDir string // parent directory containing WAL and block dirs series *seriesMap wal *wal.WAL nextRef atomic.Uint64 @@ -23,6 +26,7 @@ type Head struct { } // Open creates or recovers a Head backed by a WAL in walDir. +// The dataDir (parent of walDir) is used for writing blocks. func Open(walDir string, walOpts wal.Options) (*Head, error) { w, err := wal.Open(walDir, walOpts) if err != nil { @@ -30,8 +34,9 @@ func Open(walDir string, walOpts wal.Options) (*Head, error) { } h := &Head{ - series: newSeriesMap(), - wal: w, + dataDir: filepath.Dir(walDir), + series: newSeriesMap(), + wal: w, } if err := h.replay(); err != nil { @@ -145,3 +150,71 @@ func (h *Head) MaxTime() int64 { return h.maxTime.Load() } func (h *Head) Close() error { return h.wal.Close() } + +// FlushOlderThan collects all sealed chunks with maxT <= threshold from all +// series, writes them to an immutable block, and truncates the WAL. +// +// The ordering invariant is enforced: block fsync -> meta.json write -> WAL truncate. +// Returns the block ULID (empty string if nothing to flush) and any error. +func (h *Head) FlushOlderThan(maxT int64) (string, error) { + var flushData []block.SeriesFlush + + h.series.forEach(func(s *memSeries) { + s.mu.Lock() + defer s.mu.Unlock() + + var toFlush []chunkMeta + var remaining []chunkMeta + for _, cm := range s.sealed { + if cm.maxT <= maxT { + toFlush = append(toFlush, cm) + } else { + remaining = append(remaining, cm) + } + } + if len(toFlush) == 0 { + return + } + + sf := block.SeriesFlush{ + Ref: s.ref, + Labels: s.labels, + } + for _, cm := range toFlush { + sf.Chunks = append(sf.Chunks, block.ChunkData{ + MinT: cm.minT, + MaxT: cm.maxT, + Data: append([]byte(nil), cm.chunk.Bytes()...), + }) + } + flushData = append(flushData, sf) + + // Clear flushed chunks from the series. + s.sealed = remaining + }) + + if len(flushData) == 0 { + return "", nil + } + + // Write block. Flush handles: chunk files + index + fsync + meta.json. + ulid, err := block.Flush(h.dataDir, flushData) + if err != nil { + return "", fmt.Errorf("head: flush block: %w", err) + } + + // WAL truncation: safe because the block is fully fsynced. + // Truncate all segments below the current one — the flushed data is now + // in the block and doesn't need WAL replay. + lastSeg := h.wal.LastSegment() + if err := h.wal.Truncate(lastSeg); err != nil { + return ulid, fmt.Errorf("head: truncate WAL: %w", err) + } + + return ulid, nil +} + +// DataDir returns the data directory (parent of WAL dir). +func (h *Head) DataDir() string { + return h.dataDir +} diff --git a/internal/head/head_test.go b/internal/head/head_test.go index 7889a6f..e4080f8 100644 --- a/internal/head/head_test.go +++ b/internal/head/head_test.go @@ -2,10 +2,12 @@ package head import ( "math" + "os" "path/filepath" "sync" "testing" + "git.dvdt.dev/david/ingot/internal/block" "git.dvdt.dev/david/ingot/internal/wal" "git.dvdt.dev/david/ingot/labels" "github.com/stretchr/testify/assert" @@ -432,6 +434,181 @@ func TestWALReplay(t *testing.T) { } } +func TestFlushOlderThan(t *testing.T) { + tests := []struct { + name string + numSamples int // samples per series (each at 15s intervals) + flushMaxT int64 + wantBlockSeries int // number of series in the block + wantBlockExists bool + }{ + { + name: "flush_sealed_chunks", + numSamples: 250, // 2 sealed chunks (120 each) + 10 active + flushMaxT: math.MaxInt64, + wantBlockSeries: 1, + wantBlockExists: true, + }, + { + name: "nothing_to_flush", + numSamples: 50, // only active chunk, no sealed + flushMaxT: math.MaxInt64, + wantBlockSeries: 0, + wantBlockExists: false, + }, + { + name: "partial_flush_by_time", + numSamples: 250, + flushMaxT: 120 * 15000, // only flush first sealed chunk + wantBlockSeries: 1, + wantBlockExists: true, + }, + } + + for _, tc := range tests { + 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) + 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()) + + // Flush. + ulid, err := h.FlushOlderThan(tc.flushMaxT) + require.NoError(t, err) + + if !tc.wantBlockExists { + assert.Empty(t, ulid) + return + } + + assert.NotEmpty(t, ulid) + + // 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 { + it, err := br.SeriesChunkIterator(ref, math.MinInt64, math.MaxInt64) + require.NoError(t, err) + count := 0 + for it.Next() { + count++ + } + require.NoError(t, it.Err()) + assert.Greater(t, count, 0, "block should contain samples") + } + + // 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") + }) + } +} + +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) + } + require.NoError(t, app.Commit()) + + // Count WAL segments before flush. + walEntries, err := os.ReadDir(walDir) + require.NoError(t, err) + segsBefore := len(walEntries) + + // Flush. + _, err = h.FlushOlderThan(math.MaxInt64) + 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") + + require.NoError(t, h.Close()) + + // Re-open: head should recover from WAL (only unflushed data). + h2, err := Open(walDir, wal.Options{}) + require.NoError(t, err) + defer h2.Close() + + // 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()) +} + func TestConcurrentAppend(t *testing.T) { tests := []struct { name string diff --git a/internal/head/striped.go b/internal/head/striped.go index 3ac51a4..4d25b0a 100644 --- a/internal/head/striped.go +++ b/internal/head/striped.go @@ -70,6 +70,18 @@ func (sm *seriesMap) set(hash uint64, s *memSeries) { rs.mu.Unlock() } +// forEach calls fn for every series in the map. The series lock is NOT held. +func (sm *seriesMap) forEach(fn func(s *memSeries)) { + for i := range sm.refStripes { + rs := &sm.refStripes[i] + rs.mu.RLock() + for _, s := range rs.m { + fn(s) + } + rs.mu.RUnlock() + } +} + // remove deletes a series from both maps. func (sm *seriesMap) remove(hash uint64, s *memSeries) { hs := &sm.hashStripes[hash%numStripes] diff --git a/internal/index/index.go b/internal/index/index.go new file mode 100644 index 0000000..aab6fe3 --- /dev/null +++ b/internal/index/index.go @@ -0,0 +1,54 @@ +// Package index implements the binary index file format for ingot blocks. +// +// An index file contains four sections: +// +// [Header 5B] [Symbol Table] [Series] [Postings] [TOC 28B] +// +// The TOC at the end stores offsets to each section. Readers seek to the +// end, read the TOC, and use the offsets to locate each section. +package index + +import ( + "errors" + + "git.dvdt.dev/david/ingot/labels" +) + +const ( + indexMagic uint32 = 0x0FACEDB1 + indexVersion byte = 1 + headerLen = 5 // 4-byte magic + 1-byte version + tocLen = 28 // 3 offsets (24) + CRC (4) +) + +var ( + ErrInvalidMagic = errors.New("index: invalid magic number") + ErrInvalidVersion = errors.New("index: unsupported version") + ErrCorruptTOC = errors.New("index: corrupt TOC (CRC mismatch)") + ErrCorruptIndex = errors.New("index: corrupt index data") + ErrTooShort = errors.New("index: file too short") +) + +// ChunkRef encodes a chunk's location as (segment << 32 | offset). +type ChunkRef uint64 + +func NewChunkRef(segment, offset uint32) ChunkRef { + return ChunkRef(uint64(segment)<<32 | uint64(offset)) +} + +func (r ChunkRef) Segment() uint32 { return uint32(r >> 32) } +func (r ChunkRef) Offset() uint32 { return uint32(r) } + +// ChunkMeta describes a chunk's time range and location in a chunk file. +type ChunkMeta struct { + MinT int64 + MaxT int64 + Ref ChunkRef +} + +// SeriesEntry is the input for writing and the output for reading a series. +type SeriesEntry struct { + Ref uint64 + Labels []labels.Label + Chunks []ChunkMeta +} diff --git a/internal/index/index_test.go b/internal/index/index_test.go new file mode 100644 index 0000000..96c3c76 --- /dev/null +++ b/internal/index/index_test.go @@ -0,0 +1,208 @@ +package index + +import ( + "bytes" + "testing" + + "git.dvdt.dev/david/ingot/labels" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func writeIndex(t *testing.T, entries []SeriesEntry) []byte { + t.Helper() + var buf bytes.Buffer + w := NewWriter(&buf) + for _, e := range entries { + w.AddSeries(e) + } + _, err := w.WriteTo() + require.NoError(t, err) + return buf.Bytes() +} + +func TestIndexRoundTrip(t *testing.T) { + tests := []struct { + name string + entries []SeriesEntry + }{ + { + name: "single_series_single_chunk", + entries: []SeriesEntry{ + { + Ref: 1, + Labels: []labels.Label{{Name: "__name__", Value: "temp"}}, + Chunks: []ChunkMeta{{MinT: 1000, MaxT: 2000, Ref: NewChunkRef(1, 0)}}, + }, + }, + }, + { + name: "single_series_multiple_chunks", + entries: []SeriesEntry{ + { + Ref: 1, + Labels: []labels.Label{{Name: "__name__", Value: "temp"}, {Name: "room", Value: "office"}}, + Chunks: []ChunkMeta{ + {MinT: 1000, MaxT: 2000, Ref: NewChunkRef(1, 0)}, + {MinT: 2001, MaxT: 3000, Ref: NewChunkRef(1, 500)}, + {MinT: 3001, MaxT: 4000, Ref: NewChunkRef(1, 1000)}, + }, + }, + }, + }, + { + name: "multiple_series", + entries: []SeriesEntry{ + { + Ref: 1, + Labels: []labels.Label{{Name: "__name__", Value: "temp"}, {Name: "room", Value: "office"}}, + Chunks: []ChunkMeta{{MinT: 1000, MaxT: 2000, Ref: NewChunkRef(1, 0)}}, + }, + { + Ref: 2, + Labels: []labels.Label{{Name: "__name__", Value: "humidity"}, {Name: "room", Value: "office"}}, + Chunks: []ChunkMeta{{MinT: 1000, MaxT: 2000, Ref: NewChunkRef(1, 200)}}, + }, + { + Ref: 3, + Labels: []labels.Label{{Name: "__name__", Value: "temp"}, {Name: "room", Value: "kitchen"}}, + Chunks: []ChunkMeta{{MinT: 1000, MaxT: 2000, Ref: NewChunkRef(1, 400)}}, + }, + }, + }, + { + name: "empty", + entries: nil, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + data := writeIndex(t, tc.entries) + + r, err := NewReader(data) + require.NoError(t, err) + + // Verify series count and content. + gotSeries := r.Series() + require.Equal(t, len(tc.entries), len(gotSeries)) + + for i, want := range tc.entries { + got := gotSeries[i] + assert.Equal(t, want.Ref, got.Ref, "series %d ref", i) + assert.Equal(t, want.Labels, got.Labels, "series %d labels", i) + require.Equal(t, len(want.Chunks), len(got.Chunks), "series %d chunk count", i) + for j, wc := range want.Chunks { + assert.Equal(t, wc.MinT, got.Chunks[j].MinT, "series %d chunk %d minT", i, j) + assert.Equal(t, wc.MaxT, got.Chunks[j].MaxT, "series %d chunk %d maxT", i, j) + assert.Equal(t, wc.Ref, got.Chunks[j].Ref, "series %d chunk %d ref", i, j) + } + + // SeriesByRef lookup. + byRef, ok := r.SeriesByRef(want.Ref) + assert.True(t, ok, "series %d lookup by ref", i) + assert.Equal(t, want.Ref, byRef.Ref) + } + + // Verify postings. + for _, e := range tc.entries { + for _, l := range e.Labels { + refs := r.Postings(l.Name, l.Value) + assert.Contains(t, refs, e.Ref, "postings for %s=%s should contain ref %d", l.Name, l.Value, e.Ref) + } + } + + // Verify missing lookups return empty/false. + _, ok := r.SeriesByRef(999999) + assert.False(t, ok) + assert.Nil(t, r.Postings("nonexistent", "value")) + }) + } +} + +func TestIndexPostingsSorted(t *testing.T) { + entries := []SeriesEntry{ + {Ref: 5, Labels: []labels.Label{{Name: "room", Value: "office"}}, Chunks: []ChunkMeta{{MinT: 0, MaxT: 1, Ref: 0}}}, + {Ref: 2, Labels: []labels.Label{{Name: "room", Value: "office"}}, Chunks: []ChunkMeta{{MinT: 0, MaxT: 1, Ref: 0}}}, + {Ref: 8, Labels: []labels.Label{{Name: "room", Value: "office"}}, Chunks: []ChunkMeta{{MinT: 0, MaxT: 1, Ref: 0}}}, + } + + data := writeIndex(t, entries) + r, err := NewReader(data) + require.NoError(t, err) + + refs := r.Postings("room", "office") + require.Equal(t, 3, len(refs)) + assert.Equal(t, uint64(2), refs[0]) + assert.Equal(t, uint64(5), refs[1]) + assert.Equal(t, uint64(8), refs[2]) +} + +func TestIndexChunkRefEncoding(t *testing.T) { + tests := []struct { + name string + segment uint32 + offset uint32 + }{ + {"zero", 0, 0}, + {"first_segment", 1, 0}, + {"with_offset", 1, 12345}, + {"large_values", 100, 536870912}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + ref := NewChunkRef(tc.segment, tc.offset) + assert.Equal(t, tc.segment, ref.Segment()) + assert.Equal(t, tc.offset, ref.Offset()) + }) + } +} + +func TestIndexCorruptData(t *testing.T) { + tests := []struct { + name string + data []byte + wantErr error + }{ + { + name: "too_short", + data: []byte{1, 2, 3}, + wantErr: ErrTooShort, + }, + { + name: "bad_magic", + data: func() []byte { + d := writeIndex(t, nil) + d[0] = 0xFF + return d + }(), + wantErr: ErrInvalidMagic, + }, + { + name: "bad_version", + data: func() []byte { + d := writeIndex(t, nil) + d[4] = 99 + return d + }(), + wantErr: ErrInvalidVersion, + }, + { + name: "corrupt_toc_crc", + data: func() []byte { + d := writeIndex(t, nil) + d[len(d)-1] ^= 0xFF // flip CRC bits + return d + }(), + wantErr: ErrCorruptTOC, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + _, err := NewReader(tc.data) + assert.Equal(t, tc.wantErr, err) + }) + } +} diff --git a/internal/index/reader.go b/internal/index/reader.go new file mode 100644 index 0000000..8bc08cc --- /dev/null +++ b/internal/index/reader.go @@ -0,0 +1,215 @@ +package index + +import ( + "encoding/binary" + "hash/crc32" + + "git.dvdt.dev/david/ingot/labels" +) + +// Reader reads an index from a byte slice (typically mmap'd). +type Reader struct { + data []byte + + symbols []string + series []SeriesEntry + seriesByRef map[uint64]int // ref -> index into series + postings map[labelPair][]uint64 // label pair -> sorted refs +} + +// NewReader parses an index from data. +func NewReader(data []byte) (*Reader, error) { + if len(data) < headerLen+tocLen { + return nil, ErrTooShort + } + + // Validate header. + magic := binary.BigEndian.Uint32(data[:4]) + if magic != indexMagic { + return nil, ErrInvalidMagic + } + if data[4] != indexVersion { + return nil, ErrInvalidVersion + } + + // Read TOC from the last 28 bytes. + tocStart := len(data) - tocLen + toc := data[tocStart:] + + // Validate TOC CRC. + wantCRC := binary.BigEndian.Uint32(toc[24:28]) + gotCRC := crc32.Checksum(toc[:24], castagnoliTable) + if gotCRC != wantCRC { + return nil, ErrCorruptTOC + } + + symbolsOff := int(binary.BigEndian.Uint64(toc[0:8])) + seriesOff := int(binary.BigEndian.Uint64(toc[8:16])) + postingsOff := int(binary.BigEndian.Uint64(toc[16:24])) + + r := &Reader{ + data: data, + seriesByRef: make(map[uint64]int), + postings: make(map[labelPair][]uint64), + } + + if err := r.readSymbols(symbolsOff); err != nil { + return nil, err + } + if err := r.readSeries(seriesOff); err != nil { + return nil, err + } + if err := r.readPostings(postingsOff, tocStart); err != nil { + return nil, err + } + + return r, nil +} + +func (r *Reader) readSymbols(off int) error { + if off+4 > len(r.data) { + return ErrCorruptIndex + } + numSymbols := int(binary.BigEndian.Uint32(r.data[off : off+4])) + off += 4 + + r.symbols = make([]string, 0, numSymbols) + for i := 0; i < numSymbols; i++ { + if off+2 > len(r.data) { + return ErrCorruptIndex + } + slen := int(binary.BigEndian.Uint16(r.data[off : off+2])) + off += 2 + if off+slen > len(r.data) { + return ErrCorruptIndex + } + r.symbols = append(r.symbols, string(r.data[off:off+slen])) + off += slen + } + return nil +} + +func (r *Reader) readSeries(off int) error { + if off+4 > len(r.data) { + return ErrCorruptIndex + } + numSeries := int(binary.BigEndian.Uint32(r.data[off : off+4])) + off += 4 + + r.series = make([]SeriesEntry, 0, numSeries) + for i := 0; i < numSeries; i++ { + if off+8 > len(r.data) { + return ErrCorruptIndex + } + ref := binary.BigEndian.Uint64(r.data[off : off+8]) + off += 8 + + if off+2 > len(r.data) { + return ErrCorruptIndex + } + numLabels := int(binary.BigEndian.Uint16(r.data[off : off+2])) + off += 2 + + ls := make([]labels.Label, numLabels) + for j := 0; j < numLabels; j++ { + if off+8 > len(r.data) { + return ErrCorruptIndex + } + nameIdx := int(binary.BigEndian.Uint32(r.data[off : off+4])) + off += 4 + valueIdx := int(binary.BigEndian.Uint32(r.data[off : off+4])) + off += 4 + + if nameIdx >= len(r.symbols) || valueIdx >= len(r.symbols) { + return ErrCorruptIndex + } + ls[j] = labels.Label{Name: r.symbols[nameIdx], Value: r.symbols[valueIdx]} + } + + if off+4 > len(r.data) { + return ErrCorruptIndex + } + numChunks := int(binary.BigEndian.Uint32(r.data[off : off+4])) + off += 4 + + chunks := make([]ChunkMeta, numChunks) + for j := 0; j < numChunks; j++ { + if off+24 > len(r.data) { + return ErrCorruptIndex + } + chunks[j].MinT = int64(binary.BigEndian.Uint64(r.data[off : off+8])) + off += 8 + chunks[j].MaxT = int64(binary.BigEndian.Uint64(r.data[off : off+8])) + off += 8 + chunks[j].Ref = ChunkRef(binary.BigEndian.Uint64(r.data[off : off+8])) + off += 8 + } + + entry := SeriesEntry{Ref: ref, Labels: ls, Chunks: chunks} + r.seriesByRef[ref] = len(r.series) + r.series = append(r.series, entry) + } + return nil +} + +func (r *Reader) readPostings(off, limit int) error { + if off+4 > limit { + return ErrCorruptIndex + } + numEntries := int(binary.BigEndian.Uint32(r.data[off : off+4])) + off += 4 + + for i := 0; i < numEntries; i++ { + if off+12 > limit { + return ErrCorruptIndex + } + nameIdx := int(binary.BigEndian.Uint32(r.data[off : off+4])) + off += 4 + valueIdx := int(binary.BigEndian.Uint32(r.data[off : off+4])) + off += 4 + numRefs := int(binary.BigEndian.Uint32(r.data[off : off+4])) + off += 4 + + if nameIdx >= len(r.symbols) || valueIdx >= len(r.symbols) { + return ErrCorruptIndex + } + + if off+numRefs*8 > limit { + return ErrCorruptIndex + } + + refs := make([]uint64, numRefs) + for j := 0; j < numRefs; j++ { + refs[j] = binary.BigEndian.Uint64(r.data[off : off+8]) + off += 8 + } + + key := labelPair{r.symbols[nameIdx], r.symbols[valueIdx]} + r.postings[key] = refs + } + return nil +} + +// Symbols returns all symbols in the index. +func (r *Reader) Symbols() []string { + return r.symbols +} + +// Series returns all series entries. +func (r *Reader) Series() []SeriesEntry { + return r.series +} + +// SeriesByRef looks up a series by its ref. +func (r *Reader) SeriesByRef(ref uint64) (SeriesEntry, bool) { + idx, ok := r.seriesByRef[ref] + if !ok { + return SeriesEntry{}, false + } + return r.series[idx], true +} + +// Postings returns sorted series refs for the given label pair. +func (r *Reader) Postings(name, value string) []uint64 { + return r.postings[labelPair{name, value}] +} diff --git a/internal/index/writer.go b/internal/index/writer.go new file mode 100644 index 0000000..410ff44 --- /dev/null +++ b/internal/index/writer.go @@ -0,0 +1,273 @@ +package index + +import ( + "encoding/binary" + "hash/crc32" + "io" + "sort" + + "git.dvdt.dev/david/ingot/labels" +) + +var castagnoliTable = crc32.MakeTable(crc32.Castagnoli) + +// Writer builds an index file from series data. +type Writer struct { + w io.Writer + off int // bytes written so far + + symbols []string // ordered symbol list + symbolIdx map[string]int // string -> index in symbols + series []SeriesEntry + postings map[labelPair][]uint64 // label pair -> sorted series refs +} + +type labelPair struct { + name, value string +} + +// NewWriter creates a Writer that writes to w. +func NewWriter(w io.Writer) *Writer { + return &Writer{ + w: w, + symbolIdx: make(map[string]int), + postings: make(map[labelPair][]uint64), + } +} + +// AddSeries adds a series to the index. Series must be added before calling +// WriteTo. Labels must be sorted. +func (iw *Writer) AddSeries(entry SeriesEntry) { + for _, l := range entry.Labels { + iw.addSymbol(l.Name) + iw.addSymbol(l.Value) + } + iw.series = append(iw.series, entry) + + for _, l := range entry.Labels { + key := labelPair{l.Name, l.Value} + iw.postings[key] = append(iw.postings[key], entry.Ref) + } +} + +func (iw *Writer) addSymbol(s string) { + if _, ok := iw.symbolIdx[s]; ok { + return + } + idx := len(iw.symbols) + iw.symbols = append(iw.symbols, s) + iw.symbolIdx[s] = idx +} + +// WriteTo writes the complete index file. Returns bytes written and any error. +func (iw *Writer) WriteTo() (int, error) { + // Sort symbols for deterministic output. + sort.Strings(iw.symbols) + iw.symbolIdx = make(map[string]int, len(iw.symbols)) + for i, s := range iw.symbols { + iw.symbolIdx[s] = i + } + + iw.off = 0 + + // Header. + if err := iw.writeHeader(); err != nil { + return iw.off, err + } + + // Symbol table. + symbolsOff := iw.off + if err := iw.writeSymbols(); err != nil { + return iw.off, err + } + + // Series. + seriesOff := iw.off + if err := iw.writeSeries(); err != nil { + return iw.off, err + } + + // Postings. + postingsOff := iw.off + if err := iw.writePostings(); err != nil { + return iw.off, err + } + + // TOC. + if err := iw.writeTOC(symbolsOff, seriesOff, postingsOff); err != nil { + return iw.off, err + } + + return iw.off, nil +} + +func (iw *Writer) write(b []byte) error { + n, err := iw.w.Write(b) + iw.off += n + return err +} + +func (iw *Writer) writeHeader() error { + var buf [headerLen]byte + binary.BigEndian.PutUint32(buf[:4], indexMagic) + buf[4] = indexVersion + return iw.write(buf[:]) +} + +func (iw *Writer) writeSymbols() error { + var buf [4]byte + binary.BigEndian.PutUint32(buf[:], uint32(len(iw.symbols))) + if err := iw.write(buf[:]); err != nil { + return err + } + + var lenbuf [2]byte + for _, s := range iw.symbols { + binary.BigEndian.PutUint16(lenbuf[:], uint16(len(s))) + if err := iw.write(lenbuf[:]); err != nil { + return err + } + if err := iw.write([]byte(s)); err != nil { + return err + } + } + return nil +} + +func (iw *Writer) writeSeries() error { + var buf [8]byte + + // numSeries + binary.BigEndian.PutUint32(buf[:4], uint32(len(iw.series))) + if err := iw.write(buf[:4]); err != nil { + return err + } + + for _, s := range iw.series { + // ref + binary.BigEndian.PutUint64(buf[:8], s.Ref) + if err := iw.write(buf[:8]); err != nil { + return err + } + + // numLabels + binary.BigEndian.PutUint16(buf[:2], uint16(len(s.Labels))) + if err := iw.write(buf[:2]); err != nil { + return err + } + + // labels as symbol refs + for _, l := range s.Labels { + binary.BigEndian.PutUint32(buf[:4], uint32(iw.symbolIdx[l.Name])) + if err := iw.write(buf[:4]); err != nil { + return err + } + binary.BigEndian.PutUint32(buf[:4], uint32(iw.symbolIdx[l.Value])) + if err := iw.write(buf[:4]); err != nil { + return err + } + } + + // numChunks + binary.BigEndian.PutUint32(buf[:4], uint32(len(s.Chunks))) + if err := iw.write(buf[:4]); err != nil { + return err + } + + // chunks + for _, cm := range s.Chunks { + binary.BigEndian.PutUint64(buf[:8], uint64(cm.MinT)) + if err := iw.write(buf[:8]); err != nil { + return err + } + binary.BigEndian.PutUint64(buf[:8], uint64(cm.MaxT)) + if err := iw.write(buf[:8]); err != nil { + return err + } + binary.BigEndian.PutUint64(buf[:8], uint64(cm.Ref)) + if err := iw.write(buf[:8]); err != nil { + return err + } + } + } + return nil +} + +func (iw *Writer) writePostings() error { + // Sort postings keys for deterministic output. + keys := make([]labelPair, 0, len(iw.postings)) + for k := range iw.postings { + keys = append(keys, k) + } + sort.Slice(keys, func(i, j int) bool { + if keys[i].name != keys[j].name { + return keys[i].name < keys[j].name + } + return keys[i].value < keys[j].value + }) + + var buf [8]byte + + // numEntries + binary.BigEndian.PutUint32(buf[:4], uint32(len(keys))) + if err := iw.write(buf[:4]); err != nil { + return err + } + + for _, key := range keys { + refs := iw.postings[key] + + // nameSymIdx + binary.BigEndian.PutUint32(buf[:4], uint32(iw.symbolIdx[key.name])) + if err := iw.write(buf[:4]); err != nil { + return err + } + // valueSymIdx + binary.BigEndian.PutUint32(buf[:4], uint32(iw.symbolIdx[key.value])) + if err := iw.write(buf[:4]); err != nil { + return err + } + + // Sort refs. + sort.Slice(refs, func(i, j int) bool { return refs[i] < refs[j] }) + + // numRefs + binary.BigEndian.PutUint32(buf[:4], uint32(len(refs))) + if err := iw.write(buf[:4]); err != nil { + return err + } + + for _, ref := range refs { + binary.BigEndian.PutUint64(buf[:8], ref) + if err := iw.write(buf[:8]); err != nil { + return err + } + } + } + return nil +} + +func (iw *Writer) writeTOC(symbolsOff, seriesOff, postingsOff int) error { + var buf [tocLen]byte + binary.BigEndian.PutUint64(buf[0:8], uint64(symbolsOff)) + binary.BigEndian.PutUint64(buf[8:16], uint64(seriesOff)) + binary.BigEndian.PutUint64(buf[16:24], uint64(postingsOff)) + crc := crc32.Checksum(buf[:24], castagnoliTable) + binary.BigEndian.PutUint32(buf[24:28], crc) + return iw.write(buf[:]) +} + +// SymbolIndex returns the index of a symbol string, for use in lookups. +func (iw *Writer) SymbolIndex(s string) (int, bool) { + idx, ok := iw.symbolIdx[s] + return idx, ok +} + +// labelValues extracts the sorted label set for a series, resolving symbol refs. +func resolveLabelNames(ls []labels.Label) []labelPair { + pairs := make([]labelPair, len(ls)) + for i, l := range ls { + pairs[i] = labelPair{l.Name, l.Value} + } + return pairs +}