From 2822947687100d57ce1b90f6b34a54c9c5e3353d Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 25 May 2026 14:56:39 -0600 Subject: [PATCH] refactor(gorilla-merger): import coldpart from asap-gorilla-go, delete internal copy The cold "Part" format moved to the shared asap-gorilla-go/coldpart package so the edge agent can produce parts too. Re-point the merger's cold part store and decode-on-read query path at github.com/ProjectASAP/asap-gorilla-go/coldpart (resolved via the existing replace to the sibling asap-gorilla-go checkout) and delete the now-duplicate internal/coldpart package. No behavior change; the wire format and StoreAPI are byte-for-byte identical. Depends on the ASAPCollector change that adds asap-gorilla-go/coldpart; merge that PR first. Co-Authored-By: Claude Opus 4.7 (1M context) --- gorilla-merger/internal/coldpart/coldpart.go | 611 ------------------ .../internal/coldpart/coldpart_test.go | 580 ----------------- .../internal/merger/coldpartstore.go | 3 +- .../internal/merger/coldpartstore_test.go | 3 +- gorilla-merger/internal/merger/coldquery.go | 2 +- 5 files changed, 3 insertions(+), 1196 deletions(-) delete mode 100644 gorilla-merger/internal/coldpart/coldpart.go delete mode 100644 gorilla-merger/internal/coldpart/coldpart_test.go diff --git a/gorilla-merger/internal/coldpart/coldpart.go b/gorilla-merger/internal/coldpart/coldpart.go deleted file mode 100644 index 5eea2204e..000000000 --- a/gorilla-merger/internal/coldpart/coldpart.go +++ /dev/null @@ -1,611 +0,0 @@ -// Package coldpart reads and writes the cold "Part" object: the on-disk / -// in-S3 container for a tenant's raw metric samples over one time block. A Part -// bundles many series, each stored as one or more best-of-N lossless value -// chunks (asap-gorilla-go/intchunk), behind a label index + symbol table so a -// reader can answer Series(matchers, mint, maxt) WITHOUT decoding any chunk -// body — chunk bodies are decoded lazily only for the series that match. -// -// This is the format-defining building block of the decode-on-read cold tier. -// It is intentionally standalone: it does NOT touch the existing fragment -// ingest -> tsdb path, and the agent does not emit this format yet. A future -// decode-on-read StoreAPI builds Series() responses on top of OpenPart + -// (*Part).Series. -// -// On-disk layout (all multi-byte integers little-endian unless noted uvarint): -// -// [part header] magic "ASAPCC1" | u8 version | i64 block_start_ms | -// i64 block_end_ms | uvarint series_count -// [chunks] per-series intchunk value chunks, concatenated. A series may -// occupy >1 intchunk chunk (intchunk cuts on residual overflow); -// its whole run is [chunk_off, chunk_off+chunk_len). -// [index] per series, sorted by labels: label refs into the symbol -// table, u64 chunk_off, u32 chunk_len, uvarint chunk_count, -// i64 min_ts, i64 max_ts. -// [symbol table] deduped label strings; the index references them by ordinal. -// [footer] u64 index_off | u64 index_len | u64 symtab_off | u32 crc32c -// -// The crc32c (Castagnoli) covers every byte of the part before the crc field -// and is verified by OpenPart, so a corrupt object is rejected before any chunk -// is decoded. -package coldpart - -import ( - "encoding/binary" - "errors" - "fmt" - "hash/crc32" - "io" - "sort" - - "github.com/ProjectASAP/asap-gorilla-go/intchunk" - "github.com/prometheus/prometheus/model/labels" -) - -// Magic is the 7-byte part magic. Version 1 is the layout documented above. -const ( - Magic = "ASAPCC1" - Version = uint8(1) - - magicLen = len(Magic) - footerLen = 8 + 8 + 8 + 4 // index_off + index_len + symtab_off + crc32c -) - -// crc32cTable is the Castagnoli polynomial table used for the footer checksum. -var crc32cTable = crc32.MakeTable(crc32.Castagnoli) - -// Errors returned by the package. -var ( - ErrShort = errors.New("coldpart: buffer too short") - ErrBadMagic = errors.New("coldpart: bad magic") - ErrBadVersion = errors.New("coldpart: unsupported version") - ErrBadCRC = errors.New("coldpart: crc32c mismatch") - ErrCorrupt = errors.New("coldpart: corrupt part") - ErrNoSamples = errors.New("coldpart: series has no samples") - ErrBadTimeRange = errors.New("coldpart: block_end_ms < block_start_ms") -) - -// Sample is one timestamp (ms) / float64-value point. It re-exports -// intchunk.Sample so callers need not import intchunk to read decoded points. -type Sample = intchunk.Sample - -// Series is one logical series: an immutable label set plus its time-ordered -// samples. Samples MUST be sorted by ascending timestamp before WritePart — -// the value codec and the [min_ts,max_ts] index entry assume time order. -type Series struct { - Labels labels.Labels - Samples []Sample -} - -// Options configures WritePart. It is reserved for future codec/layout knobs -// (e.g. the shared-timestamp grouped layout); the zero value is the default and -// is what every current caller should pass. -type Options struct{} - -// --------------------------------------------------------------------------- -// Write path -// --------------------------------------------------------------------------- - -// WritePart encodes header + per-series intchunk value chunks + label index + -// symbol table + footer (with crc32c) to w. Series may be supplied in any -// order; WritePart sorts them by label set so the index is canonical and a -// reader can rely on sorted-by-series order. Every series must have at least -// one sample. -// -// blockStartMs/blockEndMs describe the time block this part covers; they are -// recorded verbatim in the header and are not required to bound the per-series -// sample timestamps (the per-series [min_ts,max_ts] index entries are the -// authoritative time bounds used by Series()). -func WritePart(w io.Writer, blockStartMs, blockEndMs int64, series []Series, _ Options) error { - if blockEndMs < blockStartMs { - return ErrBadTimeRange - } - - // Sort a copy by label set so we never mutate the caller's slice order. - sorted := make([]Series, len(series)) - copy(sorted, series) - sort.Slice(sorted, func(i, j int) bool { - return labels.Compare(sorted[i].Labels, sorted[j].Labels) < 0 - }) - - // Build the symbol table (deduped label strings) and remember each string's - // ordinal so the index can reference symbols compactly. - symtab := newSymbolTable() - for i := range sorted { - if len(sorted[i].Samples) == 0 { - return fmt.Errorf("%w: series %d", ErrNoSamples, i) - } - sorted[i].Labels.Range(func(l labels.Label) { - symtab.intern(l.Name) - symtab.intern(l.Value) - }) - } - - // Encode each series' chunk run and record its index entry. We assemble the - // chunks region first (so chunk_off is known), then the index, then the - // symbol table, then the footer. - type idxEntry struct { - labelRefs []uint64 // [name0,val0,name1,val1,...] ordinals into symtab - chunkOff uint64 - chunkLen uint32 - chunkLens []int // per-intchunk-chunk byte lengths within the run - minTS int64 - maxTS int64 - } - - var chunksRegion []byte - entries := make([]idxEntry, len(sorted)) - for i := range sorted { - s := sorted[i] - res, err := intchunk.Encode(s.Samples) - if err != nil { - return fmt.Errorf("coldpart: encode series %d: %w", i, err) - } - off := uint64(len(chunksRegion)) - var runLen int - lens := make([]int, 0, len(res.Chunks)) - for _, c := range res.Chunks { - chunksRegion = append(chunksRegion, c...) - runLen += len(c) - lens = append(lens, len(c)) - } - - refs := make([]uint64, 0, sorted[i].Labels.Len()*2) - s.Labels.Range(func(l labels.Label) { - refs = append(refs, symtab.ref(l.Name), symtab.ref(l.Value)) - }) - - entries[i] = idxEntry{ - labelRefs: refs, - chunkOff: off, - chunkLen: uint32(runLen), - chunkLens: lens, - minTS: s.Samples[0].T, - maxTS: s.Samples[len(s.Samples)-1].T, - } - } - - // --- assemble the full byte image, then crc + emit --- - var buf []byte - bw := &writer{buf: &buf} - - // Header. - bw.bytes([]byte(Magic)) - bw.u8(Version) - bw.i64(blockStartMs) - bw.i64(blockEndMs) - bw.uvarint(uint64(len(sorted))) - - // Chunks region. Each entry's chunkOff was computed relative to the region - // start; we add the region's absolute position so the index stores absolute - // part offsets the reader can slice directly. - chunksStart := uint64(len(buf)) - bw.bytes(chunksRegion) - - // Index. - indexOff := uint64(len(buf)) - for i := range entries { - e := entries[i] - bw.uvarint(uint64(len(e.labelRefs) / 2)) // label count - for _, r := range e.labelRefs { - bw.uvarint(r) - } - bw.u64(chunksStart + e.chunkOff) // absolute offset of the series' chunk run - bw.u32(e.chunkLen) - bw.uvarint(uint64(len(e.chunkLens))) // chunk_count - for _, cl := range e.chunkLens { - bw.uvarint(uint64(cl)) // per-chunk byte length, in run order - } - bw.i64(e.minTS) - bw.i64(e.maxTS) - } - indexLen := uint64(len(buf)) - indexOff - - // Symbol table. - symtabOff := uint64(len(buf)) - bw.uvarint(uint64(len(symtab.syms))) - for _, s := range symtab.syms { - bw.uvarint(uint64(len(s))) - bw.bytes([]byte(s)) - } - - // Footer (crc32c covers everything written so far). - bw.u64(indexOff) - bw.u64(indexLen) - bw.u64(symtabOff) - crc := crc32.Checksum(buf, crc32cTable) - bw.u32(crc) - - if _, err := w.Write(buf); err != nil { - return fmt.Errorf("coldpart: write: %w", err) - } - return nil -} - -// --------------------------------------------------------------------------- -// Read path -// --------------------------------------------------------------------------- - -// Part is a parsed, validated cold part. OpenPart populates its header and -// per-series index WITHOUT decoding any chunk body; chunk bodies are decoded -// lazily by Series(). A Part holds a reference to the underlying buffer; the -// caller must keep that buffer alive (and unmodified) for the Part's lifetime. -type Part struct { - Version uint8 - BlockStartMs int64 - BlockEndMs int64 - - buf []byte - symbols []string - series []indexSeries -} - -// indexSeries is one parsed index entry. Labels are materialized eagerly (cheap -// — just symbol-table lookups); only the chunk body is deferred. -type indexSeries struct { - lbls labels.Labels - chunkOff uint64 - chunkLen uint32 - chunkLens []uint64 // per-intchunk-chunk byte lengths within the run - minTS int64 - maxTS int64 -} - -// SeriesData is a matched series with its decoded, time-ordered samples. -type SeriesData struct { - Labels labels.Labels - Samples []Sample -} - -// OpenPart validates the magic/version/crc and parses the footer -> index -> -// symbol table of b. It does NOT decode chunk bodies. The returned Part borrows -// b; do not mutate b while the Part is in use. -func OpenPart(b []byte) (*Part, error) { - if len(b) < magicLen+1+8+8+1+footerLen { - return nil, ErrShort - } - if string(b[:magicLen]) != Magic { - return nil, ErrBadMagic - } - - // Verify crc32c over everything but the trailing 4-byte crc field. - stored := binary.LittleEndian.Uint32(b[len(b)-4:]) - if crc32.Checksum(b[:len(b)-4], crc32cTable) != stored { - return nil, ErrBadCRC - } - - // Header. - r := &reader{buf: b, pos: magicLen} - ver, err := r.u8() - if err != nil { - return nil, err - } - if ver != Version { - return nil, fmt.Errorf("%w: %d", ErrBadVersion, ver) - } - blockStart, err := r.i64() - if err != nil { - return nil, err - } - blockEnd, err := r.i64() - if err != nil { - return nil, err - } - seriesCount, err := r.uvarint() - if err != nil { - return nil, err - } - - // Footer. - footerStart := len(b) - footerLen - fr := &reader{buf: b, pos: footerStart} - indexOff, _ := fr.u64() - indexLen, _ := fr.u64() - symtabOff, _ := fr.u64() - if indexOff > uint64(len(b)) || symtabOff > uint64(len(b)) || - indexOff+indexLen > uint64(len(b)) || indexOff > symtabOff { - return nil, fmt.Errorf("%w: footer offsets out of range", ErrCorrupt) - } - - // Symbol table: [uvarint count] then count x ([uvarint len] bytes). - sr := &reader{buf: b, pos: int(symtabOff)} - symCount, err := sr.uvarint() - if err != nil { - return nil, err - } - symbols := make([]string, 0, symCount) - for i := uint64(0); i < symCount; i++ { - n, err := sr.uvarint() - if err != nil { - return nil, err - } - s, err := sr.take(int(n)) - if err != nil { - return nil, err - } - symbols = append(symbols, string(s)) - } - - // Index: seriesCount entries, each ending before symtabOff. - ir := &reader{buf: b, pos: int(indexOff)} - parsed := make([]indexSeries, 0, seriesCount) - for i := uint64(0); i < seriesCount; i++ { - nLabels, err := ir.uvarint() - if err != nil { - return nil, err - } - lb := labels.NewBuilder(labels.EmptyLabels()) - for j := uint64(0); j < nLabels; j++ { - nameRef, err := ir.uvarint() - if err != nil { - return nil, err - } - valRef, err := ir.uvarint() - if err != nil { - return nil, err - } - if nameRef >= symCount || valRef >= symCount { - return nil, fmt.Errorf("%w: symbol ref out of range", ErrCorrupt) - } - lb.Set(symbols[nameRef], symbols[valRef]) - } - chunkOff, err := ir.u64() - if err != nil { - return nil, err - } - chunkLen, err := ir.u32() - if err != nil { - return nil, err - } - chunkCount, err := ir.uvarint() - if err != nil { - return nil, err - } - chunkLens := make([]uint64, 0, chunkCount) - var lensSum uint64 - for j := uint64(0); j < chunkCount; j++ { - cl, err := ir.uvarint() - if err != nil { - return nil, err - } - chunkLens = append(chunkLens, cl) - lensSum += cl - } - minTS, err := ir.i64() - if err != nil { - return nil, err - } - maxTS, err := ir.i64() - if err != nil { - return nil, err - } - if chunkOff+uint64(chunkLen) > uint64(len(b)) { - return nil, fmt.Errorf("%w: chunk run out of range", ErrCorrupt) - } - if lensSum != uint64(chunkLen) { - return nil, fmt.Errorf("%w: chunk length sum %d != run length %d", ErrCorrupt, lensSum, chunkLen) - } - parsed = append(parsed, indexSeries{ - lbls: lb.Labels(), - chunkOff: chunkOff, - chunkLen: chunkLen, - chunkLens: chunkLens, - minTS: minTS, - maxTS: maxTS, - }) - } - - return &Part{ - Version: ver, - BlockStartMs: blockStart, - BlockEndMs: blockEnd, - buf: b, - symbols: symbols, - series: parsed, - }, nil -} - -// NumSeries reports how many series the part indexes (without decoding any). -func (p *Part) NumSeries() int { return len(p.series) } - -// SeriesLabels returns the label set of every indexed series, in part-stored -// (sorted-by-labels) order, WITHOUT decoding any chunk body. It lets a caller -// (e.g. a part manifest) record which series a part covers for overlap/matcher -// pre-filtering. The returned slice is freshly allocated; the labels.Labels -// values were materialized at OpenPart time and are safe to retain. -func (p *Part) SeriesLabels() []labels.Labels { - out := make([]labels.Labels, len(p.series)) - for i := range p.series { - out[i] = p.series[i].lbls - } - return out -} - -// Series returns every indexed series whose [min_ts,max_ts] overlaps the -// half-open-ish inclusive window [mintMs,maxtMs] AND whose labels satisfy all -// matchers, each with its samples decoded. Chunk bodies are decoded lazily — -// only for the matched, time-overlapping series. The result is in part-stored -// (sorted-by-labels) order. -// -// Time filtering uses inclusive overlap: a series is included when -// series.min_ts <= maxtMs && series.max_ts >= mintMs. Matcher filtering applies -// every matcher with AND semantics via (*labels.Matcher).Matches against the -// series' value for that matcher's label name (the empty string when absent), -// which makes =, !=, =~, !~ behave exactly as Prometheus selectors do. -func (p *Part) Series(matchers []*labels.Matcher, mintMs, maxtMs int64) ([]SeriesData, error) { - var out []SeriesData - for i := range p.series { - s := &p.series[i] - // Time-window overlap (inclusive on both ends). - if s.minTS > maxtMs || s.maxTS < mintMs { - continue - } - if !MatchesAll(s.lbls, matchers) { - continue - } - samples, err := p.decodeSeries(s) - if err != nil { - return nil, err - } - out = append(out, SeriesData{Labels: s.lbls, Samples: samples}) - } - return out, nil -} - -// decodeSeries decodes the chunk run of one indexed series. The run is the -// concatenation of one-or-more self-contained intchunk chunks (intchunk cuts a -// series into several chunks on residual overflow). intchunk exposes no -// "bytes-consumed" boundary on a concatenation, so the part index records each -// chunk's byte length explicitly; decodeSeries re-slices the run on those -// boundaries and decodes each chunk with intchunk. The common single-chunk case -// decodes the whole run directly. -func (p *Part) decodeSeries(s *indexSeries) ([]Sample, error) { - run := p.buf[s.chunkOff : s.chunkOff+uint64(s.chunkLen)] - if len(s.chunkLens) <= 1 { - samples, err := intchunk.DecodeChunk(run) - if err != nil { - return nil, fmt.Errorf("coldpart: decode chunk: %w", err) - } - return samples, nil - } - chunks := splitChunks(run, s.chunkLens) - samples, err := intchunk.DecodeChunks(chunks) - if err != nil { - return nil, fmt.Errorf("coldpart: decode chunks: %w", err) - } - return samples, nil -} - -// splitChunks slices run into the sub-chunks described by lens. The caller -// (OpenPart) has already validated that the lengths sum to len(run). -func splitChunks(run []byte, lens []uint64) [][]byte { - chunks := make([][]byte, 0, len(lens)) - off := uint64(0) - for _, l := range lens { - chunks = append(chunks, run[off:off+l]) - off += l - } - return chunks -} - -// MatchesAll reports whether ls satisfies every matcher (AND semantics), with a -// matcher applied against ls's value for its label name (the empty string when -// absent) so =, !=, =~, !~ behave exactly as Prometheus selectors do. A nil -// matcher is skipped. It is exported so a part manifest can apply the SAME -// matcher semantics when pre-filtering whole parts. -func MatchesAll(ls labels.Labels, matchers []*labels.Matcher) bool { - for _, m := range matchers { - if m == nil { - continue - } - if !m.Matches(ls.Get(m.Name)) { - return false - } - } - return true -} - -// --------------------------------------------------------------------------- -// symbol table (write side) -// --------------------------------------------------------------------------- - -type symbolTable struct { - syms []string - idx map[string]uint64 -} - -func newSymbolTable() *symbolTable { - return &symbolTable{idx: make(map[string]uint64)} -} - -// intern adds s to the table if absent. -func (t *symbolTable) intern(s string) { - if _, ok := t.idx[s]; ok { - return - } - t.idx[s] = uint64(len(t.syms)) - t.syms = append(t.syms, s) -} - -// ref returns the ordinal of an already-interned string. -func (t *symbolTable) ref(s string) uint64 { return t.idx[s] } - -// --------------------------------------------------------------------------- -// little-endian / varint writer + reader helpers -// --------------------------------------------------------------------------- - -type writer struct{ buf *[]byte } - -func (w *writer) bytes(b []byte) { *w.buf = append(*w.buf, b...) } -func (w *writer) u8(v uint8) { *w.buf = append(*w.buf, v) } - -func (w *writer) u32(v uint32) { - var b [4]byte - binary.LittleEndian.PutUint32(b[:], v) - *w.buf = append(*w.buf, b[:]...) -} - -func (w *writer) u64(v uint64) { - var b [8]byte - binary.LittleEndian.PutUint64(b[:], v) - *w.buf = append(*w.buf, b[:]...) -} - -func (w *writer) i64(v int64) { w.u64(uint64(v)) } - -func (w *writer) uvarint(v uint64) { - var tmp [binary.MaxVarintLen64]byte - n := binary.PutUvarint(tmp[:], v) - *w.buf = append(*w.buf, tmp[:n]...) -} - -type reader struct { - buf []byte - pos int -} - -func (r *reader) take(n int) ([]byte, error) { - if n < 0 || r.pos+n > len(r.buf) { - return nil, ErrShort - } - b := r.buf[r.pos : r.pos+n] - r.pos += n - return b, nil -} - -func (r *reader) u8() (uint8, error) { - b, err := r.take(1) - if err != nil { - return 0, err - } - return b[0], nil -} - -func (r *reader) u32() (uint32, error) { - b, err := r.take(4) - if err != nil { - return 0, err - } - return binary.LittleEndian.Uint32(b), nil -} - -func (r *reader) u64() (uint64, error) { - b, err := r.take(8) - if err != nil { - return 0, err - } - return binary.LittleEndian.Uint64(b), nil -} - -func (r *reader) i64() (int64, error) { - v, err := r.u64() - return int64(v), err -} - -func (r *reader) uvarint() (uint64, error) { - v, n := binary.Uvarint(r.buf[r.pos:]) - if n <= 0 { - return 0, ErrCorrupt - } - r.pos += n - return v, nil -} diff --git a/gorilla-merger/internal/coldpart/coldpart_test.go b/gorilla-merger/internal/coldpart/coldpart_test.go deleted file mode 100644 index e94624056..000000000 --- a/gorilla-merger/internal/coldpart/coldpart_test.go +++ /dev/null @@ -1,580 +0,0 @@ -package coldpart - -import ( - "bytes" - "hash/crc32" - "math" - "testing" - - "github.com/ProjectASAP/asap-gorilla-go/intchunk" - "github.com/prometheus/prometheus/model/labels" -) - -// --------------------------------------------------------------------------- -// helpers -// --------------------------------------------------------------------------- - -// lbls builds a label set from alternating name/value pairs. -func lbls(kv ...string) labels.Labels { - if len(kv)%2 != 0 { - panic("lbls: odd number of args") - } - pairs := make([]labels.Label, 0, len(kv)/2) - for i := 0; i < len(kv); i += 2 { - pairs = append(pairs, labels.Label{Name: kv[i], Value: kv[i+1]}) - } - return labels.New(pairs...) -} - -// mustMatcher builds a matcher or fails the test. -func mustMatcher(t *testing.T, ty labels.MatchType, n, v string) *labels.Matcher { - t.Helper() - m, err := labels.NewMatcher(ty, n, v) - if err != nil { - t.Fatalf("NewMatcher(%v,%q,%q): %v", ty, n, v, err) - } - return m -} - -// assertSamplesEqual fails unless got==want in length, every timestamp, and -// every value BIT-EXACTLY (bit-exactness is the whole point for the -// high-precision-float / Gorilla-fallback case). -func assertSamplesEqual(t *testing.T, what string, got, want []Sample) { - t.Helper() - if len(got) != len(want) { - t.Fatalf("%s: got %d samples, want %d", what, len(got), len(want)) - } - for i := range want { - if got[i].T != want[i].T || math.Float64bits(got[i].V) != math.Float64bits(want[i].V) { - t.Fatalf("%s: sample %d mismatch: got (%d,%v) want (%d,%v)", - what, i, got[i].T, got[i].V, want[i].T, want[i].V) - } - } -} - -// findSeries returns the SeriesData whose labels equal want, or fails. -func findSeries(t *testing.T, got []SeriesData, want labels.Labels) SeriesData { - t.Helper() - for _, sd := range got { - if labels.Compare(sd.Labels, want) == 0 { - return sd - } - } - t.Fatalf("series %s not found in result", want.String()) - return SeriesData{} -} - -// seriesValues builds an ascending-timestamp series at 1s cadence from values. -func seriesValues(start int64, vals ...float64) []Sample { - out := make([]Sample, len(vals)) - for i, v := range vals { - out[i] = Sample{T: start + int64(i)*1000, V: v} - } - return out -} - -// roundTrip writes the series to a part, reopens it, queries all series with -// the widest window, and returns the parsed Part plus the matched series. -func roundTrip(t *testing.T, blockStart, blockEnd int64, series []Series) (*Part, []SeriesData) { - t.Helper() - var buf bytes.Buffer - if err := WritePart(&buf, blockStart, blockEnd, series, Options{}); err != nil { - t.Fatalf("WritePart: %v", err) - } - p, err := OpenPart(buf.Bytes()) - if err != nil { - t.Fatalf("OpenPart: %v", err) - } - got, err := p.Series(nil, math.MinInt64, math.MaxInt64) - if err != nil { - t.Fatalf("Series: %v", err) - } - return p, got -} - -// --------------------------------------------------------------------------- -// codec-exercising series (one per intchunk sub-codec) -// --------------------------------------------------------------------------- -// -// These value patterns were verified to select the named intchunk codec via -// best-of-N (see the codec assertions in TestSubCodecsRoundTrip). - -func gaugeIntSeries() []Sample { // -> INT_FOR_DELTA (fixed-width) - return seriesValues(1_000_000, 10, 11, 9, 12, 8, 15, 7, 20, 5, 25) -} - -func counterLinearSeries() []Sample { // -> INT_FOR_DOD - vals := make([]float64, 20) - for i := range vals { - vals[i] = float64(i * 7) - } - return seriesValues(1_000_000, vals...) -} - -func decimalSeries() []Sample { // fixed-decimal gauge -> INT_FOR_DELTA - return seriesValues(1_000_000, 1.1, 1.2, 1.3, 1.1, 1.5, 1.9, 1.2) -} - -func skewedSeries() []Sample { // skewed residuals -> INT_FOR_DELTA_VARINT - out := make([]Sample, 0, 30) - base := int64(0) - for i := 0; i < 30; i++ { - if i == 15 { - base += 1_000_000 - } else { - base++ - } - out = append(out, Sample{T: 1_000_000 + int64(i)*1000, V: float64(base)}) - } - return out -} - -func highPrecSeries() []Sample { // true high-precision floats -> GORILLA_XOR fallback - return seriesValues(1_000_000, - 0.1234567890123456, 3.141592653589793, - 2.718281828459045, 1.4142135623730951, math.Pi*1e-7) -} - -// --------------------------------------------------------------------------- -// tests -// --------------------------------------------------------------------------- - -// TestSubCodecsRoundTrip writes one series per intchunk sub-codec into a single -// part and asserts each decodes back BIT-EXACTLY, and that our value patterns -// actually exercise the distinct codecs (gauge/counter/decimal/varint/float). -func TestSubCodecsRoundTrip(t *testing.T) { - type tc struct { - name string - lset labels.Labels - samples []Sample - wantTag intchunk.CodecTag - } - cases := []tc{ - {"gauge", lbls("__name__", "cpu", "kind", "gauge"), gaugeIntSeries(), intchunk.CodecIntForDelta}, - {"counter", lbls("__name__", "reqs", "kind", "counter"), counterLinearSeries(), intchunk.CodecIntForDoD}, - {"decimal", lbls("__name__", "temp", "kind", "decimal"), decimalSeries(), intchunk.CodecIntForDelta}, - {"skewed", lbls("__name__", "skew", "kind", "varint"), skewedSeries(), intchunk.CodecIntForDeltaVarint}, - {"highprec", lbls("__name__", "ratio", "kind", "float"), highPrecSeries(), intchunk.CodecGorillaXOR}, - } - - // Assert each pattern selects the intended codec, so the round-trip below - // genuinely exercises every sub-codec path. - for _, c := range cases { - res, err := intchunk.Encode(c.samples) - if err != nil { - t.Fatalf("%s: intchunk.Encode: %v", c.name, err) - } - if res.Tag != c.wantTag { - t.Fatalf("%s: codec = %s, want %s", c.name, res.Tag, c.wantTag) - } - } - - series := make([]Series, len(cases)) - for i, c := range cases { - series[i] = Series{Labels: c.lset, Samples: c.samples} - } - - _, got := roundTrip(t, 1_000_000, 2_000_000, series) - if len(got) != len(cases) { - t.Fatalf("got %d series, want %d", len(got), len(cases)) - } - for _, c := range cases { - sd := findSeries(t, got, c.lset) - assertSamplesEqual(t, c.name, sd.Samples, c.samples) - } -} - -// TestHeaderAndCount checks the parsed header fields and series count. -func TestHeaderAndCount(t *testing.T) { - series := []Series{ - {Labels: lbls("__name__", "a"), Samples: seriesValues(0, 1, 2, 3)}, - {Labels: lbls("__name__", "b"), Samples: seriesValues(0, 4, 5, 6)}, - } - p, got := roundTrip(t, 1000, 7_201_000, series) - if p.Version != Version { - t.Fatalf("version = %d, want %d", p.Version, Version) - } - if p.BlockStartMs != 1000 || p.BlockEndMs != 7_201_000 { - t.Fatalf("block range = [%d,%d], want [1000,7201000]", p.BlockStartMs, p.BlockEndMs) - } - if p.NumSeries() != 2 { - t.Fatalf("NumSeries = %d, want 2", p.NumSeries()) - } - if len(got) != 2 { - t.Fatalf("Series returned %d, want 2", len(got)) - } -} - -// TestSeriesSortedByLabels verifies series come back canonical (sorted) order -// regardless of input order. -func TestSeriesSortedByLabels(t *testing.T) { - series := []Series{ - {Labels: lbls("__name__", "zeta"), Samples: seriesValues(0, 1)}, - {Labels: lbls("__name__", "alpha"), Samples: seriesValues(0, 2)}, - {Labels: lbls("__name__", "mu"), Samples: seriesValues(0, 3)}, - } - _, got := roundTrip(t, 0, 1000, series) - want := []string{"alpha", "mu", "zeta"} - if len(got) != len(want) { - t.Fatalf("got %d series", len(got)) - } - for i, w := range want { - if name := got[i].Labels.Get("__name__"); name != w { - t.Fatalf("series[%d] = %q, want %q", i, name, w) - } - } -} - -// TestMatcherFiltering exercises =, !=, =~ matchers (AND semantics) and a -// matcher against an absent label name. -func TestMatcherFiltering(t *testing.T) { - series := []Series{ - {Labels: lbls("__name__", "http_requests", "job", "api", "code", "200"), Samples: seriesValues(0, 1, 2)}, - {Labels: lbls("__name__", "http_requests", "job", "api", "code", "500"), Samples: seriesValues(0, 3, 4)}, - {Labels: lbls("__name__", "http_requests", "job", "web", "code", "200"), Samples: seriesValues(0, 5, 6)}, - {Labels: lbls("__name__", "cpu_seconds", "job", "api"), Samples: seriesValues(0, 7, 8)}, - } - var buf bytes.Buffer - if err := WritePart(&buf, 0, 10_000, series, Options{}); err != nil { - t.Fatalf("WritePart: %v", err) - } - p, err := OpenPart(buf.Bytes()) - if err != nil { - t.Fatalf("OpenPart: %v", err) - } - - full := int64(math.MaxInt64) - min := int64(math.MinInt64) - - // = exact name - got, _ := p.Series([]*labels.Matcher{ - mustMatcher(t, labels.MatchEqual, "__name__", "http_requests"), - }, min, full) - if len(got) != 3 { - t.Fatalf("= filter: got %d, want 3", len(got)) - } - - // = AND = (name + job) - got, _ = p.Series([]*labels.Matcher{ - mustMatcher(t, labels.MatchEqual, "__name__", "http_requests"), - mustMatcher(t, labels.MatchEqual, "job", "api"), - }, min, full) - if len(got) != 2 { - t.Fatalf("=&= filter: got %d, want 2", len(got)) - } - - // != excludes code 500 - got, _ = p.Series([]*labels.Matcher{ - mustMatcher(t, labels.MatchEqual, "__name__", "http_requests"), - mustMatcher(t, labels.MatchNotEqual, "code", "500"), - }, min, full) - if len(got) != 2 { - t.Fatalf("!= filter: got %d, want 2", len(got)) - } - - // =~ regex on job - got, _ = p.Series([]*labels.Matcher{ - mustMatcher(t, labels.MatchRegexp, "job", "a.*"), - }, min, full) - // matches api (http x2) + api (cpu) = 3 - if len(got) != 3 { - t.Fatalf("=~ filter: got %d, want 3", len(got)) - } - - // !~ regex on code, with code label absent on cpu_seconds. - // MatchNotRegexp on an empty value: "200" !~ "5.." is true, "" !~ "5.." true. - got, _ = p.Series([]*labels.Matcher{ - mustMatcher(t, labels.MatchNotRegexp, "code", "5.."), - }, min, full) - // excludes only the code=500 series => 3 remain - if len(got) != 3 { - t.Fatalf("!~ filter: got %d, want 3", len(got)) - } - - // = on an absent label matches the empty string: code="" hits cpu_seconds. - got, _ = p.Series([]*labels.Matcher{ - mustMatcher(t, labels.MatchEqual, "code", ""), - }, min, full) - if len(got) != 1 || got[0].Labels.Get("__name__") != "cpu_seconds" { - t.Fatalf(`code="" filter: got %d, want 1 (cpu_seconds)`, len(got)) - } -} - -// TestTimeWindowOverlap verifies inclusive [min_ts,max_ts] vs [mint,maxt] -// overlap filtering and that decoded samples are NOT clipped to the window -// (the read primitive returns the whole matched series; clipping is the -// caller's concern, matching the design's chunk-granularity overlap). -func TestTimeWindowOverlap(t *testing.T) { - series := []Series{ - {Labels: lbls("__name__", "early"), Samples: seriesValues(1000, 1, 2, 3)}, // [1000,3000] - {Labels: lbls("__name__", "mid"), Samples: seriesValues(5000, 4, 5, 6)}, // [5000,7000] - {Labels: lbls("__name__", "late"), Samples: seriesValues(10000, 7, 8, 9)}, // [10000,12000] - } - var buf bytes.Buffer - if err := WritePart(&buf, 0, 20_000, series, Options{}); err != nil { - t.Fatalf("WritePart: %v", err) - } - p, err := OpenPart(buf.Bytes()) - if err != nil { - t.Fatalf("OpenPart: %v", err) - } - - // Window [4000,8000] overlaps only "mid". - got, _ := p.Series(nil, 4000, 8000) - if len(got) != 1 || got[0].Labels.Get("__name__") != "mid" { - t.Fatalf("window [4000,8000]: got %d series, want only mid", len(got)) - } - - // Boundary-touch: window ending exactly at early's max_ts (3000) overlaps. - got, _ = p.Series(nil, 3000, 3000) - if len(got) != 1 || got[0].Labels.Get("__name__") != "early" { - t.Fatalf("window [3000,3000]: got %d series, want only early", len(got)) - } - - // Window entirely before everything. - got, _ = p.Series(nil, 0, 999) - if len(got) != 0 { - t.Fatalf("window [0,999]: got %d series, want 0", len(got)) - } - - // Wide window: all three, samples unclipped. - got, _ = p.Series(nil, math.MinInt64, math.MaxInt64) - if len(got) != 3 { - t.Fatalf("wide window: got %d series, want 3", len(got)) - } - mid := findSeries(t, got, lbls("__name__", "mid")) - assertSamplesEqual(t, "mid unclipped", mid.Samples, seriesValues(5000, 4, 5, 6)) -} - -// TestCRCCorruption mutates a body byte and asserts OpenPart rejects it. -func TestCRCCorruption(t *testing.T) { - series := []Series{{Labels: lbls("__name__", "x"), Samples: seriesValues(0, 1, 2, 3)}} - var buf bytes.Buffer - if err := WritePart(&buf, 0, 5000, series, Options{}); err != nil { - t.Fatalf("WritePart: %v", err) - } - b := buf.Bytes() - - // Flip a byte in the chunks region (well before the footer). - corrupt := append([]byte(nil), b...) - corrupt[magicLen+1+8+8+1] ^= 0xFF // first chunk byte after header - if _, err := OpenPart(corrupt); err != ErrBadCRC { - t.Fatalf("corrupt body: got err %v, want ErrBadCRC", err) - } - - // Flip a byte in the stored crc itself. - corrupt2 := append([]byte(nil), b...) - corrupt2[len(corrupt2)-1] ^= 0x01 - if _, err := OpenPart(corrupt2); err != ErrBadCRC { - t.Fatalf("corrupt crc: got err %v, want ErrBadCRC", err) - } - - // Untouched bytes still open fine. - if _, err := OpenPart(b); err != nil { - t.Fatalf("clean part failed to open: %v", err) - } -} - -// TestBadMagicAndVersion checks header validation. -func TestBadMagicAndVersion(t *testing.T) { - series := []Series{{Labels: lbls("__name__", "x"), Samples: seriesValues(0, 1, 2)}} - var buf bytes.Buffer - if err := WritePart(&buf, 0, 5000, series, Options{}); err != nil { - t.Fatalf("WritePart: %v", err) - } - b := buf.Bytes() - - badMagic := append([]byte(nil), b...) - badMagic[0] = 'X' - if _, err := OpenPart(badMagic); err != ErrBadMagic { - t.Fatalf("bad magic: got %v, want ErrBadMagic", err) - } - - // Bumping the version byte changes the body, so recompute the crc to isolate - // the version check from the crc check. - badVer := append([]byte(nil), b...) - badVer[magicLen] = 99 - // recompute crc over all-but-last-4 - fixCRC(badVer) - if _, err := OpenPart(badVer); err == nil || err == ErrBadCRC { - t.Fatalf("bad version: got %v, want ErrBadVersion", err) - } -} - -// TestEmptyPart writes a part with zero series and reads it back. -func TestEmptyPart(t *testing.T) { - var buf bytes.Buffer - if err := WritePart(&buf, 100, 200, nil, Options{}); err != nil { - t.Fatalf("WritePart(empty): %v", err) - } - p, err := OpenPart(buf.Bytes()) - if err != nil { - t.Fatalf("OpenPart(empty): %v", err) - } - if p.NumSeries() != 0 { - t.Fatalf("NumSeries = %d, want 0", p.NumSeries()) - } - if p.BlockStartMs != 100 || p.BlockEndMs != 200 { - t.Fatalf("block range = [%d,%d], want [100,200]", p.BlockStartMs, p.BlockEndMs) - } - got, err := p.Series(nil, math.MinInt64, math.MaxInt64) - if err != nil { - t.Fatalf("Series(empty): %v", err) - } - if len(got) != 0 { - t.Fatalf("Series(empty) returned %d", len(got)) - } -} - -// TestSingleSampleSeries covers the n==1 chunk edge case. -func TestSingleSampleSeries(t *testing.T) { - series := []Series{ - {Labels: lbls("__name__", "one"), Samples: []Sample{{T: 42, V: 3.5}}}, - } - _, got := roundTrip(t, 0, 100, series) - if len(got) != 1 { - t.Fatalf("got %d series, want 1", len(got)) - } - assertSamplesEqual(t, "single", got[0].Samples, []Sample{{T: 42, V: 3.5}}) -} - -// TestEmptySamplesRejected: a series with no samples is an error at write time. -func TestEmptySamplesRejected(t *testing.T) { - series := []Series{ - {Labels: lbls("__name__", "ok"), Samples: seriesValues(0, 1, 2)}, - {Labels: lbls("__name__", "empty"), Samples: nil}, - } - var buf bytes.Buffer - err := WritePart(&buf, 0, 1000, series, Options{}) - if err == nil { - t.Fatal("WritePart with empty series: want error, got nil") - } -} - -// TestBadTimeRange: block_end < block_start is rejected. -func TestBadTimeRange(t *testing.T) { - series := []Series{{Labels: lbls("__name__", "x"), Samples: seriesValues(0, 1)}} - var buf bytes.Buffer - if err := WritePart(&buf, 5000, 1000, series, Options{}); err != ErrBadTimeRange { - t.Fatalf("bad range: got %v, want ErrBadTimeRange", err) - } -} - -// TestShortBuffer: a truncated buffer is rejected, not panicked. -func TestShortBuffer(t *testing.T) { - if _, err := OpenPart([]byte("AS")); err != ErrShort { - t.Fatalf("tiny buffer: got %v, want ErrShort", err) - } - if _, err := OpenPart(nil); err != ErrShort { - t.Fatalf("nil buffer: got %v, want ErrShort", err) - } -} - -// TestSymbolDedup checks the symbol table dedups shared label strings: a part -// with many series sharing __name__/job must round-trip exactly while the -// symbol table stays far smaller than the naive sum of label strings. -func TestSymbolDedup(t *testing.T) { - const n = 30 - var series []Series - for i := 0; i < n; i++ { - series = append(series, Series{ - // __name__ ("node_cpu") and job ("node-exporter") repeat across all - // series; only the instance value is unique. - Labels: lbls("__name__", "node_cpu", "job", "node-exporter", "instance", "host-"+itoa(i)), - Samples: seriesValues(int64(i)*1000, float64(i), float64(i)+1), - }) - } - var buf bytes.Buffer - if err := WritePart(&buf, 0, 100_000, series, Options{}); err != nil { - t.Fatalf("WritePart: %v", err) - } - p, err := OpenPart(buf.Bytes()) - if err != nil { - t.Fatalf("OpenPart: %v", err) - } - got, _ := p.Series(nil, math.MinInt64, math.MaxInt64) - if len(got) != n { - t.Fatalf("got %d series, want %d", len(got), n) - } - // Distinct symbols = {__name__, node_cpu, job, node-exporter, instance} (5) - // + n unique instance values = n+5. The naive (non-deduped) count would be - // n*6. Assert dedup happened. - wantSyms := n + 5 - if len(p.symbols) != wantSyms { - t.Fatalf("symbol table = %d symbols, want %d (deduped)", len(p.symbols), wantSyms) - } - // Every series still resolves by its unique instance via a matcher. - for i := 0; i < n; i++ { - want := lbls("__name__", "node_cpu", "job", "node-exporter", "instance", "host-"+itoa(i)) - sd := findSeries(t, got, want) - assertSamplesEqual(t, "dedup series "+itoa(i), sd.Samples, seriesValues(int64(i)*1000, float64(i), float64(i)+1)) - } -} - -// itoa is a tiny non-allocating-ish int->string for test label values. -func itoa(i int) string { - if i == 0 { - return "0" - } - var b [20]byte - pos := len(b) - neg := i < 0 - if neg { - i = -i - } - for i > 0 { - pos-- - b[pos] = byte('0' + i%10) - i /= 10 - } - if neg { - pos-- - b[pos] = '-' - } - return string(b[pos:]) -} - -// TestSplitChunksMultiRun directly validates the multi-chunk run split path: -// the part index records per-chunk byte lengths so a series spanning >1 -// intchunk chunk (intchunk's overflow re-base cut) decodes correctly. Because -// best-of-N rarely *picks* a multi-chunk INT encoding over Gorilla, we validate -// the splitter against a hand-built concatenation of two real intchunk chunks. -func TestSplitChunksMultiRun(t *testing.T) { - a := seriesValues(0, 1, 2, 3, 4, 5) - b := seriesValues(6000, 100, 101, 102) - encA, err := intchunk.Encode(a) - if err != nil { - t.Fatalf("encode a: %v", err) - } - encB, err := intchunk.Encode(b) - if err != nil { - t.Fatalf("encode b: %v", err) - } - // Both are single-chunk; concatenate to simulate a 2-chunk run. - var run []byte - var lens []uint64 - for _, c := range append(append([][]byte{}, encA.Chunks...), encB.Chunks...) { - run = append(run, c...) - lens = append(lens, uint64(len(c))) - } - chunks := splitChunks(run, lens) - if len(chunks) != len(lens) { - t.Fatalf("split into %d, want %d", len(chunks), len(lens)) - } - got, err := intchunk.DecodeChunks(chunks) - if err != nil { - t.Fatalf("DecodeChunks: %v", err) - } - want := append(append([]Sample{}, a...), b...) - assertSamplesEqual(t, "multi-run", got, want) -} - -// fixCRC recomputes the trailing crc32c over the part so a test can mutate -// header/index bytes and still pass the crc gate (to isolate other checks). -func fixCRC(b []byte) { - crc := crc32.Checksum(b[:len(b)-4], crc32.MakeTable(crc32.Castagnoli)) - b[len(b)-4] = byte(crc) - b[len(b)-3] = byte(crc >> 8) - b[len(b)-2] = byte(crc >> 16) - b[len(b)-1] = byte(crc >> 24) -} diff --git a/gorilla-merger/internal/merger/coldpartstore.go b/gorilla-merger/internal/merger/coldpartstore.go index 35ecec5df..d01787fec 100644 --- a/gorilla-merger/internal/merger/coldpartstore.go +++ b/gorilla-merger/internal/merger/coldpartstore.go @@ -12,14 +12,13 @@ import ( "strings" "sync" + "github.com/ProjectASAP/asap-gorilla-go/coldpart" kitlog "github.com/go-kit/log" "github.com/go-kit/log/level" "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/prometheus/model/labels" "github.com/thanos-io/objstore" objclient "github.com/thanos-io/objstore/client" - - "github.com/ProjectASAP/asapquery-backend/gorilla-merger/internal/coldpart" ) // BucketCloser is the subset of objstore.Bucket main needs to manage the cold diff --git a/gorilla-merger/internal/merger/coldpartstore_test.go b/gorilla-merger/internal/merger/coldpartstore_test.go index f7856e68b..7da074129 100644 --- a/gorilla-merger/internal/merger/coldpartstore_test.go +++ b/gorilla-merger/internal/merger/coldpartstore_test.go @@ -7,12 +7,11 @@ import ( "net/http/httptest" "testing" + "github.com/ProjectASAP/asap-gorilla-go/coldpart" "github.com/prometheus/prometheus/model/labels" "github.com/prometheus/prometheus/tsdb/chunkenc" "github.com/thanos-io/objstore" "github.com/thanos-io/thanos/pkg/store/storepb" - - "github.com/ProjectASAP/asapquery-backend/gorilla-merger/internal/coldpart" ) // --------------------------------------------------------------------------- diff --git a/gorilla-merger/internal/merger/coldquery.go b/gorilla-merger/internal/merger/coldquery.go index e916d6544..d23c05105 100644 --- a/gorilla-merger/internal/merger/coldquery.go +++ b/gorilla-merger/internal/merger/coldquery.go @@ -4,11 +4,11 @@ import ( "context" "fmt" + "github.com/ProjectASAP/asap-gorilla-go/coldpart" "github.com/prometheus/prometheus/model/labels" "github.com/prometheus/prometheus/tsdb/chunkenc" "github.com/ProjectASAP/asapquery-backend/gorilla-merger/internal/coldchunk" - "github.com/ProjectASAP/asapquery-backend/gorilla-merger/internal/coldpart" ) // ColdChunk is one in-window XOR chunk for a cold series, with its own time