From 871ab18998cd8e8aa8aeb474d7d41f0d52bad1be Mon Sep 17 00:00:00 2001 From: zz_y Date: Sun, 24 May 2026 11:00:45 -0600 Subject: [PATCH] fix(gorilla-merger): custom StoreAPI over tsdb.DB (bypass thanos TSDBStore crash, #39) thanos store.TSDBStore.Series fatally OOMs ("runtime: out of memory", ~8EB) when thanos-query issues a Series RPC against the merger. Under Prometheus's default `stringlabels` build labels.Labels is a packed struct{ data string } (16 bytes), not a []Label slice. TSDBStore.Series builds storepb labels via labelpb.ZLabelsFromPromLabels (an unsafe *(*[]ZLabel)(unsafe.Pointer(&lset)) reinterpret assuming the []Label layout) and then wraps the stream in a resortingServer whose Send calls ReAllocZLabelsStrings(..,false) -> string(noAllocBytes(name)). On the packed-string layout those reads see garbage string lengths and the process dies. thanos v0.41.0 is the latest release and requires Go 1.25, so there is no version/toolchain escape. Replace store.NewTSDBStore with a thin custom storepb.StoreServer (customStore) over the embedded *tsdb.DB: - Series: ChunkQuerier(sorted=true) -> for each series, append external labels once via ExtendSortedLabels (dedup, external wins), build ZLabels by COPYING each Name/Value (zLabelsCopy) instead of the unsafe ZLabelsFromPromLabels, and emit raw XOR AggrChunks (copied bytes). Sends directly, NO resortingServer/flushable wrapper (Querier sorting is sufficient), so ReAllocZLabelsStrings is never invoked. - LabelNames/LabelValues: query the range and merge external label names/values. - Info: unchanged behaviour (external label set + min/max time + TsdbInfos), now sourced from customStore via the safe copying path. Unit tests drive Series in-process with a fake Store_SeriesServer: round-trip (no crash; labels = metric+attrs+external once with no dups; chunks decode back to the ingested samples), SkipChunks, external-label gating, and LabelNames/LabelValues. Full suite + go vet pass. Co-Authored-By: Claude Opus 4.7 (1M context) --- gorilla-merger/internal/merger/customstore.go | 377 +++++++++++++++++ .../internal/merger/customstore_test.go | 382 ++++++++++++++++++ gorilla-merger/internal/merger/storeapi.go | 37 +- 3 files changed, 779 insertions(+), 17 deletions(-) create mode 100644 gorilla-merger/internal/merger/customstore.go create mode 100644 gorilla-merger/internal/merger/customstore_test.go diff --git a/gorilla-merger/internal/merger/customstore.go b/gorilla-merger/internal/merger/customstore.go new file mode 100644 index 00000000..cc46cb63 --- /dev/null +++ b/gorilla-merger/internal/merger/customstore.go @@ -0,0 +1,377 @@ +package merger + +import ( + "context" + "math" + "sort" + + kitlog "github.com/go-kit/log" + "github.com/go-kit/log/level" + "github.com/prometheus/prometheus/model/labels" + "github.com/prometheus/prometheus/storage" + "github.com/thanos-io/thanos/pkg/info/infopb" + "github.com/thanos-io/thanos/pkg/store/labelpb" + "github.com/thanos-io/thanos/pkg/store/storepb" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// customStore is a minimal storepb.StoreServer implemented directly over the +// merger's embedded *tsdb.DB, replacing thanos's store.TSDBStore. +// +// WHY a custom server instead of store.TSDBStore: +// +// Under Prometheus v0.308's default `stringlabels` build, labels.Labels is a +// single packed string (struct{ data string }), NOT a []Label slice. Thanos +// store.TSDBStore.Series builds storepb.Series labels via +// labelpb.ZLabelsFromPromLabels, an UNSAFE *(*[]ZLabel)(unsafe.Pointer(&lset)) +// reinterpret that assumes the []Label layout. On stringlabels it reads the +// packed-string struct as a slice header, producing ZLabels whose string +// headers carry garbage lengths. TSDBStore.Series then wraps the stream in a +// resortingServer whose Send calls labelpb.ReAllocZLabelsStrings(.., false) -> +// string(noAllocBytes(name)), which tries to materialize a ~8EB string and the +// process dies with "runtime: out of memory". (thanos v0.41.0 is the latest +// release and requires Go 1.25, so there is no version/toolchain escape.) +// +// This server avoids BOTH unsafe paths: +// - It never calls ZLabelsFromPromLabels / ReAllocZLabelsStrings / the +// resortingServer/flushable wrappers. +// - It builds every ZLabel by COPYING the Name/Value strings field-by-field +// (see zLabelsCopy), and relies on Querier(sorted=true) for ordering, so no +// post-hoc resort (and thus no ReAllocZLabelsStrings) is needed. +type customStore struct { + db chunkQueryable + extLset labels.Labels + logger kitlog.Logger + storepb.UnimplementedStoreServer +} + +// chunkQueryable is the slice of *tsdb.DB the custom store needs. Narrowed to +// an interface so the Series path can be unit-tested without a full gRPC stack. +type chunkQueryable interface { + ChunkQuerier(mint, maxt int64) (storage.ChunkQuerier, error) + Querier(mint, maxt int64) (storage.Querier, error) + StartTime() (int64, error) +} + +// newCustomStore builds the custom StoreServer over db with the given external +// labels (must be sorted; labels.New/labels.Builder already sorts). +func newCustomStore(db chunkQueryable, extLset labels.Labels, logger kitlog.Logger) *customStore { + if logger == nil { + logger = kitlog.NewNopLogger() + } + return &customStore{db: db, extLset: extLset, logger: logger} +} + +// timeRange mirrors TSDBStore.TimeRange: min = head StartTime (the oldest +// sample currently held), max = +inf so the open window is always queried. +func (s *customStore) timeRange() (int64, int64) { + var minTime int64 = math.MinInt64 + if st, err := s.db.StartTime(); err == nil { + minTime = st + } + return minTime, math.MaxInt64 +} + +// labelSet advertises the merger's external labels to the Info service. Built +// via the SAFE copying path (ZLabelSetsFromPromLabels copies each Name/Value), +// never the unsafe ZLabelsFromPromLabels. +func (s *customStore) labelSet() []labelpb.ZLabelSet { + if s.extLset.IsEmpty() { + return []labelpb.ZLabelSet{} + } + return labelpb.ZLabelSetsFromPromLabels(s.extLset) +} + +// tsdbInfos advertises this store's single TSDB (external labels + time range) +// to the Info service, mirroring TSDBStore.TSDBInfos. +func (s *customStore) tsdbInfos() []infopb.TSDBInfo { + sets := s.labelSet() + if len(sets) == 0 { + return []infopb.TSDBInfo{} + } + mint, maxt := s.timeRange() + return []infopb.TSDBInfo{ + { + Labels: labelpb.ZLabelSet{Labels: sets[0].Labels}, + MinTime: mint, + MaxTime: maxt, + }, + } +} + +// zLabelsCopy converts prom labels to []labelpb.ZLabel by COPYING each +// Name/Value string. This is the SAFE analogue of labelpb.ZLabelsFromPromLabels +// (which is a zero-copy unsafe reinterpret that crashes on stringlabels). Go +// strings are immutable, so assigning lbl.Name/lbl.Value yields independent +// string headers pointing at validly-sized backing arrays. +func zLabelsCopy(lset labels.Labels) []labelpb.ZLabel { + out := make([]labelpb.ZLabel, 0, lset.Len()) + lset.Range(func(l labels.Label) { + out = append(out, labelpb.ZLabel{Name: l.Name, Value: l.Value}) + }) + return out +} + +// completeLabels appends the merger's external labels to a series' own labels, +// external labels winning on conflict, returning a freshly-built (safely +// allocated) sorted labels.Labels. labelpb.ExtendSortedLabels uses a +// labels.Builder internally, so the result is a normal allocation (no aliasing +// of the querier's transient backing memory) and is safe to outlive the +// querier. +func completeLabels(seriesLset, extLset labels.Labels) labels.Labels { + return labelpb.ExtendSortedLabels(seriesLset, extLset) +} + +// promMatchers converts storepb matchers to prom matchers and applies the +// external-label gate: if a matcher targets an external label it must match the +// merger's value (else this store has nothing to contribute and we return +// match=false); matchers that don't touch external labels are passed to Select. +func (s *customStore) promMatchers(ms []storepb.LabelMatcher) (match bool, matchers []*labels.Matcher, err error) { + tms, err := storepb.MatchersToPromMatchers(ms...) + if err != nil { + return false, nil, err + } + if s.extLset.IsEmpty() { + return true, tms, nil + } + var kept []*labels.Matcher + for _, tm := range tms { + extVal := s.extLset.Get(tm.Name) + if extVal == "" { + kept = append(kept, tm) + continue + } + if !tm.Matches(extVal) { + // The external label exists but the matcher excludes our value; + // this store has no matching series. + return false, nil, nil + } + // Matcher targets an external label and matches our value: it is + // satisfied by the appended external label, so drop it from the + // querier matchers (the stored series do not carry external labels). + } + return true, kept, nil +} + +// Series streams matching series (labels + raw XOR chunks) for the requested +// range. Series are emitted already sorted by label set (Querier sorted=true +// plus a consistent external-label append preserves order), so NO resorting +// wrapper is used. +func (s *customStore) Series(r *storepb.SeriesRequest, srv storepb.Store_SeriesServer) error { + match, matchers, err := s.promMatchers(r.Matchers) + if err != nil { + return status.Error(codes.InvalidArgument, err.Error()) + } + if !match { + return nil + } + if len(matchers) == 0 { + return status.Error(codes.InvalidArgument, "no matchers specified (excluding external labels)") + } + + ctx := srv.Context() + + // Drop any external labels the caller asked to strip (replica dedup). + extToRemove := map[string]struct{}{} + for _, l := range r.WithoutReplicaLabels { + extToRemove[l] = struct{}{} + } + finalExt := rmExtLabels(s.extLset, extToRemove) + + q, err := s.db.ChunkQuerier(r.MinTime, r.MaxTime) + if err != nil { + return status.Error(codes.Internal, err.Error()) + } + defer func() { + if cerr := q.Close(); cerr != nil { + level.Warn(s.logger).Log("msg", "close chunk querier", "err", cerr) + } + }() + + hints := &storage.SelectHints{ + Start: r.MinTime, + End: r.MaxTime, + Limit: int(r.Limit), + DisableTrimming: true, + } + set := q.Select(ctx, true /*sorted*/, hints, matchers...) + + for set.Next() { + series := set.At() + + full := completeLabels(series.Labels(), finalExt) + // SAFE label copy — never ZLabelsFromPromLabels. + zls := zLabelsCopy(full) + + if r.SkipChunks { + if err := srv.Send(storepb.NewSeriesResponse(&storepb.Series{Labels: zls})); err != nil { + return status.Error(codes.Aborted, err.Error()) + } + continue + } + + var chks []storepb.AggrChunk + chIt := series.Iterator(nil) + for chIt.Next() { + meta := chIt.At() + if meta.Chunk == nil { + return status.Errorf(codes.Internal, "customStore: unpopulated chunk at ref %v", meta.Ref) + } + src := meta.Chunk.Bytes() + // Copy the chunk bytes: the querier may recycle/mmap-back them. + data := make([]byte, len(src)) + copy(data, src) + chks = append(chks, storepb.AggrChunk{ + MinTime: meta.MinTime, + MaxTime: meta.MaxTime, + Raw: &storepb.Chunk{ + // storepb chunk encoding is one less than the tsdb one + // (tsdb EncXOR=1 -> storepb Chunk_XOR=0). + Type: storepb.Chunk_Encoding(meta.Chunk.Encoding() - 1), + Data: data, + }, + }) + } + if err := chIt.Err(); err != nil { + return status.Error(codes.Internal, err.Error()) + } + + if err := srv.Send(storepb.NewSeriesResponse(&storepb.Series{Labels: zls, Chunks: chks})); err != nil { + return status.Error(codes.Aborted, err.Error()) + } + } + if err := set.Err(); err != nil { + return status.Error(codes.Internal, err.Error()) + } + for _, w := range set.Warnings().AsErrors() { + if err := srv.Send(storepb.NewWarnSeriesResponse(w)); err != nil { + return status.Error(codes.Aborted, err.Error()) + } + } + return nil +} + +// LabelNames returns the union of stored label names (matching the request +// range/matchers) and the merger's external label names. +func (s *customStore) LabelNames(ctx context.Context, r *storepb.LabelNamesRequest) (*storepb.LabelNamesResponse, error) { + match, matchers, err := s.promMatchers(r.Matchers) + if err != nil { + return nil, status.Error(codes.InvalidArgument, err.Error()) + } + if !match { + return &storepb.LabelNamesResponse{}, nil + } + + q, err := s.db.Querier(r.Start, r.End) + if err != nil { + return nil, status.Error(codes.Internal, err.Error()) + } + defer func() { _ = q.Close() }() + + hints := &storage.LabelHints{Limit: int(r.Limit)} + res, _, err := q.LabelNames(ctx, hints, matchers...) + if err != nil { + return nil, status.Error(codes.Internal, err.Error()) + } + + extToRemove := map[string]struct{}{} + for _, l := range r.WithoutReplicaLabels { + extToRemove[l] = struct{}{} + } + seen := map[string]struct{}{} + for _, n := range res { + seen[n] = struct{}{} + } + s.extLset.Range(func(l labels.Label) { + if _, drop := extToRemove[l.Name]; drop { + return + } + if _, dup := seen[l.Name]; dup { + return + } + res = append(res, l.Name) + seen[l.Name] = struct{}{} + }) + sort.Strings(res) + + // Copy out: block label memory may be freed when the querier closes. + out := make([]string, len(res)) + for i := range res { + out[i] = cloneString(res[i]) + } + return &storepb.LabelNamesResponse{Names: out}, nil +} + +// LabelValues returns values for one label name across the request range, +// including the merger's value for an external label. +func (s *customStore) LabelValues(ctx context.Context, r *storepb.LabelValuesRequest) (*storepb.LabelValuesResponse, error) { + if r.Label == "" { + return nil, status.Error(codes.InvalidArgument, "label name parameter cannot be empty") + } + for _, l := range r.WithoutReplicaLabels { + if l == r.Label { + return &storepb.LabelValuesResponse{}, nil + } + } + + match, matchers, err := s.promMatchers(r.Matchers) + if err != nil { + return nil, status.Error(codes.InvalidArgument, err.Error()) + } + if !match { + return &storepb.LabelValuesResponse{}, nil + } + + q, err := s.db.Querier(r.Start, r.End) + if err != nil { + return nil, status.Error(codes.Internal, err.Error()) + } + defer func() { _ = q.Close() }() + + // External label: its only value is the merger's, gated on there being a + // matching series when extra matchers are present. + if extVal := s.extLset.Get(r.Label); extVal != "" { + if len(matchers) == 0 { + return &storepb.LabelValuesResponse{Values: []string{extVal}}, nil + } + hints := &storage.SelectHints{Start: r.Start, End: r.End, Func: "series", Limit: int(r.Limit)} + ss := q.Select(ctx, false, hints, matchers...) + if ss.Next() { + return &storepb.LabelValuesResponse{Values: []string{extVal}}, nil + } + if serr := ss.Err(); serr != nil { + return nil, status.Error(codes.Internal, serr.Error()) + } + return &storepb.LabelValuesResponse{}, nil + } + + hints := &storage.LabelHints{Limit: int(r.Limit)} + res, _, err := q.LabelValues(ctx, r.Label, hints, matchers...) + if err != nil { + return nil, status.Error(codes.Internal, err.Error()) + } + out := make([]string, len(res)) + for i := range res { + out[i] = cloneString(res[i]) + } + return &storepb.LabelValuesResponse{Values: out}, nil +} + +// rmExtLabels returns extLset with any names in remove dropped, freshly built. +func rmExtLabels(extLset labels.Labels, remove map[string]struct{}) labels.Labels { + if len(remove) == 0 { + return extLset + } + b := labels.NewBuilder(extLset) + for n := range remove { + b.Del(n) + } + return b.Labels() +} + +// cloneString returns an independent copy of s (detaches from any larger +// backing array a block querier may have handed us). +func cloneString(s string) string { + return string([]byte(s)) +} diff --git a/gorilla-merger/internal/merger/customstore_test.go b/gorilla-merger/internal/merger/customstore_test.go new file mode 100644 index 00000000..778c4e5a --- /dev/null +++ b/gorilla-merger/internal/merger/customstore_test.go @@ -0,0 +1,382 @@ +package merger + +import ( + "bytes" + "compress/gzip" + "context" + "net/http" + "net/http/httptest" + "sort" + "testing" + "time" + + gorilla "github.com/ProjectASAP/asap-gorilla-go" + "github.com/prometheus/prometheus/model/labels" + "github.com/prometheus/prometheus/tsdb/chunkenc" + "github.com/thanos-io/thanos/pkg/store/labelpb" + "github.com/thanos-io/thanos/pkg/store/storepb" + "google.golang.org/grpc" +) + +// fakeSeriesServer is an in-process storepb.Store_SeriesServer that collects +// every SeriesResponse the custom store emits. Only Send and Context are +// exercised by customStore.Series; the rest of grpc.ServerStream is embedded as +// a nil interface (never called) so we satisfy the interface without a real +// gRPC connection. +type fakeSeriesServer struct { + grpc.ServerStream + ctx context.Context + responses []*storepb.SeriesResponse +} + +func (f *fakeSeriesServer) Send(r *storepb.SeriesResponse) error { + f.responses = append(f.responses, r) + return nil +} + +func (f *fakeSeriesServer) Context() context.Context { + if f.ctx == nil { + return context.Background() + } + return f.ctx +} + +// ingestFragments POSTs the given fragments through the real HTTP ingest +// handler (gzip body), asserting a 200. +func ingestFragments(t *testing.T, s *Storage, frags ...gorilla.Fragment) { + t.Helper() + frame := gorilla.EncodeFragmentBatch(frags) + + ingester := NewIngester(s, nil) + srv := httptest.NewServer(http.HandlerFunc(ingester.HandleIngest)) + t.Cleanup(srv.Close) + + var gzBuf bytes.Buffer + gw := gzip.NewWriter(&gzBuf) + if _, werr := gw.Write(frame); werr != nil { + t.Fatalf("gzip write: %v", werr) + } + if cerr := gw.Close(); cerr != nil { + t.Fatalf("gzip close: %v", cerr) + } + + req, _ := http.NewRequest(http.MethodPost, srv.URL+"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/ingest/gorilla", &gzBuf) + req.Header.Set("Content-Encoding", "gzip") + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatalf("post: %v", err) + } + defer resp.Body.Close() + if resp.StatusCode != http.StatusOK { + t.Fatalf("ingest: expected 200, got %d", resp.StatusCode) + } +} + +// labelsOf converts a response's ZLabels back to a sorted prom labels set by +// COPYING (so the test never depends on the unsafe ZLabelsToPromLabels path). +func labelsOf(t *testing.T, zls []labelpb.ZLabel) labels.Labels { + t.Helper() + b := labels.NewBuilder(labels.EmptyLabels()) + for _, z := range zls { + b.Set(z.Name, z.Value) + } + return b.Labels() +} + +// samplesFromChunks decodes all XOR chunks in a response into time-ordered +// samples, asserting the chunk encoding is XOR. +func samplesFromChunks(t *testing.T, chks []storepb.AggrChunk) []sample { + t.Helper() + var out []sample + for _, c := range chks { + if c.Raw == nil { + t.Fatalf("chunk has no Raw payload") + } + if c.Raw.Type != storepb.Chunk_XOR { + t.Fatalf("chunk type = %v, want XOR", c.Raw.Type) + } + chk, err := chunkenc.FromData(chunkenc.EncXOR, c.Raw.Data) + if err != nil { + t.Fatalf("decode xor chunk: %v", err) + } + it := chk.Iterator(nil) + for it.Next() == chunkenc.ValFloat { + tt, vv := it.At() + out = append(out, sample{t: tt, v: vv}) + } + if it.Err() != nil { + t.Fatalf("chunk iterator: %v", it.Err()) + } + } + return out +} + +// TestCustomStoreSeriesRoundTrip is the key proof for the crash fix: it drives +// the custom Series RPC over real ingested data with a fake stream server and +// asserts (1) NO crash/OOM, (2) labels = metric labels + external labels once +// (no dups), (3) the returned chunks decode back to the ingested samples. +func TestCustomStoreSeriesRoundTrip(t *testing.T) { + dir := t.TempDir() + st, err := OpenStorage(StorageOptions{Dir: dir}) + if err != nil { + t.Fatalf("open storage: %v", err) + } + t.Cleanup(func() { _ = st.Close() }) + + ext := labels.FromStrings("merger", "test-merger", "tier", "cold") + st.SetExternalLabels(ext) + + base := time.Now().UnixMilli() + fragA := makeFragment(t, "http_requests_total", + map[string]string{"job": "api", "instance": "a"}, "agent-1", + []sample{{base, 1}, {base + 1000, 2}, {base + 2000, 3}}) + fragB := makeFragment(t, "http_requests_total", + map[string]string{"job": "api", "instance": "b"}, "agent-2", + []sample{{base, 10}, {base + 1000, 20}}) + ingestFragments(t, st, fragA, fragB) + + cs := newCustomStore(st.DB, ext, nil) + + req := &storepb.SeriesRequest{ + MinTime: base - 60_000, + MaxTime: base + 60_000, + Matchers: []storepb.LabelMatcher{ + {Type: storepb.LabelMatcher_EQ, Name: labels.MetricName, Value: "http_requests_total"}, + }, + } + fss := &fakeSeriesServer{ctx: context.Background()} + + // THE CRASH TEST: with thanos store.TSDBStore this Series call OOMs inside + // ReAllocZLabelsStrings. The custom store must complete cleanly. + if err := cs.Series(req, fss); err != nil { + t.Fatalf("Series returned error: %v", err) + } + + type got struct { + lset labels.Labels + samples []sample + } + var results []got + for _, r := range fss.responses { + series := r.GetSeries() + if series == nil { + // Tolerate warnings/hints but there should be none here. + continue + } + results = append(results, got{ + lset: labelsOf(t, series.Labels), + samples: samplesFromChunks(t, series.Chunks), + }) + } + if len(results) != 2 { + t.Fatalf("expected 2 series, got %d", len(results)) + } + + // Series must be sorted by label set (Querier sorted=true + consistent ext + // append). Verify ordering and exact label sets (metric+attrs+ext, deduped). + wantA := labels.FromStrings( + labels.MetricName, "http_requests_total", + "job", "api", "instance", "a", + "merger", "test-merger", "tier", "cold") + wantB := labels.FromStrings( + labels.MetricName, "http_requests_total", + "job", "api", "instance", "b", + "merger", "test-merger", "tier", "cold") + + if labels.Compare(results[0].lset, wantA) != 0 { + t.Fatalf("series[0] labels:\n got %s\n want %s", results[0].lset.String(), wantA.String()) + } + if labels.Compare(results[1].lset, wantB) != 0 { + t.Fatalf("series[1] labels:\n got %s\n want %s", results[1].lset.String(), wantB.String()) + } + if labels.Compare(results[0].lset, results[1].lset) >= 0 { + t.Fatalf("series not sorted: %s !< %s", results[0].lset.String(), results[1].lset.String()) + } + + // Verify no DUPLICATE label names crept in (each name appears once). + for i, r := range results { + seen := map[string]int{} + r.lset.Range(func(l labels.Label) { seen[l.Name]++ }) + for n, c := range seen { + if c != 1 { + t.Fatalf("series[%d] label %q appears %d times (want 1)", i, n, c) + } + } + } + + assertSamples(t, "A", results[0].samples, []sample{{base, 1}, {base + 1000, 2}, {base + 2000, 3}}) + assertSamples(t, "B", results[1].samples, []sample{{base, 10}, {base + 1000, 20}}) +} + +// TestCustomStoreSeriesSkipChunks verifies the SkipChunks path returns labels +// (with external labels appended once) and NO chunks, without crashing. +func TestCustomStoreSeriesSkipChunks(t *testing.T) { + dir := t.TempDir() + st, err := OpenStorage(StorageOptions{Dir: dir}) + if err != nil { + t.Fatalf("open storage: %v", err) + } + t.Cleanup(func() { _ = st.Close() }) + + ext := labels.FromStrings("merger", "test-merger") + st.SetExternalLabels(ext) + + base := time.Now().UnixMilli() + ingestFragments(t, st, makeFragment(t, "up", + map[string]string{"job": "api"}, "agent-1", + []sample{{base, 1}, {base + 1000, 1}})) + + cs := newCustomStore(st.DB, ext, nil) + req := &storepb.SeriesRequest{ + MinTime: base - 60_000, + MaxTime: base + 60_000, + SkipChunks: true, + Matchers: []storepb.LabelMatcher{ + {Type: storepb.LabelMatcher_EQ, Name: labels.MetricName, Value: "up"}, + }, + } + fss := &fakeSeriesServer{ctx: context.Background()} + if err := cs.Series(req, fss); err != nil { + t.Fatalf("Series(SkipChunks) error: %v", err) + } + + var n int + for _, r := range fss.responses { + series := r.GetSeries() + if series == nil { + continue + } + n++ + if len(series.Chunks) != 0 { + t.Fatalf("SkipChunks: expected 0 chunks, got %d", len(series.Chunks)) + } + want := labels.FromStrings(labels.MetricName, "up", "job", "api", "merger", "test-merger") + if labels.Compare(labelsOf(t, series.Labels), want) != 0 { + t.Fatalf("SkipChunks labels:\n got %s\n want %s", labelsOf(t, series.Labels).String(), want.String()) + } + } + if n != 1 { + t.Fatalf("SkipChunks: expected 1 series, got %d", n) + } +} + +// TestCustomStoreSeriesExternalLabelGate verifies that a matcher on an external +// label whose value DOESN'T match the merger's value yields no series, and one +// that DOES match is satisfied without being passed to the querier. +func TestCustomStoreSeriesExternalLabelGate(t *testing.T) { + dir := t.TempDir() + st, err := OpenStorage(StorageOptions{Dir: dir}) + if err != nil { + t.Fatalf("open storage: %v", err) + } + t.Cleanup(func() { _ = st.Close() }) + + ext := labels.FromStrings("merger", "m1") + st.SetExternalLabels(ext) + + base := time.Now().UnixMilli() + ingestFragments(t, st, makeFragment(t, "up", map[string]string{"job": "api"}, "agent-1", + []sample{{base, 1}})) + + cs := newCustomStore(st.DB, ext, nil) + + // Non-matching external value -> 0 series. + reqMiss := &storepb.SeriesRequest{ + MinTime: base - 60_000, MaxTime: base + 60_000, + Matchers: []storepb.LabelMatcher{ + {Type: storepb.LabelMatcher_EQ, Name: labels.MetricName, Value: "up"}, + {Type: storepb.LabelMatcher_EQ, Name: "merger", Value: "other"}, + }, + } + fssMiss := &fakeSeriesServer{ctx: context.Background()} + if err := cs.Series(reqMiss, fssMiss); err != nil { + t.Fatalf("Series(miss) error: %v", err) + } + if got := countSeries(fssMiss.responses); got != 0 { + t.Fatalf("non-matching external matcher: expected 0 series, got %d", got) + } + + // Matching external value -> 1 series. + reqHit := &storepb.SeriesRequest{ + MinTime: base - 60_000, MaxTime: base + 60_000, + Matchers: []storepb.LabelMatcher{ + {Type: storepb.LabelMatcher_EQ, Name: labels.MetricName, Value: "up"}, + {Type: storepb.LabelMatcher_EQ, Name: "merger", Value: "m1"}, + }, + } + fssHit := &fakeSeriesServer{ctx: context.Background()} + if err := cs.Series(reqHit, fssHit); err != nil { + t.Fatalf("Series(hit) error: %v", err) + } + if got := countSeries(fssHit.responses); got != 1 { + t.Fatalf("matching external matcher: expected 1 series, got %d", got) + } +} + +// TestCustomStoreLabelNamesValues verifies LabelNames/LabelValues merge the +// external labels and return the stored ones. +func TestCustomStoreLabelNamesValues(t *testing.T) { + dir := t.TempDir() + st, err := OpenStorage(StorageOptions{Dir: dir}) + if err != nil { + t.Fatalf("open storage: %v", err) + } + t.Cleanup(func() { _ = st.Close() }) + + ext := labels.FromStrings("merger", "m1") + st.SetExternalLabels(ext) + + base := time.Now().UnixMilli() + ingestFragments(t, st, + makeFragment(t, "up", map[string]string{"job": "api"}, "a", []sample{{base, 1}}), + makeFragment(t, "up", map[string]string{"job": "web"}, "b", []sample{{base, 1}})) + + cs := newCustomStore(st.DB, ext, nil) + ctx := context.Background() + + ln, err := cs.LabelNames(ctx, &storepb.LabelNamesRequest{Start: base - 60_000, End: base + 60_000}) + if err != nil { + t.Fatalf("LabelNames: %v", err) + } + gotNames := append([]string(nil), ln.Names...) + sort.Strings(gotNames) + wantNames := []string{labels.MetricName, "job", "merger"} + if len(gotNames) != len(wantNames) { + t.Fatalf("LabelNames = %v, want %v", gotNames, wantNames) + } + for i := range wantNames { + if gotNames[i] != wantNames[i] { + t.Fatalf("LabelNames = %v, want %v", gotNames, wantNames) + } + } + + // LabelValues for a stored label. + lv, err := cs.LabelValues(ctx, &storepb.LabelValuesRequest{Label: "job", Start: base - 60_000, End: base + 60_000}) + if err != nil { + t.Fatalf("LabelValues(job): %v", err) + } + gotJob := append([]string(nil), lv.Values...) + sort.Strings(gotJob) + if len(gotJob) != 2 || gotJob[0] != "api" || gotJob[1] != "web" { + t.Fatalf("LabelValues(job) = %v, want [api web]", gotJob) + } + + // LabelValues for the external label returns the merger's value. + lvExt, err := cs.LabelValues(ctx, &storepb.LabelValuesRequest{Label: "merger", Start: base - 60_000, End: base + 60_000}) + if err != nil { + t.Fatalf("LabelValues(merger): %v", err) + } + if len(lvExt.Values) != 1 || lvExt.Values[0] != "m1" { + t.Fatalf("LabelValues(merger) = %v, want [m1]", lvExt.Values) + } +} + +func countSeries(responses []*storepb.SeriesResponse) int { + n := 0 + for _, r := range responses { + if r.GetSeries() != nil { + n++ + } + } + return n +} diff --git a/gorilla-merger/internal/merger/storeapi.go b/gorilla-merger/internal/merger/storeapi.go index c8f5d92a..f19aec8f 100644 --- a/gorilla-merger/internal/merger/storeapi.go +++ b/gorilla-merger/internal/merger/storeapi.go @@ -9,7 +9,6 @@ import ( "github.com/thanos-io/thanos/pkg/component" "github.com/thanos-io/thanos/pkg/info" "github.com/thanos-io/thanos/pkg/info/infopb" - "github.com/thanos-io/thanos/pkg/store" "github.com/thanos-io/thanos/pkg/store/labelpb" "github.com/thanos-io/thanos/pkg/store/storepb" "google.golang.org/grpc" @@ -18,49 +17,55 @@ import ( // StoreAPI serves the Thanos StoreAPI (gRPC) over the embedded tsdb.DB. This is // the open-window (<2h pending) query surface; thanos-query fans out to it // alongside the store-gateway (which serves the >=2h S3 blocks), then unions. +// +// It registers a CUSTOM storepb.StoreServer (customStore) rather than thanos's +// store.TSDBStore. store.TSDBStore.Series fatally OOMs under Prometheus's +// default `stringlabels` build because its label encoding (ZLabelsFromPromLabels +// + resortingServer/ReAllocZLabelsStrings) is a zero-copy unsafe reinterpret +// that assumes the []Label layout, not the packed-string one. See customstore.go. type StoreAPI struct { srv *grpc.Server - tsdbStr *store.TSDBStore + store *customStore listener net.Listener addr string } -// NewStoreAPI wraps the tsdb.DB in a Thanos store.TSDBStore (component -// "receive") and prepares a gRPC server bound to addr. +// NewStoreAPI wraps the tsdb.DB in the custom StoreServer (component "receive") +// and prepares a gRPC server bound to addr. func NewStoreAPI(s *Storage, extLset labels.Labels, logger kitlog.Logger, addr string) (*StoreAPI, error) { if logger == nil { logger = kitlog.NewNopLogger() } - // store.TSDBStore requires the external label set to be sorted; labels.New + // The custom store requires the external label set to be sorted; labels.New // / labels.Builder already returns sorted labels. - tsdbStore := store.NewTSDBStore(logger, s.DB, component.Receive, extLset) + cs := newCustomStore(s.DB, extLset, logger) grpcSrv := grpc.NewServer() - storepb.RegisterStoreServer(grpcSrv, tsdbStore) + storepb.RegisterStoreServer(grpcSrv, cs) // thanos-query (v0.41) discovers an endpoint via the Info service; a // Store-only server is reachable but undiscoverable ("neither info nor // store client found"), so register Info too — mirroring the sidecar. infoSrv := info.NewInfoServer( component.Receive.String(), - info.WithLabelSetFunc(func() []labelpb.ZLabelSet { return tsdbStore.LabelSet() }), + info.WithLabelSetFunc(func() []labelpb.ZLabelSet { return cs.labelSet() }), info.WithStoreInfoFunc(func() (*infopb.StoreInfo, error) { - mint, maxt := tsdbStore.TimeRange() + mint, maxt := cs.timeRange() return &infopb.StoreInfo{ MinTime: mint, MaxTime: maxt, SupportsSharding: true, SupportsWithoutReplicaLabels: true, - TsdbInfos: tsdbStore.TSDBInfos(), + TsdbInfos: cs.tsdbInfos(), }, nil }), ) info.RegisterInfoServer(infoSrv)(grpcSrv) return &StoreAPI{ - srv: grpcSrv, - tsdbStr: tsdbStore, - addr: addr, + srv: grpcSrv, + store: cs, + addr: addr, }, nil } @@ -93,10 +98,8 @@ func (a *StoreAPI) Addr() string { return a.addr } -// Stop gracefully stops the gRPC server and closes the TSDBStore. +// Stop gracefully stops the gRPC server. The underlying tsdb.DB is owned by +// Storage and closed by the caller, not here. func (a *StoreAPI) Stop() { a.srv.GracefulStop() - if a.tsdbStr != nil { - a.tsdbStr.Close() - } }