store/prometheus: ensure sidecar respects ext labels

pkg/store/prometheus.go

@@ -242,13 +242,13 @@ func (p *PrometheusStore) Series(r *storepb.SeriesRequest, s storepb.Store_Serie
  // remote read.
  contentType := httpResp.Header.Get("Content-Type")
  if strings.HasPrefix(contentType, "application/x-protobuf") {
- return p.handleSampledPrometheusResponse(s, httpResp, queryPrometheusSpan, enableChunkHashCalculation, extLsetToRemove)
+ return p.handleSampledPrometheusResponse(s, httpResp, queryPrometheusSpan, extLset, enableChunkHashCalculation, extLsetToRemove)
  }
 
  if !strings.HasPrefix(contentType, "application/x-streamed-protobuf; proto=prometheus.ChunkedReadResponse") {
  return errors.Errorf("not supported remote read content type: %s", contentType)
  }
- return p.handleStreamedPrometheusResponse(s, shardMatcher, httpResp, queryPrometheusSpan, enableChunkHashCalculation, extLsetToRemove)
+ return p.handleStreamedPrometheusResponse(s, shardMatcher, httpResp, queryPrometheusSpan, extLset, enableChunkHashCalculation, extLsetToRemove)
 }
 
 func (p *PrometheusStore) queryPrometheus(
@@ -326,6 +326,7 @@ func (p *PrometheusStore) handleSampledPrometheusResponse(
  s storepb.Store_SeriesServer,
  httpResp *http.Response,
  querySpan tracing.Span,
+ extLset labels.Labels,
  calculateChecksums bool,
  extLsetToRemove map[string]struct{},
 ) error {
@@ -342,9 +343,10 @@ func (p *PrometheusStore) handleSampledPrometheusResponse(
  span.SetTag("series_count", len(resp.Results[0].Timeseries))
 
  for _, e := range resp.Results[0].Timeseries {
- // Sampled remote read handler already adds external labels for us:
- // https://github.com/prometheus/prometheus/blob/3f6f5d3357e232abe53f1775f893fdf8f842712c/storage/remote/read_handler.go#L166.
- lset := rmLabels(labelpb.ZLabelsToPromLabels(e.Labels), extLsetToRemove)
+ // https://github.com/prometheus/prometheus/blob/3f6f5d3357e232abe53f1775f893fdf8f842712c/storage/remote/read_handler.go#L166
+ // MergeLabels() prefers local labels over external labels but we prefer
+ // external labels hence we need to do this:
+ lset := rmLabels(labelpb.ExtendSortedLabels(labelpb.ZLabelsToPromLabels(e.Labels), extLset), extLsetToRemove)
  if len(e.Samples) == 0 {
  // As found in https://github.com/thanos-io/thanos/issues/381
  // Prometheus can give us completely empty time series. Ignore these with log until we figure out that
@@ -379,6 +381,7 @@ func (p *PrometheusStore) handleStreamedPrometheusResponse(
  shardMatcher *storepb.ShardMatcher,
  httpResp *http.Response,
  querySpan tracing.Span,
+ extLset labels.Labels,
  calculateChecksums bool,
  extLsetToRemove map[string]struct{},
 ) error {
@@ -419,10 +422,10 @@ func (p *PrometheusStore) handleStreamedPrometheusResponse(
 
  framesNum++
  for _, series := range res.ChunkedSeries {
- // Streamed remote read handler already adds
- // external labels:
+ // MergeLabels() prefers local labels over external labels but we prefer
+ // external labels hence we need to do this:
  // https://github.com/prometheus/prometheus/blob/3f6f5d3357e232abe53f1775f893fdf8f842712c/storage/remote/codec.go#L210.
- completeLabelset := rmLabels(labelpb.ZLabelsToPromLabels(series.Labels), extLsetToRemove)
+ completeLabelset := rmLabels(labelpb.ExtendSortedLabels(labelpb.ZLabelsToPromLabels(series.Labels), extLset), extLsetToRemove)
  if !shardMatcher.MatchesLabels(completeLabelset) {
  continue
  }

test/e2e/query_test.go

@@ -2241,3 +2241,53 @@ func TestConnectedQueriesWithLazyProxy(t *testing.T) {
  }, time.Now, promclient.QueryOptions{}, 0)
 
 }
+
+func TestSidecarPrefersExtLabels(t *testing.T) {
+ t.Parallel()
+
+ e, err := e2e.NewDockerEnvironment("sidecar-extlbls")
+ testutil.Ok(t, err)
+ t.Cleanup(e2ethanos.CleanScenario(t, e))
+
+ promCfg := `global:
+ external_labels:
+ region: test`
+
+ prom, sidecar := e2ethanos.NewPrometheusWithSidecar(e, "p1", promCfg, "", e2ethanos.DefaultPrometheusImage(), "", "remote-write-receiver")
+ testutil.Ok(t, e2e.StartAndWaitReady(prom, sidecar))
+
+ endpoints := []string{
+ sidecar.InternalEndpoint("grpc"),
+ }
+ querier := e2ethanos.
+ NewQuerierBuilder(e, "1", endpoints...).
+ Init()
+ testutil.Ok(t, e2e.StartAndWaitReady(querier))
+
+ now := time.Now()
+ ctx := context.Background()
+ m := model.Sample{
+ Metric: map[model.LabelName]model.LabelValue{
+ "__name__": "sidecar_test_metric",
+ "instance": model.LabelValue("test"),
+ "region": model.LabelValue("foo"),
+ },
+ Value: model.SampleValue(2),
+ Timestamp: model.TimeFromUnixNano(now.Add(time.Hour).UnixNano()),
+ }
+ testutil.Ok(t, synthesizeSamples(ctx, prom, []model.Sample{m}))
+
+ retv, _ := instantQuery(t, context.Background(), querier.Endpoint("http"), func() string {
+ return "sidecar_test_metric"
+ }, func() time.Time { return now.Add(time.Hour) }, promclient.QueryOptions{}, 1)
+
+ testutil.Equals(t, model.Vector{&model.Sample{
+ Metric: model.Metric{
+ "__name__": "sidecar_test_metric",
+ "instance": "test",
+ "region": "test",
+ },
+ Value: model.SampleValue(2),
+ Timestamp: model.TimeFromUnixNano(now.Add(time.Hour).UnixNano()),
+ }}, retv)
+}