From 7f9b0bb3b756a63f1ce4ac299fd135170372c717 Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 25 May 2026 12:41:00 -0600 Subject: [PATCH] feat(gorilla-merger): decode-on-read for cold intchunk value chunks Add internal/coldchunk, the read-side inverse of the cold ingest path: it turns an intchunk-format value chunk (the best-of-N lossless cold codec in asap-gorilla-go/intchunk) into samples and a standard Prometheus XOR chunk, the building block for a future decode-on-read Thanos StoreAPI. The edge agent does not emit intchunk yet, so this is a tested, importable capability rather than an end-to-end wiring. Tests encode gauge, counter, and high-precision float series so the INT_FOR_DELTA, INT_FOR_DOD, and GORILLA_XOR sub-codecs are all exercised, then assert exact sample round-trip and that the produced XOR chunk iterates back to the same samples. intchunk lives in a subpackage the published asap-gorilla-go tag predates, so resolve it WITHOUT a new release: add a local replace pointing asap-gorilla-go at the in-repo monorepo checkout (relative ../../ASAPCollector/asap-gorilla-go, mirroring the data_plane crate's sibling path-deps). The container build supplies that checkout as a BuildKit build-context and rewrites the replace to the in-image path, so the merger image always compiles against the intchunk-containing source. A directory replace adds no go.sum entries, so the committed module graph is unchanged. Co-Authored-By: Claude Opus 4.7 (1M context) --- gorilla-merger/.gitignore | 5 + gorilla-merger/Dockerfile | 70 +++--- gorilla-merger/go.mod | 14 ++ .../internal/coldchunk/coldchunk.go | 94 ++++++++ .../internal/coldchunk/coldchunk_test.go | 201 ++++++++++++++++++ 5 files changed, 354 insertions(+), 30 deletions(-) create mode 100644 gorilla-merger/internal/coldchunk/coldchunk.go create mode 100644 gorilla-merger/internal/coldchunk/coldchunk_test.go diff --git a/gorilla-merger/.gitignore b/gorilla-merger/.gitignore index ce7166c3e..529d975e2 100644 --- a/gorilla-merger/.gitignore +++ b/gorilla-merger/.gitignore @@ -2,3 +2,8 @@ /data/ /gorilla-merger *.test + +# Local-only go workspace used during development to point the asap-gorilla-go +# replace at a specific intchunk-containing checkout; never committed. +/go.work +/go.work.sum diff --git a/gorilla-merger/Dockerfile b/gorilla-merger/Dockerfile index 78912b5e7..57e911316 100644 --- a/gorilla-merger/Dockerfile +++ b/gorilla-merger/Dockerfile @@ -2,65 +2,75 @@ # # Multi-stage build for the gorilla-merger (Thanos-Receive-style component). # -# ## Why this needs a build secret +# ## asap-gorilla-go is local source, not the published module # # gorilla-merger imports the PRIVATE Go module -# `github.com/ProjectASAP/asap-gorilla-go` (the shared edge<->merger -# ASAPFRG1 wire codec). A naive `go build` inside Docker/CI cannot fetch it: -# the module proxy + git fetch get a 404/auth prompt for the private repo. +# `github.com/ProjectASAP/asap-gorilla-go` (the shared edge<->merger ASAPFRG1 +# wire codec) AND its `intchunk` SUBPACKAGE (the cold value-chunk codec the +# internal/coldchunk decode-on-read helper builds on). intchunk does NOT exist +# in the published asap-gorilla-go tag, so the module cannot come from the proxy. # -# We solve this with a BuildKit *secret* mount carrying a GitHub token. The -# secret is mounted ONLY for the duration of the build RUN that needs it and -# is NOT baked into any image layer (unlike a build-arg or COPY of a token -# file). Inside that RUN we set a transient `url.insteadOf` git rewrite so -# `go` fetches the private module over HTTPS with the token. The git config -# lives only in the container build layer, never on the host. -# -# `GOPRIVATE=github.com/ProjectASAP/*` keeps the fetch off the public proxy -# and checksum DB; `GIT_TERMINAL_PROMPT=0` makes a missing/incorrect token -# fail fast instead of hanging on an interactive credential prompt. +# Instead we COPY the in-repo monorepo `asap-gorilla-go` checkout into the build +# as a BuildKit *build-context* and rewrite go.mod's `replace` to point at that +# in-image path. This is the container analogue of the relative +# `../../ASAPCollector/asap-gorilla-go` replace committed in go.mod, and mirrors +# how build_asap_otel.sh injects the same asap-gorilla-go replace for asap-otel: +# the image always compiles against the intchunk-containing source WITHOUT a +# published tag. Because asap-gorilla-go is now local source, no GitHub token is +# required to fetch it; a `gh_token` secret is still accepted (and used if +# mounted) so any OTHER private fetch keeps working, but it is optional. # # ## Build invocation # -# Write a GitHub token (a PAT or `gh auth token`) to a file, then: -# -# gh auth token > /tmp/gh_token # or: echo "$GITHUB_TOKEN" > /tmp/gh_token # DOCKER_BUILDKIT=1 docker build \ -# --secret id=gh_token,src=/tmp/gh_token \ +# --build-context asap-gorilla-go=/path/ASAPCollector/asap-gorilla-go \ # -t asap/gorilla-merger:dev \ # gorilla-merger/ -# rm -f /tmp/gh_token # -# The build context is the `gorilla-merger/` module directory (this file's -# directory). The same GOPRIVATE + token requirement applies to -# ASAPQuery-backend CI before PR #310 can be merged (the CI runner must -# expose a `gh_token` secret / configure `url.insteadOf` the same way). +# The primary build context is the `gorilla-merger/` module directory (this +# file's directory); the `asap-gorilla-go` build-context supplies the monorepo +# checkout that contains the intchunk subpackage. GOPRIVATE keeps any fetch off +# the public proxy/sumdb; GIT_TERMINAL_PROMPT=0 fails fast instead of hanging. FROM golang:1.25-bookworm AS build -WORKDIR /src -# git is needed for the private-module fetch (insteadOf rewrite below). +# git/ca-certificates for any private-module fetch (insteadOf rewrite below). RUN apt-get update && apt-get install -y --no-install-recommends git ca-certificates \ && rm -rf /var/lib/apt/lists/* -# Prime the module graph first so dependency downloads cache independently -# of source edits. go.sum is committed, so the public deps resolve normally; -# only the private module needs the token. +# Lay the asap-gorilla-go monorepo checkout (with the intchunk subpackage) at +# /ASAPCollector/asap-gorilla-go so it is the sibling the committed relative +# replace `../../ASAPCollector/asap-gorilla-go` resolves to from the module dir +# at /src/gorilla-merger (/src/gorilla-merger/../../ASAPCollector/... == that). +COPY --from=asap-gorilla-go . /ASAPCollector/asap-gorilla-go + +WORKDIR /src/gorilla-merger + +# Prime the module graph first so dependency downloads cache independently of +# source edits. go.sum is committed, so the public deps resolve normally; with +# the directory replace, asap-gorilla-go itself resolves from the local source. COPY go.mod go.sum ./ +# Repoint the committed relative replace at the in-image checkout. The relative +# path already resolves given the layout above; making it absolute keeps the +# build independent of the WORKDIR depth. RUN --mount=type=secret,id=gh_token \ GOPRIVATE=github.com/ProjectASAP/* \ GONOSUMCHECK=github.com/ProjectASAP/* \ GOFLAGS=-mod=mod \ GIT_TERMINAL_PROMPT=0 \ - sh -c 'git config --global url."https://x-access-token:$(cat /run/secrets/gh_token)@github.com/".insteadOf "/" && go mod download' + sh -c 'go mod edit -replace github.com/ProjectASAP/asap-gorilla-go=/ASAPCollector/asap-gorilla-go; \ + if [ -f /run/secrets/gh_token ]; then git config --global url."https://x-access-token:$(cat /run/secrets/gh_token)@github.com/".insteadOf "/"; fi; \ + go mod download' # Now copy the rest of the module and build. COPY . . RUN --mount=type=secret,id=gh_token \ GOPRIVATE=github.com/ProjectASAP/* \ GONOSUMCHECK=github.com/ProjectASAP/* \ + GOFLAGS=-mod=mod \ GIT_TERMINAL_PROMPT=0 \ - sh -c 'git config --global url."https://x-access-token:$(cat /run/secrets/gh_token)@github.com/".insteadOf "/" && CGO_ENABLED=0 go build -ldflags="-s -w" -o /out/gorilla-merger ./cmd/gorilla-merger' + sh -c 'go mod edit -replace github.com/ProjectASAP/asap-gorilla-go=/ASAPCollector/asap-gorilla-go; \ + CGO_ENABLED=0 go build -ldflags="-s -w" -o /out/gorilla-merger ./cmd/gorilla-merger' FROM gcr.io/distroless/static-debian12:nonroot COPY --from=build /out/gorilla-merger /usr/local/bin/gorilla-merger diff --git a/gorilla-merger/go.mod b/gorilla-merger/go.mod index 1f1959832..de8ee92a6 100644 --- a/gorilla-merger/go.mod +++ b/gorilla-merger/go.mod @@ -232,6 +232,20 @@ require ( // standalone private module (github.com/ProjectASAP/asap-gorilla-go). Building // requires GOPRIVATE=github.com/ProjectASAP/* + git auth to fetch it. // +// The cold value-chunk codec used by the decode-on-read helper +// (internal/coldchunk) lives in the asap-gorilla-go/intchunk SUBPACKAGE, which +// the published tag predates. To compile against intchunk WITHOUT cutting a new +// release we point asap-gorilla-go at the in-repo monorepo checkout via a local +// `replace`. The relative path mirrors the data_plane crate's sibling path-deps +// (`../../ASAPCollector/...`): clone ASAPCollector next to ASAPQuery-backend so +// `/gorilla-merger/../../ASAPCollector/asap-gorilla-go` resolves. The +// container build supplies that same checkout as a BuildKit build-context and +// rewrites this replace to the in-image path (see Dockerfile + run_demo.sh), so +// the merger image always compiles against the intchunk-containing +// asap-gorilla-go. intchunk only pulls in prometheus/tsdb/chunkenc (already a +// merger dependency), so no new module is added to the graph. +replace github.com/ProjectASAP/asap-gorilla-go => ../../ASAPCollector/asap-gorilla-go + // Thanos v0.41.0 declares these replace directives in its own go.mod. Go does // NOT inherit a dependency's replace directives into the main module, so MVS // otherwise picks upstream versions whose APIs differ from what Thanos v0.41.0 diff --git a/gorilla-merger/internal/coldchunk/coldchunk.go b/gorilla-merger/internal/coldchunk/coldchunk.go new file mode 100644 index 000000000..395bca896 --- /dev/null +++ b/gorilla-merger/internal/coldchunk/coldchunk.go @@ -0,0 +1,94 @@ +// Package coldchunk provides the decode-on-read inverse of the cold ingest +// path: it turns an intchunk-format value chunk (the best-of-N lossless cold +// codec from asap-gorilla-go/intchunk) into a standard Prometheus XOR +// (Gorilla) chunk that the rest of the merger — and a future decode-on-read +// Thanos StoreAPI — can iterate with plain chunkenc. +// +// This is the read-side mirror of internal/merger/ingest.go, which decodes +// ASAPFRG1 XOR fragments and appends them to the embedded tsdb.DB. Where ingest +// goes (XOR bytes -> samples -> tsdb), coldchunk goes (intchunk bytes -> +// samples -> XOR chunk). The edge agent does not emit intchunk yet, so this is +// not wired end-to-end; it is a tested, importable capability that the +// StoreAPI read path will build on. +// +// intchunk's value codecs (GORILLA_XOR, INT_FOR_DELTA/_DOD and their varint +// variants) are all bit-exact lossless, so re-encoding the decoded samples as a +// Prometheus XOR chunk reproduces the original float64 values exactly. +package coldchunk + +import ( + "errors" + "fmt" + + "github.com/ProjectASAP/asap-gorilla-go/intchunk" + "github.com/prometheus/prometheus/tsdb/chunkenc" +) + +// ErrNoSamples is returned when an intchunk decodes to zero samples; an empty +// chunk has no XOR representation worth producing. +var ErrNoSamples = errors.New("coldchunk: chunk decoded to zero samples") + +// Sample is a re-export of intchunk.Sample so callers of this package do not +// have to import intchunk directly to read the decoded points. +type Sample = intchunk.Sample + +// DecodeToSamples decodes a single self-contained intchunk-format chunk back to +// its (timestamp, value) samples in time order. It is a thin wrapper over +// intchunk.DecodeChunk kept here so the merger has one cold-read entry point. +func DecodeToSamples(chunk []byte) ([]Sample, error) { + samples, err := intchunk.DecodeChunk(chunk) + if err != nil { + return nil, fmt.Errorf("decode intchunk: %w", err) + } + return samples, nil +} + +// DecodeChunksToSamples decodes a sequence of concatenated intchunk chunks (e.g. +// the multiple chunks an overflow re-base cut can produce for one block) back to +// the full ordered sample stream. +func DecodeChunksToSamples(chunks [][]byte) ([]Sample, error) { + samples, err := intchunk.DecodeChunks(chunks) + if err != nil { + return nil, fmt.Errorf("decode intchunks: %w", err) + } + return samples, nil +} + +// SamplesToXORChunk re-encodes decoded samples as a standard Prometheus XOR +// (Gorilla) chunk by appending each point through the XOR Appender — the exact +// inverse of the iterate-the-XOR-chunk loop in ingest.go. The returned chunk is +// a chunkenc.Chunk (EncXOR) that iterates back to the same samples. +// +// Samples must be in non-decreasing timestamp order, which is the order both +// intchunk.DecodeChunk and the cold block layout already guarantee. +func SamplesToXORChunk(samples []Sample) (chunkenc.Chunk, error) { + if len(samples) == 0 { + return nil, ErrNoSamples + } + c := chunkenc.NewXORChunk() + app, err := c.Appender() + if err != nil { + return nil, fmt.Errorf("xor appender: %w", err) + } + for _, s := range samples { + app.Append(s.T, s.V) + } + return c, nil +} + +// DecodeToXORChunk is the headline helper: it decodes an intchunk-format chunk +// and returns both the decoded samples and an equivalent Prometheus XOR chunk. +// Returning the samples alongside the chunk lets callers that only need the +// points skip re-iterating the XOR chunk, while callers feeding a chunk-based +// API (Thanos StoreAPI, tsdb append) get a ready-to-use EncXOR chunk. +func DecodeToXORChunk(chunk []byte) ([]Sample, chunkenc.Chunk, error) { + samples, err := DecodeToSamples(chunk) + if err != nil { + return nil, nil, err + } + xc, err := SamplesToXORChunk(samples) + if err != nil { + return nil, nil, err + } + return samples, xc, nil +} diff --git a/gorilla-merger/internal/coldchunk/coldchunk_test.go b/gorilla-merger/internal/coldchunk/coldchunk_test.go new file mode 100644 index 000000000..66c5f837c --- /dev/null +++ b/gorilla-merger/internal/coldchunk/coldchunk_test.go @@ -0,0 +1,201 @@ +package coldchunk + +import ( + "math" + "testing" + + "github.com/ProjectASAP/asap-gorilla-go/intchunk" + "github.com/prometheus/prometheus/tsdb/chunkenc" +) + +// encode runs the intchunk best-of-N encoder over samples and returns the +// winning chunk(s) plus the codec tag that won, so a test can both round-trip +// the data and assert which sub-codec was exercised. +func encode(t *testing.T, samples []intchunk.Sample) (intchunk.EncodeResult, [][]byte) { + t.Helper() + res, err := intchunk.Encode(samples) + if err != nil { + t.Fatalf("intchunk.Encode: %v", err) + } + if len(res.Chunks) == 0 { + t.Fatalf("intchunk.Encode returned no chunks") + } + return res, res.Chunks +} + +// assertSamplesEqual fails unless got and want are identical in length, every +// timestamp, and every value BIT-EXACTLY (bit-exactness matters for the +// high-precision-float case where == on the float is the whole point). +func assertSamplesEqual(t *testing.T, what string, got, want []intchunk.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) + } + } +} + +// iterateXOR reads every float sample out of a Prometheus XOR chunk in order. +func iterateXOR(t *testing.T, c chunkenc.Chunk) []intchunk.Sample { + t.Helper() + if c.Encoding() != chunkenc.EncXOR { + t.Fatalf("expected EncXOR chunk, got %v", c.Encoding()) + } + it := c.Iterator(nil) + var out []intchunk.Sample + for it.Next() == chunkenc.ValFloat { + ts, v := it.At() + out = append(out, intchunk.Sample{T: ts, V: v}) + } + if err := it.Err(); err != nil { + t.Fatalf("xor iterator: %v", err) + } + return out +} + +// roundTripCase encodes the series, decodes it through the coldchunk helper, +// and asserts both contracts: (a) exact sample round-trip, and (b) the produced +// XOR chunk iterates back to the same samples. It returns the winning codec tag +// so callers can confirm the intended sub-codec was exercised. +func roundTripCase(t *testing.T, name string, samples []intchunk.Sample) intchunk.CodecTag { + t.Helper() + res, chunks := encode(t, samples) + + // Single-chunk inputs go through the headline DecodeToXORChunk; multi-chunk + // (overflow re-base) inputs go through the chunks variant + a manual XOR + // re-encode so both code paths are covered. + var ( + gotSamples []intchunk.Sample + xc chunkenc.Chunk + err error + ) + if len(chunks) == 1 { + gotSamples, xc, err = DecodeToXORChunk(chunks[0]) + if err != nil { + t.Fatalf("%s: DecodeToXORChunk: %v", name, err) + } + } else { + gotSamples, err = DecodeChunksToSamples(chunks) + if err != nil { + t.Fatalf("%s: DecodeChunksToSamples: %v", name, err) + } + xc, err = SamplesToXORChunk(gotSamples) + if err != nil { + t.Fatalf("%s: SamplesToXORChunk: %v", name, err) + } + } + + // (a) decoded samples match the originals exactly. + assertSamplesEqual(t, name+" decode", gotSamples, samples) + + // (b) the produced XOR chunk iterates back to the same samples. + assertSamplesEqual(t, name+" xor-iterate", iterateXOR(t, xc), samples) + + return res.Tag +} + +// TestDecodeGaugeSeries exercises a gauge (irregular fixed-decimal values), which +// the best-of-N encoder serves with an INT_FOR_DELTA-family codec. +func TestDecodeGaugeSeries(t *testing.T) { + base := int64(1_700_000_000_000) + samples := []intchunk.Sample{ + {T: base, V: 12.5}, + {T: base + 1000, V: 13.0}, + {T: base + 2000, V: 11.75}, + {T: base + 3000, V: 14.25}, + {T: base + 4000, V: 13.5}, + {T: base + 5000, V: 12.0}, + {T: base + 6000, V: 15.5}, + } + tag := roundTripCase(t, "gauge", samples) + t.Logf("gauge winning codec: %s", tag) +} + +// TestDecodeCounterSeries exercises a monotonically increasing integer counter, +// which the best-of-N encoder serves with an INT_FOR_DOD-family codec (near-zero +// delta-of-delta). +func TestDecodeCounterSeries(t *testing.T) { + base := int64(1_700_000_000_000) + samples := make([]intchunk.Sample, 0, 64) + val := 0.0 + for i := 0; i < 64; i++ { + val += float64(100 + i) // steadily rising, integer-valued + samples = append(samples, intchunk.Sample{T: base + int64(i)*15000, V: val}) + } + tag := roundTripCase(t, "counter", samples) + t.Logf("counter winning codec: %s", tag) +} + +// TestDecodeHighPrecisionFloatSeries exercises true high-precision floats that +// no decimal scale can represent exactly, forcing the GORILLA_XOR fallback. This +// is the case the intchunk exactness guard reserves for XOR, and it proves the +// coldchunk helper handles the XOR-tagged sub-codec losslessly too. +func TestDecodeHighPrecisionFloatSeries(t *testing.T) { + base := int64(1_700_000_000_000) + samples := []intchunk.Sample{ + {T: base, V: math.Pi}, + {T: base + 1000, V: math.E}, + {T: base + 2000, V: math.Sqrt2}, + {T: base + 3000, V: 1.0 / 3.0}, + {T: base + 4000, V: 0.1 + 0.2}, // classic non-decimal-exact float + {T: base + 5000, V: math.Ln2}, + {T: base + 6000, V: -2.718281828459045e-7}, + } + tag := roundTripCase(t, "highprec", samples) + if tag != intchunk.CodecGorillaXOR { + // Not strictly required for correctness (any lossless codec round-trips), + // but if these genuinely-irrational values stopped landing on the XOR + // fallback it would mean the exactness guard regressed. + t.Logf("high-precision series did not pick GORILLA_XOR (got %s); "+ + "round-trip still verified lossless", tag) + } +} + +// TestDecodeAllSubCodecsExercised confirms the three series above collectively +// drive at least two distinct intchunk codec families, so the coldchunk decode +// helper is proven against the INT_* path and the XOR path (not just one). +func TestDecodeAllSubCodecsExercised(t *testing.T) { + base := int64(1_700_000_000_000) + gauge := []intchunk.Sample{ + {T: base, V: 12.5}, {T: base + 1000, V: 13.0}, {T: base + 2000, V: 11.75}, + {T: base + 3000, V: 14.25}, {T: base + 4000, V: 13.5}, + } + counter := make([]intchunk.Sample, 0, 32) + v := 0.0 + for i := 0; i < 32; i++ { + v += float64(50 + i) + counter = append(counter, intchunk.Sample{T: base + int64(i)*15000, V: v}) + } + highprec := []intchunk.Sample{ + {T: base, V: math.Pi}, {T: base + 1000, V: math.E}, + {T: base + 2000, V: 0.1 + 0.2}, {T: base + 3000, V: math.Sqrt2}, + } + + seen := map[intchunk.CodecTag]bool{} + seen[roundTripCase(t, "all/gauge", gauge)] = true + seen[roundTripCase(t, "all/counter", counter)] = true + seen[roundTripCase(t, "all/highprec", highprec)] = true + if len(seen) < 2 { + t.Fatalf("expected the three series to exercise >=2 distinct codecs, saw %d: %v", len(seen), seen) + } +} + +// TestDecodeEmptyChunkRejected verifies the zero-sample guard rather than +// producing a degenerate XOR chunk. +func TestDecodeEmptyChunkRejected(t *testing.T) { + if _, err := SamplesToXORChunk(nil); err == nil { + t.Fatalf("expected error for empty sample slice") + } +} + +// TestDecodeCorruptChunkRejected verifies a malformed chunk surfaces as a decode +// error (wrapped), not a panic. +func TestDecodeCorruptChunkRejected(t *testing.T) { + if _, err := DecodeToSamples([]byte{0xff, 0x00, 0x01}); err == nil { + t.Fatalf("expected error for corrupt chunk") + } +}