Query /series from ingesters regardless the request start / end time range (#3037)

Signed-off-by: Marco Pracucci <[email protected]>

Author: pracucci

CHANGELOG.md

@@ -2,6 +2,7 @@
 
 ## master / unreleased
 
+* [BUGFIX] Querier: query /series from ingesters regardless the `-querier.query-ingesters-within` setting. #3035
 
 ## 1.3.0-rc.1 / 2020-08-10
 

development/tsdb-blocks-storage-s3-single-binary/config/cortex.yaml

@@ -28,6 +28,9 @@ ingester:
           host: consul:8500
       replication_factor: 1
 
+querier:
+  query_ingesters_within: 3h
+
 store_gateway:
   sharding_enabled: true
   sharding_ring:

development/tsdb-blocks-storage-s3/config/cortex.yaml

@@ -29,6 +29,8 @@ ingester:
       replication_factor: 1
 
 querier:
+  query_ingesters_within: 3h
+
   # Used when the blocks sharding is disabled.
   store_gateway_addresses: store-gateway-1:9008,store-gateway-2:9009
 

integration/e2ecortex/client.go

@@ -144,18 +144,24 @@ func (c *Client) QueryRaw(query string) (*http.Response, []byte, error) {
 	return res, body, nil
 }
 
+// Series finds series by label matchers.
+func (c *Client) Series(matches []string, start, end time.Time) ([]model.LabelSet, error) {
+	result, _, err := c.querierClient.Series(context.Background(), matches, start, end)
+	return result, err
+}
+
 // LabelValues gets label values
 func (c *Client) LabelValues(label string) (model.LabelValues, error) {
 	// Cortex currently doesn't support start/end time.
-	value, _, err := c.querierClient.LabelValues(context.Background(), label, time.Time{}, time.Time{})
-	return value, err
+	result, _, err := c.querierClient.LabelValues(context.Background(), label, time.Time{}, time.Time{})
+	return result, err
 }
 
 // LabelNames gets label names
 func (c *Client) LabelNames() ([]string, error) {
 	// Cortex currently doesn't support start/end time.
-	value, _, err := c.querierClient.LabelNames(context.Background(), time.Time{}, time.Time{})
-	return value, err
+	result, _, err := c.querierClient.LabelNames(context.Background(), time.Time{}, time.Time{})
+	return result, err
 }
 
 type addOrgIDRoundTripper struct {

integration/query_frontend_test.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/prometheus/common/model"
+	"github.com/prometheus/prometheus/pkg/labels"
 	"github.com/prometheus/prometheus/prompb"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -124,6 +125,7 @@ func runQueryFrontendTest(t *testing.T, testMissingMetricName bool, setup queryF
 	flags = mergeFlags(flags, map[string]string{
 		"-querier.cache-results":             "true",
 		"-querier.split-queries-by-interval": "24h",
+		"-querier.query-ingesters-within":    "12h", // Required by the test on query /series out of ingesters time range
 		"-frontend.memcached.addresses":      "dns+" + memcached.NetworkEndpoint(e2ecache.MemcachedPort),
 	})
 
@@ -183,6 +185,18 @@ func runQueryFrontendTest(t *testing.T, testMissingMetricName bool, setup queryF
 			require.Contains(t, string(body), "query must contain metric name")
 		}
 
+		// In this test we do ensure that the /series start/end time is ignored and Cortex
+		// always returns series in ingesters memory. No need to repeat it for each user.
+		if userID == 0 {
+			start := now.Add(-1000 * time.Hour)
+			end := now.Add(-999 * time.Hour)
+
+			result, err := c.Series([]string{"series_1"}, start, end)
+			require.NoError(t, err)
+			require.Len(t, result, 1)
+			assert.Equal(t, model.LabelSet{labels.MetricName: "series_1"}, result[0])
+		}
+
 		for q := 0; q < numQueriesPerUser; q++ {
 			go func() {
 				defer wg.Done()
@@ -197,9 +211,9 @@ func runQueryFrontendTest(t *testing.T, testMissingMetricName bool, setup queryF
 
 	wg.Wait()
 
-	extra := float64(0)
+	extra := float64(1)
 	if testMissingMetricName {
-		extra = 1
+		extra++
 	}
 	require.NoError(t, queryFrontend.WaitSumMetrics(e2e.Equals(numUsers*numQueriesPerUser+extra), "cortex_query_frontend_queries_total"))
 

pkg/querier/distributor_queryable.go

@@ -79,11 +79,21 @@ func (q *distributorQuerier) Select(_ bool, sp *storage.SelectHints, matchers ..
 	log, ctx := spanlogger.New(q.ctx, "distributorQuerier.Select")
 	defer log.Span.Finish()
 
-	minT, maxT := q.mint, q.maxt
-	if sp != nil {
-		minT, maxT = sp.Start, sp.End
+	// Kludge: Prometheus passes nil SelectParams if it is doing a 'series' operation,
+	// which needs only metadata. For this specific case we shouldn't apply the queryIngestersWithin
+	// time range manipulation, otherwise we'll end up returning no series at all for
+	// older time ranges (while in Cortex we do ignore the start/end and always return
+	// series in ingesters).
+	if sp == nil {
+		ms, err := q.distributor.MetricsForLabelMatchers(ctx, model.Time(q.mint), model.Time(q.maxt), matchers...)
+		if err != nil {
+			return storage.ErrSeriesSet(err)
+		}
+		return series.MetricsToSeriesSet(ms)
 	}
 
+	minT, maxT := sp.Start, sp.End
+
 	// If queryIngestersWithin is enabled, we do manipulate the query mint to query samples up until
 	// now - queryIngestersWithin, because older time ranges are covered by the storage. This
 	// optimization is particularly important for the blocks storage where the blocks retention in the
@@ -103,16 +113,6 @@ func (q *distributorQuerier) Select(_ bool, sp *storage.SelectHints, matchers ..
 		}
 	}
 
-	// Kludge: Prometheus passes nil SelectParams if it is doing a 'series' operation,
-	// which needs only metadata.
-	if sp == nil {
-		ms, err := q.distributor.MetricsForLabelMatchers(ctx, model.Time(minT), model.Time(maxT), matchers...)
-		if err != nil {
-			return storage.ErrSeriesSet(err)
-		}
-		return series.MetricsToSeriesSet(ms)
-	}
-
 	if q.streaming {
 		return q.streamingSelect(minT, maxT, matchers)
 	}

pkg/querier/distributor_queryable_test.go

@@ -68,6 +68,7 @@ func TestDistributorQuerier_SelectShouldHonorQueryIngestersWithin(t *testing.T)
 	now := time.Now()
 
 	tests := map[string]struct {
+		querySeries          bool
 		queryIngestersWithin time.Duration
 		queryMinT            int64
 		queryMaxT            int64
@@ -102,6 +103,14 @@ func TestDistributorQuerier_SelectShouldHonorQueryIngestersWithin(t *testing.T)
 			expectedMinT:         0,
 			expectedMaxT:         0,
 		},
+		"should not manipulate query time range if queryIngestersWithin is enabled and query max time is older, but the query is for /series": {
+			querySeries:          true,
+			queryIngestersWithin: time.Hour,
+			queryMinT:            util.TimeToMillis(now.Add(-100 * time.Minute)),
+			queryMaxT:            util.TimeToMillis(now.Add(-90 * time.Minute)),
+			expectedMinT:         util.TimeToMillis(now.Add(-100 * time.Minute)),
+			expectedMaxT:         util.TimeToMillis(now.Add(-90 * time.Minute)),
+		},
 	}
 
 	for _, streamingEnabled := range []bool{false, true} {
@@ -110,13 +119,20 @@ func TestDistributorQuerier_SelectShouldHonorQueryIngestersWithin(t *testing.T)
 				distributor := &mockDistributor{}
 				distributor.On("Query", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(model.Matrix{}, nil)
 				distributor.On("QueryStream", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(&client.QueryStreamResponse{}, nil)
+				distributor.On("MetricsForLabelMatchers", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return([]metric.Metric{}, nil)
 
 				ctx := user.InjectOrgID(context.Background(), "test")
 				queryable := newDistributorQueryable(distributor, streamingEnabled, nil, testData.queryIngestersWithin)
 				querier, err := queryable.Querier(ctx, testData.queryMinT, testData.queryMaxT)
 				require.NoError(t, err)
 
-				seriesSet := querier.Select(true, &storage.SelectHints{Start: testData.queryMinT, End: testData.queryMaxT})
+				// Select hints are not passed by Prometheus when querying /series.
+				var hints *storage.SelectHints
+				if !testData.querySeries {
+					hints = &storage.SelectHints{Start: testData.queryMinT, End: testData.queryMaxT}
+				}
+
+				seriesSet := querier.Select(true, hints)
 				require.NoError(t, seriesSet.Err())
 
 				if testData.expectedMinT == 0 && testData.expectedMaxT == 0 {
@@ -216,8 +232,9 @@ func (m *mockDistributor) LabelValuesForLabelName(context.Context, model.LabelNa
 func (m *mockDistributor) LabelNames(context.Context) ([]string, error) {
 	return nil, nil
 }
-func (m *mockDistributor) MetricsForLabelMatchers(ctx context.Context, from, through model.Time, matchers ...*labels.Matcher) ([]metric.Metric, error) {
-	return nil, nil
+func (m *mockDistributor) MetricsForLabelMatchers(ctx context.Context, from, to model.Time, matchers ...*labels.Matcher) ([]metric.Metric, error) {
+	args := m.Called(ctx, from, to, matchers)
+	return args.Get(0).([]metric.Metric), args.Error(1)
 }
 
 func (m *mockDistributor) MetricsMetadata(ctx context.Context) ([]scrape.MetricMetadata, error) {