github-mirrors/VictoriaMetrics
1
0
Fork 0
mirror of https://github.com/VictoriaMetrics/VictoriaMetrics.git synced 2025-03-11 15:34:56 +00:00
Code Issues Projects Releases Packages Wiki Activity

app/vmselect: unconditionally align time range boundaries to step for subqueries as Prometheus does

Browse source
This commit is contained in:
Aliaksandr Valialkin 2020-09-03 13:21:51 +03:00
parent fe98ba5a60
commit e706e59d49
3 changed files with 37 additions and 32 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
app/victoria-metrics/testdata/graphite/subquery-aggregation.json vendored
View file

@ -9,6 +9,6 @@
"query": ["/api/v1/query?query=min%20by%20(item)%20(min_over_time(forms_daily_count[10m:1m]))&time={TIME_S-1m}"],
"result_query": {
"status":"success",
"data":{"resultType":"vector","result":[{"metric":{"item":"x"},"value":["{TIME_S-1m}","1"]},{"metric":{"item":"y"},"value":["{TIME_S-1m}","3"]}]}
"data":{"resultType":"vector","result":[{"metric":{"item":"x"},"value":["{TIME_S-1m}","2"]},{"metric":{"item":"y"},"value":["{TIME_S-1m}","4"]}]}
}
}

25
app/vmselect/promql/eval.go
View file

@ -56,14 +56,7 @@ func AdjustStartEnd(start, end, step int64) (int64, int64) {
// Round start and end to values divisible by step in order
// to enable response caching (see EvalConfig.mayCache).
// Round start to the nearest smaller value divisible by step.
start -= start % step
// Round end to the nearest bigger value divisible by step.
adjust := end % step
if adjust > 0 {
end += step - adjust
}
start, end = alignStartEnd(start, end, step)
// Make sure that the new number of points is the same as the initial number of points.
newPoints := (end-start)/step + 1
@ -75,6 +68,17 @@ func AdjustStartEnd(start, end, step int64) (int64, int64) {
return start, end
}
func alignStartEnd(start, end, step int64) (int64, int64) {
// Round start to the nearest smaller value divisible by step.
start -= start % step
// Round end to the nearest bigger value divisible by step.
adjust := end % step
if adjust > 0 {
end += step - adjust
}
return start, end
}
// EvalConfig is the configuration required for query evaluation via Exec
type EvalConfig struct {
Start int64
@ -501,11 +505,13 @@ func evalRollupFuncWithSubquery(ec *EvalConfig, name string, rf rollupFunc, expr
ecSQ := newEvalConfig(ec)
ecSQ.Start -= window + maxSilenceInterval + step
ecSQ.End += step
ecSQ.Step = step
if err := ValidateMaxPointsPerTimeseries(ecSQ.Start, ecSQ.End, ecSQ.Step); err != nil {
return nil, err
}
ecSQ.Start, ecSQ.End = AdjustStartEnd(ecSQ.Start, ecSQ.End, ecSQ.Step)
// unconditionally align start and end args to step for subquery as Prometheus does.
ecSQ.Start, ecSQ.End = alignStartEnd(ecSQ.Start, ecSQ.End, ecSQ.Step)
tssSQ, err := evalExpr(ecSQ, re.Expr)
if err != nil {
return nil, err
@ -517,7 +523,6 @@ func evalRollupFuncWithSubquery(ec *EvalConfig, name string, rf rollupFunc, expr
}
return nil, nil
}
sharedTimestamps := getTimestamps(ec.Start, ec.End, ec.Step)
preFunc, rcs, err := getRollupConfigs(name, rf, expr, ec.Start, ec.End, ec.Step, window, ec.LookbackDelta, sharedTimestamps)
if err != nil {

42
app/vmselect/promql/exec_test.go
View file

@ -108,7 +108,7 @@ func TestExecSuccess(t *testing.T) {
q := `time() offset 0s`
r := netstorage.Result{
MetricName: metricNameExpected,
Values: []float64{900, 1100, 1300, 1500, 1700, 1900},
Values: []float64{1000, 1200, 1400, 1600, 1800, 2000},
Timestamps: timestampsExpected,
}
resultExpected := []netstorage.Result{r}
@ -119,7 +119,7 @@ func TestExecSuccess(t *testing.T) {
q := `sort((label_set(time(), "foo", "bar"), label_set(time()+10, "foo", "baz")) offset 0s)`
r1 := netstorage.Result{
MetricName: metricNameExpected,
Values: []float64{900, 1100, 1300, 1500, 1700, 1900},
Values: []float64{1000, 1200, 1400, 1600, 1800, 2000},
Timestamps: timestampsExpected,
}
r1.MetricName.Tags = []storage.Tag{{
@ -128,7 +128,7 @@ func TestExecSuccess(t *testing.T) {
}}
r2 := netstorage.Result{
MetricName: metricNameExpected,
Values: []float64{910, 1110, 1310, 1510, 1710, 1910},
Values: []float64{1010, 1210, 1410, 1610, 1810, 2010},
Timestamps: timestampsExpected,
}
r2.MetricName.Tags = []storage.Tag{{
@ -209,7 +209,7 @@ func TestExecSuccess(t *testing.T) {
}}
r2 := netstorage.Result{
MetricName: metricNameExpected,
Values: []float64{860, 1060, 1260, 1460, 1660, 1860},
Values: []float64{810, 1010, 1210, 1410, 1610, 1810},
Timestamps: timestampsExpected,
}
r2.MetricName.Tags = []storage.Tag{{
@ -224,7 +224,7 @@ func TestExecSuccess(t *testing.T) {
q := `sort((label_set(time() offset 100s, "foo", "bar"), label_set(time()+10, "foo", "baz") offset 50s) offset 400s)`
r1 := netstorage.Result{
MetricName: metricNameExpected,
Values: []float64{300, 500, 700, 900, 1100, 1300},
Values: []float64{400, 600, 800, 1000, 1200, 1400},
Timestamps: timestampsExpected,
}
r1.MetricName.Tags = []storage.Tag{{
@ -233,7 +233,7 @@ func TestExecSuccess(t *testing.T) {
}}
r2 := netstorage.Result{
MetricName: metricNameExpected,
Values: []float64{360, 560, 760, 960, 1160, 1360},
Values: []float64{410, 610, 810, 1010, 1210, 1410},
Timestamps: timestampsExpected,
}
r2.MetricName.Tags = []storage.Tag{{
@ -248,21 +248,21 @@ func TestExecSuccess(t *testing.T) {
q := `sort((label_set(time() offset -100s, "foo", "bar"), label_set(time()+10, "foo", "baz") offset -50s) offset -400s)`
r1 := netstorage.Result{
MetricName: metricNameExpected,
Values: []float64{1260, 1460, 1660, 1860, 2060, 2260},
Values: []float64{1400, 1600, 1800, 2000, 2200, 2400},
Timestamps: timestampsExpected,
}
r1.MetricName.Tags = []storage.Tag{{
Key: []byte("foo"),
Value: []byte("baz"),
Value: []byte("bar"),
}}
r2 := netstorage.Result{
MetricName: metricNameExpected,
Values: []float64{1300, 1500, 1700, 1900, 2100, 2300},
Values: []float64{1410, 1610, 1810, 2010, 2210, 2410},
Timestamps: timestampsExpected,
}
r2.MetricName.Tags = []storage.Tag{{
Key: []byte("foo"),
Value: []byte("bar"),
Value: []byte("baz"),
}}
resultExpected := []netstorage.Result{r1, r2}
f(q, resultExpected)
@ -305,7 +305,7 @@ func TestExecSuccess(t *testing.T) {
q := `time()[300s] offset 100s`
r := netstorage.Result{
MetricName: metricNameExpected,
Values: []float64{900, 1100, 1300, 1500, 1700, 1900},
Values: []float64{800, 1000, 1200, 1400, 1600, 1800},
Timestamps: timestampsExpected,
}
resultExpected := []netstorage.Result{r}
@ -338,7 +338,7 @@ func TestExecSuccess(t *testing.T) {
q := `timestamp(123)`
r := netstorage.Result{
MetricName: metricNameExpected,
Values: []float64{900, 1100, 1300, 1500, 1700, 1900},
Values: []float64{1000, 1200, 1400, 1600, 1800, 2000},
Timestamps: timestampsExpected,
}
resultExpected := []netstorage.Result{r}
@ -349,7 +349,7 @@ func TestExecSuccess(t *testing.T) {
q := `timestamp(time())`
r := netstorage.Result{
MetricName: metricNameExpected,
Values: []float64{900, 1100, 1300, 1500, 1700, 1900},
Values: []float64{1000, 1200, 1400, 1600, 1800, 2000},
Timestamps: timestampsExpected,
}
resultExpected := []netstorage.Result{r}
@ -360,7 +360,7 @@ func TestExecSuccess(t *testing.T) {
q := `timestamp(456/time()+123)`
r := netstorage.Result{
MetricName: metricNameExpected,
Values: []float64{900, 1100, 1300, 1500, 1700, 1900},
Values: []float64{1000, 1200, 1400, 1600, 1800, 2000},
Timestamps: timestampsExpected,
}
resultExpected := []netstorage.Result{r}
@ -371,7 +371,7 @@ func TestExecSuccess(t *testing.T) {
q := `timestamp(time()>=1600)`
r := netstorage.Result{
MetricName: metricNameExpected,
Values: []float64{nan, nan, nan, nan, 1700, 1900},
Values: []float64{nan, nan, nan, 1600, 1800, 2000},
Timestamps: timestampsExpected,
}
resultExpected := []netstorage.Result{r}
@ -3630,7 +3630,7 @@ func TestExecSuccess(t *testing.T) {
q := `round(geomean_over_time(alias(time()/100, "foobar")[3i]), 0.1)`
r := netstorage.Result{
MetricName: metricNameExpected,
Values: []float64{6.8, 8.8, 10.9, 12.9, 14.9, 16.9},
Values: []float64{7.8, 9.9, 11.9, 13.9, 15.9, 17.9},
Timestamps: timestampsExpected,
}
resultExpected := []netstorage.Result{r}
@ -3652,7 +3652,7 @@ func TestExecSuccess(t *testing.T) {
q := `sum2_over_time(alias(time()/100, "foobar")[3i])`
r := netstorage.Result{
MetricName: metricNameExpected,
Values: []float64{155, 251, 371, 515, 683, 875},
Values: []float64{200, 308, 440, 596, 776, 980},
Timestamps: timestampsExpected,
}
resultExpected := []netstorage.Result{r}
@ -4713,10 +4713,10 @@ func TestExecSuccess(t *testing.T) {
})
t.Run(`ru(time() offset 1i, 2000)`, func(t *testing.T) {
t.Parallel()
q := `ru(time() offset 1i, 2000)`
q := `ru(time() offset 1.5i, 2000)`
r := netstorage.Result{
MetricName: metricNameExpected,
Values: []float64{65, 55.00000000000001, 45, 35, 25, 15},
Values: []float64{70, 60, 50, 40, 30, 20},
Timestamps: timestampsExpected,
}
resultExpected := []netstorage.Result{r}
@ -4804,7 +4804,7 @@ func TestExecSuccess(t *testing.T) {
q := `integrate(time()*1e-3)`
r := netstorage.Result{
MetricName: metricNameExpected,
Values: []float64{160, 200, 240.00000000000003, 280, 320, 360},
Values: []float64{180, 220.00000000000003, 260, 300, 340.00000000000006, 380},
Timestamps: timestampsExpected,
}
resultExpected := []netstorage.Result{r}
@ -4903,7 +4903,7 @@ func TestExecSuccess(t *testing.T) {
q := `increase(2000-time())`
r := netstorage.Result{
MetricName: metricNameExpected,
Values: []float64{1100, 900, 700, 500, 300, 100},
Values: []float64{1000, 800, 600, 400, 200, 0},
Timestamps: timestampsExpected,
}
resultExpected := []netstorage.Result{r}