diff --git a/gorilla-merger/internal/merger/ingest.go b/gorilla-merger/internal/merger/ingest.go index 8999a1d5..bb1eb280 100644 --- a/gorilla-merger/internal/merger/ingest.go +++ b/gorilla-merger/internal/merger/ingest.go @@ -47,15 +47,17 @@ func NewIngester(s *Storage, logger *slog.Logger) *Ingester { // labelsFor builds the series labels for a fragment: __name__ = MetricName, // plus the fragment Attributes, plus the merger's external labels. External // labels win on conflict (they identify this merger/agent and must be stable). -func labelsFor(metricName string, attrs map[string]string, external labels.Labels) labels.Labels { +func labelsFor(metricName string, attrs map[string]string) labels.Labels { bld := labels.NewBuilder(labels.EmptyLabels()) bld.Set(labels.MetricName, metricName) for k, v := range attrs { bld.Set(k, v) } - external.Range(func(l labels.Label) { - bld.Set(l.Name, l.Value) - }) + // External labels (cluster/merger) are NOT stamped into the stored series: + // the Thanos TSDBStore appends them at query time, and the shipper writes + // them into each block's meta. Stamping them here too produced DUPLICATE + // cluster/merger labels, so thanos TSDBStore.Series -> ReAllocZLabelsStrings + // read a malformed label set and crashed (corrupt-length OOM). return bld.Labels() } @@ -92,7 +94,7 @@ func (i *Ingester) IngestBatch(ctx context.Context, raw []byte) (ingestResult, e _ = app.Rollback() return ingestResult{}, fmt.Errorf("fragment %d: decode xor chunk: %w", fi, cerr) } - ls := labelsFor(f.MetricName, f.Attributes, i.externalLabels) + ls := labelsFor(f.MetricName, f.Attributes) it := chunk.Iterator(nil) var ref storage.SeriesRef diff --git a/gorilla-merger/internal/merger/ingest_test.go b/gorilla-merger/internal/merger/ingest_test.go index 3800f757..08e76736 100644 --- a/gorilla-merger/internal/merger/ingest_test.go +++ b/gorilla-merger/internal/merger/ingest_test.go @@ -138,10 +138,11 @@ func TestIngestRoundTrip(t *testing.T) { t.Fatalf("expected 200, got %d", resp.StatusCode) } - // Series A: __name__ + its attrs + external labels. + // Series A: __name__ + its attrs only (external labels are added by the + // TSDBStore at query time, NOT stamped into the stored series). wantA := labels.FromStrings( labels.MetricName, "http_requests_total", - "job", "api", "instance", "a", "merger", "test-merger") + "job", "api", "instance", "a") gotA := readBack(t, storage, wantA) wantSamplesA := []sample{{base, 1}, {base + 1000, 2}, {base + 2000, 3}} assertSamples(t, "A", gotA, wantSamplesA) @@ -149,7 +150,7 @@ func TestIngestRoundTrip(t *testing.T) { // Series B: a distinct instance is a distinct series. wantB := labels.FromStrings( labels.MetricName, "http_requests_total", - "job", "api", "instance", "b", "merger", "test-merger") + "job", "api", "instance", "b") gotB := readBack(t, storage, wantB) wantSamplesB := []sample{{base, 10}, {base + 1000, 20}} assertSamples(t, "B", gotB, wantSamplesB) @@ -178,12 +179,14 @@ func TestIngestBadBodyReturns400(t *testing.T) { } } -func TestLabelsForExternalWins(t *testing.T) { - ext := labels.FromStrings("merger", "m1", "shared", "external") - got := labelsFor("metric", map[string]string{"shared": "frag", "job": "x"}, ext) +func TestLabelsForNoExternalStamping(t *testing.T) { + // External labels (cluster/merger) must NOT be stamped at ingest — the + // Thanos TSDBStore appends them at query time. Stamping them here too + // produced duplicate labels and crashed TSDBStore.Series. + got := labelsFor("metric", map[string]string{"shared": "frag", "job": "x"}) want := labels.FromStrings( labels.MetricName, "metric", - "job", "x", "merger", "m1", "shared", "external") + "job", "x", "shared", "frag") if labels.Compare(got, want) != 0 { t.Fatalf("labelsFor mismatch:\n got %s\n want %s", got.String(), want.String()) } diff --git a/gorilla-merger/internal/merger/storeapi.go b/gorilla-merger/internal/merger/storeapi.go index 075c9e9b..c8f5d92a 100644 --- a/gorilla-merger/internal/merger/storeapi.go +++ b/gorilla-merger/internal/merger/storeapi.go @@ -7,7 +7,10 @@ import ( kitlog "github.com/go-kit/log" "github.com/prometheus/prometheus/model/labels" "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" ) @@ -35,6 +38,25 @@ func NewStoreAPI(s *Storage, extLset labels.Labels, logger kitlog.Logger, addr s grpcSrv := grpc.NewServer() storepb.RegisterStoreServer(grpcSrv, tsdbStore) + // 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.WithStoreInfoFunc(func() (*infopb.StoreInfo, error) { + mint, maxt := tsdbStore.TimeRange() + return &infopb.StoreInfo{ + MinTime: mint, + MaxTime: maxt, + SupportsSharding: true, + SupportsWithoutReplicaLabels: true, + TsdbInfos: tsdbStore.TSDBInfos(), + }, nil + }), + ) + info.RegisterInfoServer(infoSrv)(grpcSrv) + return &StoreAPI{ srv: grpcSrv, tsdbStr: tsdbStore,