Skip to content

Commit b9cf452

Browse files
yeya24yeya24
authored
Change max query length check to use sum of individual select time range (#6015)
* change max query length check to use sum of individual select time range Signed-off-by: Ben Ye <[email protected]> * lint Signed-off-by: Ben Ye <[email protected]> * update ruler logic Signed-off-by: Ben Ye <[email protected]> * lint Signed-off-by: Ben Ye <[email protected]> --------- Signed-off-by: Ben Ye <[email protected]>
1 parent 34b87cd commit b9cf452

File tree

7 files changed

+237
-16
lines changed

7 files changed

+237
-16
lines changed

‎pkg/querier/tripperware/instantquery/limits.go

Lines changed: 4 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -5,13 +5,12 @@ import (
55
"net/http"
66
"time"
77

8-
"github.com/prometheus/prometheus/promql"
98
"github.com/prometheus/prometheus/promql/parser"
109
"github.com/weaveworks/common/httpgrpc"
1110

1211
"github.com/cortexproject/cortex/pkg/querier/tripperware"
1312
"github.com/cortexproject/cortex/pkg/tenant"
14-
"github.com/cortexproject/cortex/pkg/util"
13+
"github.com/cortexproject/cortex/pkg/util/promql"
1514
"github.com/cortexproject/cortex/pkg/util/spanlogger"
1615
"github.com/cortexproject/cortex/pkg/util/validation"
1716
)
@@ -52,10 +51,9 @@ func (l limitsMiddleware) Do(ctx context.Context, r tripperware.Request) (trippe
5251
}
5352

5453
// Enforce query length across all selectors in the query.
55-
min, max := promql.FindMinMaxTime(&parser.EvalStmt{Expr: expr, Start: util.TimeFromMillis(0), End: util.TimeFromMillis(0), LookbackDelta: l.lookbackDelta})
56-
diff := util.TimeFromMillis(max).Sub(util.TimeFromMillis(min))
57-
if diff > maxQueryLength {
58-
return nil, httpgrpc.Errorf(http.StatusBadRequest, validation.ErrQueryTooLong, diff, maxQueryLength)
54+
length := promql.FindNonOverlapQueryLength(expr, 0, 0, l.lookbackDelta)
55+
if length > maxQueryLength {
56+
return nil, httpgrpc.Errorf(http.StatusBadRequest, validation.ErrQueryTooLong, length, maxQueryLength)
5957
}
6058
}
6159

‎pkg/querier/tripperware/instantquery/limits_test.go

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,10 @@ func TestLimitsMiddleware_MaxQueryLength(t *testing.T) {
6161
maxQueryLength: thirtyDays,
6262
expectedErr: "the query time range exceeds the limit",
6363
},
64+
"shouldn't exceed time range when having multiple selects with offset": {
65+
query: `rate(up[5m]) + rate(up[5m] offset 40d) + rate(up[5m] offset 80d)`,
66+
maxQueryLength: thirtyDays,
67+
},
6468
}
6569

6670
for testName, testData := range tests {

‎pkg/querier/tripperware/queryrange/limits.go

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -7,13 +7,13 @@ import (
77

88
"github.com/go-kit/log/level"
99
"github.com/prometheus/prometheus/model/timestamp"
10-
"github.com/prometheus/prometheus/promql"
1110
"github.com/prometheus/prometheus/promql/parser"
1211
"github.com/weaveworks/common/httpgrpc"
1312

1413
"github.com/cortexproject/cortex/pkg/querier/tripperware"
1514
"github.com/cortexproject/cortex/pkg/tenant"
1615
"github.com/cortexproject/cortex/pkg/util"
16+
"github.com/cortexproject/cortex/pkg/util/promql"
1717
"github.com/cortexproject/cortex/pkg/util/spanlogger"
1818
"github.com/cortexproject/cortex/pkg/util/validation"
1919
)
@@ -88,10 +88,9 @@ func (l limitsMiddleware) Do(ctx context.Context, r tripperware.Request) (trippe
8888
}
8989

9090
// Enforce query length across all selectors in the query.
91-
min, max := promql.FindMinMaxTime(&parser.EvalStmt{Expr: expr, Start: util.TimeFromMillis(0), End: util.TimeFromMillis(0), LookbackDelta: l.lookbackDelta})
92-
diff := util.TimeFromMillis(max).Sub(util.TimeFromMillis(min))
93-
if diff > maxQueryLength {
94-
return nil, httpgrpc.Errorf(http.StatusBadRequest, validation.ErrQueryTooLong, diff, maxQueryLength)
91+
length := promql.FindNonOverlapQueryLength(expr, 0, 0, l.lookbackDelta)
92+
if length > maxQueryLength {
93+
return nil, httpgrpc.Errorf(http.StatusBadRequest, validation.ErrQueryTooLong, length, maxQueryLength)
9594
}
9695
}
9796

‎pkg/querier/tripperware/queryrange/limits_test.go

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -181,6 +181,12 @@ func TestLimitsMiddleware_MaxQueryLength(t *testing.T) {
181181
reqEndTime: now.Add(-2 * thirtyDays),
182182
expectedErr: "the query time range exceeds the limit",
183183
},
184+
"shouldn't exceed time range when having multiple selects with offset": {
185+
query: `rate(up[5m]) + rate(up[5m] offset 40d) + rate(up[5m] offset 80d)`,
186+
maxQueryLength: thirtyDays,
187+
reqStartTime: now.Add(-time.Hour),
188+
reqEndTime: now,
189+
},
184190
}
185191

186192
for testName, testData := range tests {

‎pkg/ruler/compat.go

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -25,8 +25,8 @@ import (
2525
"github.com/cortexproject/cortex/pkg/cortexpb"
2626
"github.com/cortexproject/cortex/pkg/querier"
2727
"github.com/cortexproject/cortex/pkg/querier/stats"
28-
"github.com/cortexproject/cortex/pkg/util"
2928
util_log "github.com/cortexproject/cortex/pkg/util/log"
29+
promql_util "github.com/cortexproject/cortex/pkg/util/promql"
3030
"github.com/cortexproject/cortex/pkg/util/validation"
3131
)
3232

@@ -194,10 +194,9 @@ func EngineQueryFunc(engine promql.QueryEngine, q storage.Queryable, overrides R
194194
// Fail the query in the engine.
195195
if err == nil {
196196
// Enforce query length across all selectors in the query.
197-
min, max := promql.FindMinMaxTime(&parser.EvalStmt{Expr: expr, Start: util.TimeFromMillis(0), End: util.TimeFromMillis(0), LookbackDelta: lookbackDelta})
198-
diff := util.TimeFromMillis(max).Sub(util.TimeFromMillis(min))
199-
if diff > maxQueryLength {
200-
return nil, validation.LimitError(fmt.Sprintf(validation.ErrQueryTooLong, diff, maxQueryLength))
197+
length := promql_util.FindNonOverlapQueryLength(expr, 0, 0, lookbackDelta)
198+
if length > maxQueryLength {
199+
return nil, validation.LimitError(fmt.Sprintf(validation.ErrQueryTooLong, length, maxQueryLength))
201200
}
202201
}
203202
}

‎pkg/util/promql/promql.go

Lines changed: 128 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,128 @@
1+
package promql
2+
3+
import (
4+
"math"
5+
"sort"
6+
"time"
7+
8+
"github.com/prometheus/prometheus/promql/parser"
9+
10+
"github.com/cortexproject/cortex/pkg/util"
11+
)
12+
13+
// FindNonOverlapQueryLength iterates through all the vector selectors in the statement and finds the time interval
14+
// each selector will try to process. It merges intervals to be non overlapping and calculates the total duration as
15+
// the query length. This takes into account offsets, @ modifiers, and range selectors.
16+
// If the statement does not select series, then duration 0 will be returned.
17+
func FindNonOverlapQueryLength(expr parser.Expr, start, end int64, lookbackDelta time.Duration) time.Duration {
18+
type minMaxTime struct {
19+
minTime, maxTime int64
20+
}
21+
intervals := make([]minMaxTime, 0)
22+
23+
// Whenever a MatrixSelector is evaluated, evalRange is set to the corresponding range.
24+
// The evaluation of the VectorSelector inside then evaluates the given range and unsets
25+
// the variable.
26+
var evalRange time.Duration
27+
parser.Inspect(expr, func(node parser.Node, path []parser.Node) error {
28+
switch n := node.(type) {
29+
case *parser.VectorSelector:
30+
start, end := getTimeRangesForSelector(start, end, durationMilliseconds(lookbackDelta), n, path, evalRange)
31+
intervals = append(intervals, minMaxTime{start, end})
32+
evalRange = 0
33+
case *parser.MatrixSelector:
34+
evalRange = n.Range
35+
}
36+
return nil
37+
})
38+
39+
if len(intervals) == 0 {
40+
return 0
41+
}
42+
43+
sort.Slice(intervals, func(i, j int) bool {
44+
return intervals[i].minTime < intervals[j].minTime
45+
})
46+
47+
prev := intervals[0]
48+
length := time.Duration(0)
49+
for i := 1; i < len(intervals); i++ {
50+
if intervals[i].minTime <= prev.maxTime {
51+
prev.maxTime = max(prev.maxTime, intervals[i].maxTime)
52+
} else {
53+
length += util.TimeFromMillis(prev.maxTime).Sub(util.TimeFromMillis(prev.minTime))
54+
prev = intervals[i]
55+
}
56+
}
57+
length += util.TimeFromMillis(prev.maxTime).Sub(util.TimeFromMillis(prev.minTime))
58+
return length
59+
}
60+
61+
// Copied from https://github.com/prometheus/prometheus/blob/v2.52.0/promql/engine.go#L863.
62+
func getTimeRangesForSelector(start, end, lookbackDelta int64, n *parser.VectorSelector, path []parser.Node, evalRange time.Duration) (int64, int64) {
63+
subqOffset, subqRange, subqTs := subqueryTimes(path)
64+
65+
if subqTs != nil {
66+
// The timestamp on the subquery overrides the eval statement time ranges.
67+
start = *subqTs
68+
end = *subqTs
69+
}
70+
71+
if n.Timestamp != nil {
72+
// The timestamp on the selector overrides everything.
73+
start = *n.Timestamp
74+
end = *n.Timestamp
75+
} else {
76+
offsetMilliseconds := durationMilliseconds(subqOffset)
77+
start = start - offsetMilliseconds - durationMilliseconds(subqRange)
78+
end -= offsetMilliseconds
79+
}
80+
81+
if evalRange == 0 {
82+
start -= lookbackDelta
83+
} else {
84+
// For all matrix queries we want to ensure that we have (end-start) + range selected
85+
// this way we have `range` data before the start time
86+
start -= durationMilliseconds(evalRange)
87+
}
88+
89+
offsetMilliseconds := durationMilliseconds(n.OriginalOffset)
90+
start -= offsetMilliseconds
91+
end -= offsetMilliseconds
92+
93+
return start, end
94+
}
95+
96+
// subqueryTimes returns the sum of offsets and ranges of all subqueries in the path.
97+
// If the @ modifier is used, then the offset and range is w.r.t. that timestamp
98+
// (i.e. the sum is reset when we have @ modifier).
99+
// The returned *int64 is the closest timestamp that was seen. nil for no @ modifier.
100+
// Copied from https://github.com/prometheus/prometheus/blob/v2.52.0/promql/engine.go#L803.
101+
func subqueryTimes(path []parser.Node) (time.Duration, time.Duration, *int64) {
102+
var (
103+
subqOffset, subqRange time.Duration
104+
ts int64 = math.MaxInt64
105+
)
106+
for _, node := range path {
107+
if n, ok := node.(*parser.SubqueryExpr); ok {
108+
subqOffset += n.OriginalOffset
109+
subqRange += n.Range
110+
if n.Timestamp != nil {
111+
// The @ modifier on subquery invalidates all the offset and
112+
// range till now. Hence resetting it here.
113+
subqOffset = n.OriginalOffset
114+
subqRange = n.Range
115+
ts = *n.Timestamp
116+
}
117+
}
118+
}
119+
var tsp *int64
120+
if ts != math.MaxInt64 {
121+
tsp = &ts
122+
}
123+
return subqOffset, subqRange, tsp
124+
}
125+
126+
func durationMilliseconds(d time.Duration) int64 {
127+
return int64(d / (time.Millisecond / time.Nanosecond))
128+
}

‎pkg/util/promql/promql_test.go

Lines changed: 87 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,87 @@
1+
package promql
2+
3+
import (
4+
"testing"
5+
"time"
6+
7+
"github.com/prometheus/prometheus/promql/parser"
8+
"github.com/stretchr/testify/require"
9+
)
10+
11+
func TestFindNonOverlapQueryLength(t *testing.T) {
12+
for _, tc := range []struct {
13+
name string
14+
query string
15+
expectedLength time.Duration
16+
}{
17+
{
18+
name: "number literal, no select",
19+
query: `1`,
20+
},
21+
{
22+
name: "string literal, no select",
23+
query: `"test"`,
24+
},
25+
{
26+
name: "function, no select",
27+
query: `"time()"`,
28+
},
29+
{
30+
name: "single vector selector",
31+
query: `up`,
32+
expectedLength: time.Minute * 5,
33+
},
34+
{
35+
name: "single matrix selector",
36+
query: `up[1h]`,
37+
expectedLength: time.Hour,
38+
},
39+
{
40+
name: "sum rate",
41+
query: `sum(rate(up[1h]))`,
42+
expectedLength: time.Hour,
43+
},
44+
{
45+
name: "single vector selector with offset",
46+
query: `up offset 7d`,
47+
expectedLength: time.Minute * 5,
48+
},
49+
{
50+
name: "single matrix selector with offset",
51+
query: `up[1h] offset 7d`,
52+
expectedLength: time.Hour,
53+
},
54+
{
55+
name: "multiple vector selectors, dedup time range",
56+
query: `sum(up) + sum(up) + sum(up)`,
57+
expectedLength: time.Minute * 5,
58+
},
59+
{
60+
name: "multiple matrix selectors, dedup time range",
61+
query: `sum_over_time(up[1h]) + sum_over_time(up[1h]) + sum_over_time(up[1h])`,
62+
expectedLength: time.Hour,
63+
},
64+
{
65+
name: "multiple vector selectors with offsets",
66+
query: `sum(up) + sum(up offset 1h) + sum(up offset 2h)`,
67+
expectedLength: time.Minute * 15,
68+
},
69+
{
70+
name: "multiple matrix selectors with offsets",
71+
query: `sum_over_time(up[1h]) + sum_over_time(up[1h] offset 1d) + sum_over_time(up[1h] offset 2d)`,
72+
expectedLength: time.Hour * 3,
73+
},
74+
{
75+
name: "multiple sum rate with offsets",
76+
query: `sum(rate(up[5m])) + sum(rate(up[5m] offset 1w)) + sum(rate(up[5m] offset 2w)) + sum(rate(up[5m] offset 3w)) + sum(rate(up[5m] offset 4w))`,
77+
expectedLength: time.Minute * 5 * 5,
78+
},
79+
} {
80+
t.Run(tc.name, func(t *testing.T) {
81+
expr, err := parser.ParseExpr(tc.query)
82+
require.NoError(t, err)
83+
duration := FindNonOverlapQueryLength(expr, 0, 0, time.Minute*5)
84+
require.Equal(t, tc.expectedLength, duration)
85+
})
86+
}
87+
}

0 commit comments

Comments
 (0)