[Querier] Add day range limit for LabelNames and LabelValues (#6233)

Author: afhassan

pkg/querier/querier.go

@@ -431,7 +431,7 @@ func (q querier) Select(ctx context.Context, sortSeries bool, sp *storage.Select
 
 // LabelValues implements storage.Querier.
 func (q querier) LabelValues(ctx context.Context, name string, hints *storage.LabelHints, matchers ...*labels.Matcher) ([]string, annotations.Annotations, error) {
-	ctx, stats, _, _, _, _, queriers, err := q.setupFromCtx(ctx)
+	ctx, stats, userID, mint, maxt, _, queriers, err := q.setupFromCtx(ctx)
 	if err == errEmptyTimeRange {
 		return nil, nil, nil
 	} else if err != nil {
@@ -442,6 +442,14 @@ func (q querier) LabelValues(ctx context.Context, name string, hints *storage.La
 		stats.AddQueryStorageWallTime(time.Since(startT))
 	}()
 
+	startTime := model.Time(mint)
+	endTime := model.Time(maxt)
+
+	if maxQueryLength := q.limits.MaxQueryLength(userID); maxQueryLength > 0 && endTime.Sub(startTime) > maxQueryLength {
+		limitErr := validation.LimitError(fmt.Sprintf(validation.ErrQueryTooLong, endTime.Sub(startTime), maxQueryLength))
+		return nil, nil, limitErr
+	}
+
 	if len(queriers) == 1 {
 		return queriers[0].LabelValues(ctx, name, hints, matchers...)
 	}
@@ -481,7 +489,7 @@ func (q querier) LabelValues(ctx context.Context, name string, hints *storage.La
 }
 
 func (q querier) LabelNames(ctx context.Context, hints *storage.LabelHints, matchers ...*labels.Matcher) ([]string, annotations.Annotations, error) {
-	ctx, stats, _, _, _, _, queriers, err := q.setupFromCtx(ctx)
+	ctx, stats, userID, mint, maxt, _, queriers, err := q.setupFromCtx(ctx)
 	if err == errEmptyTimeRange {
 		return nil, nil, nil
 	} else if err != nil {
@@ -492,6 +500,14 @@ func (q querier) LabelNames(ctx context.Context, hints *storage.LabelHints, matc
 		stats.AddQueryStorageWallTime(time.Since(startT))
 	}()
 
+	startTime := model.Time(mint)
+	endTime := model.Time(maxt)
+
+	if maxQueryLength := q.limits.MaxQueryLength(userID); maxQueryLength > 0 && endTime.Sub(startTime) > maxQueryLength {
+		limitErr := validation.LimitError(fmt.Sprintf(validation.ErrQueryTooLong, endTime.Sub(startTime), maxQueryLength))
+		return nil, nil, limitErr
+	}
+
 	if len(queriers) == 1 {
 		return queriers[0].LabelNames(ctx, hints, matchers...)
 	}

pkg/querier/querier_test.go

@@ -928,6 +928,79 @@ func TestQuerier_ValidateQueryTimeRange_MaxQueryLength_Series(t *testing.T) {
 	require.True(t, strings.Contains(ss.Err().Error(), "the query time range exceeds the limit (query length: 721h0m0s, limit: 720h0m0s)"))
 }
 
+func TestQuerier_ValidateQueryTimeRange_MaxQueryLength_Labels(t *testing.T) {
+	t.Parallel()
+	const maxQueryLength = 30 * 24 * time.Hour
+	tests := map[string]struct {
+		startTime            time.Time
+		endTime              time.Time
+		expected             error
+		ignoreMaxQueryLength bool
+	}{
+		"time range shorter than maxQueryLength": {
+			startTime:            time.Now().Add(-maxQueryLength).Add(time.Hour),
+			endTime:              time.Now(),
+			expected:             nil,
+			ignoreMaxQueryLength: false,
+		},
+		"time range longer than maxQueryLength": {
+			startTime:            time.Now().Add(-maxQueryLength).Add(-time.Hour),
+			endTime:              time.Now(),
+			expected:             validation.LimitError("expanding series: the query time range exceeds the limit (query length: 721h0m0s, limit: 720h0m0s)"),
+			ignoreMaxQueryLength: false,
+		},
+		"time range longer than maxQueryLength and ignoreMaxQueryLength is true": {
+			startTime:            time.Now().Add(-maxQueryLength).Add(-time.Hour),
+			endTime:              time.Now(),
+			expected:             validation.LimitError("expanding series: the query time range exceeds the limit (query length: 721h0m0s, limit: 720h0m0s)"),
+			ignoreMaxQueryLength: true,
+		},
+	}
+
+	for testName, testData := range tests {
+		t.Run(testName, func(t *testing.T) {
+			var cfg Config
+			flagext.DefaultValues(&cfg)
+			cfg.ActiveQueryTrackerDir = ""
+			cfg.IgnoreMaxQueryLength = testData.ignoreMaxQueryLength
+
+			limits := DefaultLimitsConfig()
+			limits.MaxQueryLength = model.Duration(maxQueryLength)
+			overrides, err := validation.NewOverrides(limits, nil)
+			require.NoError(t, err)
+
+			chunkStore := &emptyChunkStore{}
+			distributor := &emptyDistributor{}
+
+			queryables := []QueryableWithFilter{UseAlwaysQueryable(NewMockStoreQueryable(chunkStore))}
+			queryable, _, _ := New(cfg, overrides, distributor, queryables, nil, log.NewNopLogger())
+
+			ctx := user.InjectOrgID(context.Background(), "test")
+
+			q, err := queryable.Querier(util.TimeToMillis(testData.startTime), util.TimeToMillis(testData.endTime))
+			require.NoError(t, err)
+
+			_, _, err = q.LabelNames(ctx, &storage.LabelHints{Limit: 0})
+
+			if testData.expected != nil {
+				require.NotNil(t, err)
+				assert.True(t, strings.Contains(testData.expected.Error(), err.Error()))
+			} else {
+				assert.Nil(t, err)
+			}
+
+			_, _, err = q.LabelValues(ctx, labels.MetricName, &storage.LabelHints{Limit: 0})
+
+			if testData.expected != nil {
+				require.NotNil(t, err)
+				assert.True(t, strings.Contains(testData.expected.Error(), err.Error()))
+			} else {
+				assert.Nil(t, err)
+			}
+		})
+	}
+}
+
 func TestQuerier_ValidateQueryTimeRange_MaxQueryLookback(t *testing.T) {
 	t.Parallel()
 	const (